@Disabled @Test void processUserChangedEvent() { MessageChannel messageChangedChannel = Mockito.mock(MessageChannel.class); String id = UUID.randomUUID().toString(); Flux<Event> events = Flux.just(UserChangedEvent.builder() .aggregateId(id) .payload(UserPayload.builder() .firstName("FirstNameTest") .lastName("LastNameTest") .version(0L) .createdAt(Instant.now()) .id(id) .build()) .type(UserEventType.USER_CREATED) .build()); when(messageBus.events()).thenReturn(events); when(messageBus.documents()).thenReturn(Flux.empty()); userProcessor.onInit(); verify(messageChangedChannel, Mockito.atMost(1)).send(any(Message.class)); }
@PostConstruct void onInit() { logger.debug(() -> "UserProcessor.onInit() ->"); var userChangedStream = messageBus.events() .subscribeOn(Schedulers.parallel()) .filter(event -> event instanceof UserChangedEvent) .cast(UserChangedEvent.class) .map(FN.mapToRecord(USER_CHANGED, objectMapper, UserChangedEvent::aggregateId)); var userDocumentStream = messageBus.documents() .subscribeOn(Schedulers.parallel()) .map(this::mapUserDocumentSenderRecord); kafkaSender.send(userChangedStream) .doOnError(this::processError) .retry() .subscribe(); kafkaSender.send(userDocumentStream) .doOnError(this::processError) .retry() .subscribe(); }
UserProcessor { @PostConstruct void onInit() { logger.debug(() -> "UserProcessor.onInit() ->"); var userChangedStream = messageBus.events() .subscribeOn(Schedulers.parallel()) .filter(event -> event instanceof UserChangedEvent) .cast(UserChangedEvent.class) .map(FN.mapToRecord(USER_CHANGED, objectMapper, UserChangedEvent::aggregateId)); var userDocumentStream = messageBus.documents() .subscribeOn(Schedulers.parallel()) .map(this::mapUserDocumentSenderRecord); kafkaSender.send(userChangedStream) .doOnError(this::processError) .retry() .subscribe(); kafkaSender.send(userDocumentStream) .doOnError(this::processError) .retry() .subscribe(); } }
UserProcessor { @PostConstruct void onInit() { logger.debug(() -> "UserProcessor.onInit() ->"); var userChangedStream = messageBus.events() .subscribeOn(Schedulers.parallel()) .filter(event -> event instanceof UserChangedEvent) .cast(UserChangedEvent.class) .map(FN.mapToRecord(USER_CHANGED, objectMapper, UserChangedEvent::aggregateId)); var userDocumentStream = messageBus.documents() .subscribeOn(Schedulers.parallel()) .map(this::mapUserDocumentSenderRecord); kafkaSender.send(userChangedStream) .doOnError(this::processError) .retry() .subscribe(); kafkaSender.send(userDocumentStream) .doOnError(this::processError) .retry() .subscribe(); } UserProcessor(
      UserApplicationMessageBus messageBus,
      KafkaSender<String, String> kafkaSender,
      ObjectMapper objectMapper
  ); }
UserProcessor { @PostConstruct void onInit() { logger.debug(() -> "UserProcessor.onInit() ->"); var userChangedStream = messageBus.events() .subscribeOn(Schedulers.parallel()) .filter(event -> event instanceof UserChangedEvent) .cast(UserChangedEvent.class) .map(FN.mapToRecord(USER_CHANGED, objectMapper, UserChangedEvent::aggregateId)); var userDocumentStream = messageBus.documents() .subscribeOn(Schedulers.parallel()) .map(this::mapUserDocumentSenderRecord); kafkaSender.send(userChangedStream) .doOnError(this::processError) .retry() .subscribe(); kafkaSender.send(userDocumentStream) .doOnError(this::processError) .retry() .subscribe(); } UserProcessor(
      UserApplicationMessageBus messageBus,
      KafkaSender<String, String> kafkaSender,
      ObjectMapper objectMapper
  );  }
UserProcessor { @PostConstruct void onInit() { logger.debug(() -> "UserProcessor.onInit() ->"); var userChangedStream = messageBus.events() .subscribeOn(Schedulers.parallel()) .filter(event -> event instanceof UserChangedEvent) .cast(UserChangedEvent.class) .map(FN.mapToRecord(USER_CHANGED, objectMapper, UserChangedEvent::aggregateId)); var userDocumentStream = messageBus.documents() .subscribeOn(Schedulers.parallel()) .map(this::mapUserDocumentSenderRecord); kafkaSender.send(userChangedStream) .doOnError(this::processError) .retry() .subscribe(); kafkaSender.send(userDocumentStream) .doOnError(this::processError) .retry() .subscribe(); } UserProcessor(
      UserApplicationMessageBus messageBus,
      KafkaSender<String, String> kafkaSender,
      ObjectMapper objectMapper
  );   }