@Test public void testSendCommands() { final GlobalTransactionStepDefinition transactionPipeline = mock(GlobalTransactionStepDefinition.class); final GlobalTransaction globalTransaction = mock(GlobalTransaction.class); final BubbleMessage message = mock(BubbleMessage.class); final String gtx = "gtx"; final String FLTX = "FLTX"; final String PGTX = "PGTX"; final BaseTransactionCommand send = mock(BaseTransactionCommand.class); final BaseTransactionEvent blow = mock(BaseTransactionEvent.class); final BaseTransactionCommand prepare = mock(BaseTransactionCommand.class); final Consumer attach = mock(Consumer.class); when(mockProvision.getGlobalTransaction()).thenReturn(globalTransaction); when(globalTransaction.getId()).thenReturn(gtx); doReturn(message).when(innerGlobalTxStepVisitorUnderTest).buildBubbleMessgae(globalTransaction, FLTX, PGTX); when(transactionPipeline.getSend()).thenReturn(Stream.of(send)); when(transactionPipeline.getPrepare()).thenReturn(Stream.of(prepare)); when(transactionPipeline.getBlow()).thenReturn(Stream.of(blow)); when(transactionPipeline.getAttach()).thenReturn(Optional.of(attach)); doNothing().when(innerGlobalTxStepVisitorUnderTest).receivePayLoad(transactionPipeline); doNothing().when(innerGlobalTxStepVisitorUnderTest).receivePayLoadGTX(transactionPipeline, gtx); when(send.getId()).thenReturn("send"); when(prepare.getId()).thenReturn("prepare"); when(blow.getId()).thenReturn("blow"); innerGlobalTxStepVisitorUnderTest.sendCommands(transactionPipeline, FLTX, PGTX); verify(mockBubbleCache).addBubble("send", message); verify(mockBubbleCache).addBubble("prepare", message); verify(mockBubbleCache).addBubble("blow", message); verify(mockCommandOrchestrator).receive(send); verify(mockCommandOrchestrator).prepare(prepare); verify(mockApplicationEventPublisher).publishEvent(blow); verify(attach).accept(gtx); verify(innerGlobalTxStepVisitorUnderTest).receivePayLoad(transactionPipeline); verify(innerGlobalTxStepVisitorUnderTest).receivePayLoadGTX(transactionPipeline, gtx); }
void sendCommands(GlobalTransactionStepDefinition<? extends BaseGlobalTransactionEvent> transactionPipeline, String FLTX, String PGTX) { BubbleMessage bubbleMessage = this.buildBubbleMessgae(getProvision().getGlobalTransaction(), FLTX, PGTX); String gtx = getProvision().getGlobalTransaction().getId(); this.receivePayLoad(transactionPipeline); this.receivePayLoadGTX(transactionPipeline, gtx); transactionPipeline.getSend().forEach(c -> { getBubbleCache().addBubble(c.getId(), bubbleMessage); getCommandOrchestrator().receive(c); }); transactionPipeline.getPrepare().forEach(c -> { getBubbleCache().addBubble(c.getId(), bubbleMessage); getCommandOrchestrator().prepare(c); }); transactionPipeline.getBlow().forEach(e -> { getBubbleCache().addBubble(e.getId(), bubbleMessage); getApplicationEventPublisher().publishEvent(e); }); transactionPipeline.getAttach().ifPresent(c -> c.accept(gtx)); }
InnerGlobalTxStepVisitor extends StepGlobalTxVisitor { void sendCommands(GlobalTransactionStepDefinition<? extends BaseGlobalTransactionEvent> transactionPipeline, String FLTX, String PGTX) { BubbleMessage bubbleMessage = this.buildBubbleMessgae(getProvision().getGlobalTransaction(), FLTX, PGTX); String gtx = getProvision().getGlobalTransaction().getId(); this.receivePayLoad(transactionPipeline); this.receivePayLoadGTX(transactionPipeline, gtx); transactionPipeline.getSend().forEach(c -> { getBubbleCache().addBubble(c.getId(), bubbleMessage); getCommandOrchestrator().receive(c); }); transactionPipeline.getPrepare().forEach(c -> { getBubbleCache().addBubble(c.getId(), bubbleMessage); getCommandOrchestrator().prepare(c); }); transactionPipeline.getBlow().forEach(e -> { getBubbleCache().addBubble(e.getId(), bubbleMessage); getApplicationEventPublisher().publishEvent(e); }); transactionPipeline.getAttach().ifPresent(c -> c.accept(gtx)); } }
InnerGlobalTxStepVisitor extends StepGlobalTxVisitor { void sendCommands(GlobalTransactionStepDefinition<? extends BaseGlobalTransactionEvent> transactionPipeline, String FLTX, String PGTX) { BubbleMessage bubbleMessage = this.buildBubbleMessgae(getProvision().getGlobalTransaction(), FLTX, PGTX); String gtx = getProvision().getGlobalTransaction().getId(); this.receivePayLoad(transactionPipeline); this.receivePayLoadGTX(transactionPipeline, gtx); transactionPipeline.getSend().forEach(c -> { getBubbleCache().addBubble(c.getId(), bubbleMessage); getCommandOrchestrator().receive(c); }); transactionPipeline.getPrepare().forEach(c -> { getBubbleCache().addBubble(c.getId(), bubbleMessage); getCommandOrchestrator().prepare(c); }); transactionPipeline.getBlow().forEach(e -> { getBubbleCache().addBubble(e.getId(), bubbleMessage); getApplicationEventPublisher().publishEvent(e); }); transactionPipeline.getAttach().ifPresent(c -> c.accept(gtx)); } InnerGlobalTxStepVisitor(BubbleCache bubbleCache, ApplicationEventPublisher applicationEventPublisher,
                                    VisitorProvision provision, CommandOrchestrator commandOrchestrator, TransactionUtilities utilities); }
