@Test public void onMergeSegmentRequestExecuted() { MergeSegmentRequest mergeSegmentRequest = new MergeSegmentRequest(false, context, "MergeClient", "Processor", 1); testSubject.on(mergeSegmentRequest); assertFalse(publishedInstructions.isEmpty()); PlatformOutboundInstruction published = publishedInstructions.get("MergeClient").get(0); EventProcessorSegmentReference expected = EventProcessorSegmentReference.newBuilder() .setSegmentIdentifier(1) .setProcessorName("Processor") .build(); assertEquals(expected, published.getMergeEventProcessorSegment()); notifySuccessForInstruction(published.getInstructionId()); assertFalse(publishedInternalEvents.isEmpty()); Object event = publishedInternalEvents.get(0); assertTrue(event instanceof MergeSegmentsSucceeded); assertEquals("Processor", ((MergeSegmentsSucceeded) event).processorName()); assertEquals("MergeClient", ((MergeSegmentsSucceeded) event).clientId()); }
@EventListener public void on(EventProcessorEvents.SplitSegmentRequest event) { EventProcessorSegmentReference splitSegmentRequest = EventProcessorSegmentReference.newBuilder() .setProcessorName(event.getProcessorName()) .setSegmentIdentifier(event.getSegmentId()) .build(); PlatformOutboundInstruction instruction = PlatformOutboundInstruction.newBuilder() .setSplitEventProcessorSegment(splitSegmentRequest) .setInstructionId(UUID.randomUUID().toString()) .build(); SplitSegmentsSucceeded success = new SplitSegmentsSucceeded(event.context(), event.getClientId(), event.getProcessorName()); instructionResultSource .onInstructionResultFor(instruction.getInstructionId()) .subscribe(() -> eventPublisher.publishEvent(success), error -> LOGGER.warn("Error during segment split: {}, {}", error, instruction), timeout -> LOGGER.warn("The following operation is taking to long: {}", instruction)); instructionPublisher.publish(event.context(), event.getClientId(), instruction); }
EventProcessorService { @EventListener public void on(EventProcessorEvents.SplitSegmentRequest event) { EventProcessorSegmentReference splitSegmentRequest = EventProcessorSegmentReference.newBuilder() .setProcessorName(event.getProcessorName()) .setSegmentIdentifier(event.getSegmentId()) .build(); PlatformOutboundInstruction instruction = PlatformOutboundInstruction.newBuilder() .setSplitEventProcessorSegment(splitSegmentRequest) .setInstructionId(UUID.randomUUID().toString()) .build(); SplitSegmentsSucceeded success = new SplitSegmentsSucceeded(event.context(), event.getClientId(), event.getProcessorName()); instructionResultSource .onInstructionResultFor(instruction.getInstructionId()) .subscribe(() -> eventPublisher.publishEvent(success), error -> LOGGER.warn("Error during segment split: {}, {}", error, instruction), timeout -> LOGGER.warn("The following operation is taking to long: {}", instruction)); instructionPublisher.publish(event.context(), event.getClientId(), instruction); } }
EventProcessorService { @EventListener public void on(EventProcessorEvents.SplitSegmentRequest event) { EventProcessorSegmentReference splitSegmentRequest = EventProcessorSegmentReference.newBuilder() .setProcessorName(event.getProcessorName()) .setSegmentIdentifier(event.getSegmentId()) .build(); PlatformOutboundInstruction instruction = PlatformOutboundInstruction.newBuilder() .setSplitEventProcessorSegment(splitSegmentRequest) .setInstructionId(UUID.randomUUID().toString()) .build(); SplitSegmentsSucceeded success = new SplitSegmentsSucceeded(event.context(), event.getClientId(), event.getProcessorName()); instructionResultSource .onInstructionResultFor(instruction.getInstructionId()) .subscribe(() -> eventPublisher.publishEvent(success), error -> LOGGER.warn("Error during segment split: {}, {}", error, instruction), timeout -> LOGGER.warn("The following operation is taking to long: {}", instruction)); instructionPublisher.publish(event.context(), event.getClientId(), instruction); } @Autowired  EventProcessorService(PlatformService instructionPublisher,
                                 InstructionResultSource.Factory instructionResultSource,
                                 AxonServerEventPublisher eventPublisher);  EventProcessorService(
            InstructionPublisher instructionPublisher,
            InstructionResultSource.Factory instructionResultSource,
            AxonServerEventPublisher eventPublisher); }
EventProcessorService { @EventListener public void on(EventProcessorEvents.SplitSegmentRequest event) { EventProcessorSegmentReference splitSegmentRequest = EventProcessorSegmentReference.newBuilder() .setProcessorName(event.getProcessorName()) .setSegmentIdentifier(event.getSegmentId()) .build(); PlatformOutboundInstruction instruction = PlatformOutboundInstruction.newBuilder() .setSplitEventProcessorSegment(splitSegmentRequest) .setInstructionId(UUID.randomUUID().toString()) .build(); SplitSegmentsSucceeded success = new SplitSegmentsSucceeded(event.context(), event.getClientId(), event.getProcessorName()); instructionResultSource .onInstructionResultFor(instruction.getInstructionId()) .subscribe(() -> eventPublisher.publishEvent(success), error -> LOGGER.warn("Error during segment split: {}, {}", error, instruction), timeout -> LOGGER.warn("The following operation is taking to long: {}", instruction)); instructionPublisher.publish(event.context(), event.getClientId(), instruction); } @Autowired  EventProcessorService(PlatformService instructionPublisher,
                                 InstructionResultSource.Factory instructionResultSource,
                                 AxonServerEventPublisher eventPublisher);  EventProcessorService(
            InstructionPublisher instructionPublisher,
            InstructionResultSource.Factory instructionResultSource,
            AxonServerEventPublisher eventPublisher); @EventListener void on(EventProcessorEvents.SplitSegmentRequest event); @EventListener void on(EventProcessorEvents.MergeSegmentRequest event); @EventListener void on(EventProcessorEvents.ReleaseSegmentRequest event); }
EventProcessorService { @EventListener public void on(EventProcessorEvents.SplitSegmentRequest event) { EventProcessorSegmentReference splitSegmentRequest = EventProcessorSegmentReference.newBuilder() .setProcessorName(event.getProcessorName()) .setSegmentIdentifier(event.getSegmentId()) .build(); PlatformOutboundInstruction instruction = PlatformOutboundInstruction.newBuilder() .setSplitEventProcessorSegment(splitSegmentRequest) .setInstructionId(UUID.randomUUID().toString()) .build(); SplitSegmentsSucceeded success = new SplitSegmentsSucceeded(event.context(), event.getClientId(), event.getProcessorName()); instructionResultSource .onInstructionResultFor(instruction.getInstructionId()) .subscribe(() -> eventPublisher.publishEvent(success), error -> LOGGER.warn("Error during segment split: {}, {}", error, instruction), timeout -> LOGGER.warn("The following operation is taking to long: {}", instruction)); instructionPublisher.publish(event.context(), event.getClientId(), instruction); } @Autowired  EventProcessorService(PlatformService instructionPublisher,
                                 InstructionResultSource.Factory instructionResultSource,
                                 AxonServerEventPublisher eventPublisher);  EventProcessorService(
            InstructionPublisher instructionPublisher,
            InstructionResultSource.Factory instructionResultSource,
            AxonServerEventPublisher eventPublisher); @EventListener void on(EventProcessorEvents.SplitSegmentRequest event); @EventListener void on(EventProcessorEvents.MergeSegmentRequest event); @EventListener void on(EventProcessorEvents.ReleaseSegmentRequest event);  }