@Test(expected = DecoderNotFoundException.class) public void testOnCommandStartedWhenDecoderNotRegistred() { final InboundCommandStarted commandStarted = mock(InboundCommandStarted.class); final String subject = "subject"; final CommandMeta meta = mock(CommandMeta.class); when(mockDecoderRegistry.containsKey(subject)).thenReturn(false); when(commandStarted.getMeta()).thenReturn(meta); languageTranslatorImplUnderTest.onCommandStarted(commandStarted); verify(mockBubbleCache, never()).addBubble(anyString(), any(BubbleMessage.class)); verify(mockPublisher, never()).publishEvent(any(BaseTransactionEvent.class)); }
@Async @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) @Transactional(propagation = Propagation.REQUIRES_NEW) public void onCommandStarted(final InboundCommandStarted commandStarted) { final CommandMeta meta = commandStarted.getMeta(); final String subject = meta.getSubject(); if (decoders.containsKey(subject)) { final BaseDecoder decoder = decoders.get(subject); final BaseTransactionEvent event = decoder.decode(meta); addBubbleMessage(meta,Stream.of(event)); this.publisher.publishEvent(event); Optional<InboundCommand> inboundCommandOptional = commandRepository.findInboundommandOpt(meta.getCommandId()); inboundCommandOptional.ifPresent(cmd -> { cmd.changeStatusToCompleted(); txLogger.logTransactionState(txstatus -> log.info( "Transaction {} for InboundCommand: {}", txstatus, txLogger.printJsonObject(cmd))); }); inboundCommandOptional.orElseThrow(CommandNotFoundException::new); }else { throw new DecoderNotFoundException(subject); } }
LanguageTranslatorImpl implements LanguageTranslator { @Async @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) @Transactional(propagation = Propagation.REQUIRES_NEW) public void onCommandStarted(final InboundCommandStarted commandStarted) { final CommandMeta meta = commandStarted.getMeta(); final String subject = meta.getSubject(); if (decoders.containsKey(subject)) { final BaseDecoder decoder = decoders.get(subject); final BaseTransactionEvent event = decoder.decode(meta); addBubbleMessage(meta,Stream.of(event)); this.publisher.publishEvent(event); Optional<InboundCommand> inboundCommandOptional = commandRepository.findInboundommandOpt(meta.getCommandId()); inboundCommandOptional.ifPresent(cmd -> { cmd.changeStatusToCompleted(); txLogger.logTransactionState(txstatus -> log.info( "Transaction {} for InboundCommand: {}", txstatus, txLogger.printJsonObject(cmd))); }); inboundCommandOptional.orElseThrow(CommandNotFoundException::new); }else { throw new DecoderNotFoundException(subject); } } }
LanguageTranslatorImpl implements LanguageTranslator { @Async @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) @Transactional(propagation = Propagation.REQUIRES_NEW) public void onCommandStarted(final InboundCommandStarted commandStarted) { final CommandMeta meta = commandStarted.getMeta(); final String subject = meta.getSubject(); if (decoders.containsKey(subject)) { final BaseDecoder decoder = decoders.get(subject); final BaseTransactionEvent event = decoder.decode(meta); addBubbleMessage(meta,Stream.of(event)); this.publisher.publishEvent(event); Optional<InboundCommand> inboundCommandOptional = commandRepository.findInboundommandOpt(meta.getCommandId()); inboundCommandOptional.ifPresent(cmd -> { cmd.changeStatusToCompleted(); txLogger.logTransactionState(txstatus -> log.info( "Transaction {} for InboundCommand: {}", txstatus, txLogger.printJsonObject(cmd))); }); inboundCommandOptional.orElseThrow(CommandNotFoundException::new); }else { throw new DecoderNotFoundException(subject); } } @Autowired  LanguageTranslatorImpl(CommandRepository commandRepository, ApplicationEventPublisher publisher,
			BubbleCache bubbleCache, DecoderRegistry decoderRegistry, KermossTxLogger txLogger); }
LanguageTranslatorImpl implements LanguageTranslator { @Async @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) @Transactional(propagation = Propagation.REQUIRES_NEW) public void onCommandStarted(final InboundCommandStarted commandStarted) { final CommandMeta meta = commandStarted.getMeta(); final String subject = meta.getSubject(); if (decoders.containsKey(subject)) { final BaseDecoder decoder = decoders.get(subject); final BaseTransactionEvent event = decoder.decode(meta); addBubbleMessage(meta,Stream.of(event)); this.publisher.publishEvent(event); Optional<InboundCommand> inboundCommandOptional = commandRepository.findInboundommandOpt(meta.getCommandId()); inboundCommandOptional.ifPresent(cmd -> { cmd.changeStatusToCompleted(); txLogger.logTransactionState(txstatus -> log.info( "Transaction {} for InboundCommand: {}", txstatus, txLogger.printJsonObject(cmd))); }); inboundCommandOptional.orElseThrow(CommandNotFoundException::new); }else { throw new DecoderNotFoundException(subject); } } @Autowired  LanguageTranslatorImpl(CommandRepository commandRepository, ApplicationEventPublisher publisher,
			BubbleCache bubbleCache, DecoderRegistry decoderRegistry, KermossTxLogger txLogger); void registerDecoder(String subject, BaseDecoder baseDecoder); @Async @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) @Transactional(propagation = Propagation.REQUIRES_NEW) void onCommandStarted(final InboundCommandStarted commandStarted); @Override @EventListener void onCommandPrepared(InboundCommandPrepared commandPrepared); void addBubbleMessage(CommandMeta meta, Stream<BaseTransactionEvent> events); BubbleCache getBubbleCache(); CommandRepository getCommandRepository(); ApplicationEventPublisher getPublisher(); Map<String, BaseDecoder> getDecoders(); void setDecoders(DecoderRegistry decoders); }
LanguageTranslatorImpl implements LanguageTranslator { @Async @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) @Transactional(propagation = Propagation.REQUIRES_NEW) public void onCommandStarted(final InboundCommandStarted commandStarted) { final CommandMeta meta = commandStarted.getMeta(); final String subject = meta.getSubject(); if (decoders.containsKey(subject)) { final BaseDecoder decoder = decoders.get(subject); final BaseTransactionEvent event = decoder.decode(meta); addBubbleMessage(meta,Stream.of(event)); this.publisher.publishEvent(event); Optional<InboundCommand> inboundCommandOptional = commandRepository.findInboundommandOpt(meta.getCommandId()); inboundCommandOptional.ifPresent(cmd -> { cmd.changeStatusToCompleted(); txLogger.logTransactionState(txstatus -> log.info( "Transaction {} for InboundCommand: {}", txstatus, txLogger.printJsonObject(cmd))); }); inboundCommandOptional.orElseThrow(CommandNotFoundException::new); }else { throw new DecoderNotFoundException(subject); } } @Autowired  LanguageTranslatorImpl(CommandRepository commandRepository, ApplicationEventPublisher publisher,
			BubbleCache bubbleCache, DecoderRegistry decoderRegistry, KermossTxLogger txLogger); void registerDecoder(String subject, BaseDecoder baseDecoder); @Async @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) @Transactional(propagation = Propagation.REQUIRES_NEW) void onCommandStarted(final InboundCommandStarted commandStarted); @Override @EventListener void onCommandPrepared(InboundCommandPrepared commandPrepared); void addBubbleMessage(CommandMeta meta, Stream<BaseTransactionEvent> events); BubbleCache getBubbleCache(); CommandRepository getCommandRepository(); ApplicationEventPublisher getPublisher(); Map<String, BaseDecoder> getDecoders(); void setDecoders(DecoderRegistry decoders);  }