InnerGlobalTxStepVisitor extends StepGlobalTxVisitor { void sendCommands(GlobalTransactionStepDefinition<? extends BaseGlobalTransactionEvent> transactionPipeline, String FLTX, String PGTX) { BubbleMessage bubbleMessage = this.buildBubbleMessgae(getProvision().getGlobalTransaction(), FLTX, PGTX); String gtx = getProvision().getGlobalTransaction().getId(); this.receivePayLoad(transactionPipeline); this.receivePayLoadGTX(transactionPipeline, gtx); transactionPipeline.getSend().forEach(c -> { getBubbleCache().addBubble(c.getId(), bubbleMessage); getCommandOrchestrator().receive(c); }); transactionPipeline.getPrepare().forEach(c -> { getBubbleCache().addBubble(c.getId(), bubbleMessage); getCommandOrchestrator().prepare(c); }); transactionPipeline.getBlow().forEach(e -> { getBubbleCache().addBubble(e.getId(), bubbleMessage); getApplicationEventPublisher().publishEvent(e); }); transactionPipeline.getAttach().ifPresent(c -> c.accept(gtx)); } InnerGlobalTxStepVisitor(BubbleCache bubbleCache, ApplicationEventPublisher applicationEventPublisher,
                                    VisitorProvision provision, CommandOrchestrator commandOrchestrator, TransactionUtilities utilities); static InnerGlobalTxStepVisitorBuilder builder(); @Override void visit(GlobalTransactionStepDefinition<? extends BaseGlobalTransactionEvent> transactionPipeline); }
InnerGlobalTxStepVisitor extends StepGlobalTxVisitor { void sendCommands(GlobalTransactionStepDefinition<? extends BaseGlobalTransactionEvent> transactionPipeline, String FLTX, String PGTX) { BubbleMessage bubbleMessage = this.buildBubbleMessgae(getProvision().getGlobalTransaction(), FLTX, PGTX); String gtx = getProvision().getGlobalTransaction().getId(); this.receivePayLoad(transactionPipeline); this.receivePayLoadGTX(transactionPipeline, gtx); transactionPipeline.getSend().forEach(c -> { getBubbleCache().addBubble(c.getId(), bubbleMessage); getCommandOrchestrator().receive(c); }); transactionPipeline.getPrepare().forEach(c -> { getBubbleCache().addBubble(c.getId(), bubbleMessage); getCommandOrchestrator().prepare(c); }); transactionPipeline.getBlow().forEach(e -> { getBubbleCache().addBubble(e.getId(), bubbleMessage); getApplicationEventPublisher().publishEvent(e); }); transactionPipeline.getAttach().ifPresent(c -> c.accept(gtx)); } InnerGlobalTxStepVisitor(BubbleCache bubbleCache, ApplicationEventPublisher applicationEventPublisher,
                                    VisitorProvision provision, CommandOrchestrator commandOrchestrator, TransactionUtilities utilities); static InnerGlobalTxStepVisitorBuilder builder(); @Override void visit(GlobalTransactionStepDefinition<? extends BaseGlobalTransactionEvent> transactionPipeline);  }