@Test public void testOnFailure() { AtomicBoolean successReceived = new AtomicBoolean(false); AtomicBoolean failureReceived = new AtomicBoolean(false); AtomicBoolean timeoutReceived = new AtomicBoolean(false); InstructionResultSource instructionResultSource = testSubject.onInstructionResultFor("MyInstructionId"); instructionResultSource.subscribe(() -> successReceived.set(true), error -> failureReceived.set(true), timout -> timeoutReceived.set(true)); notifyInstructionResultReceived(InstructionResult.newBuilder() .setInstructionId("AnotherId") .build()); assertFalse(successReceived.get()); assertFalse(failureReceived.get()); assertFalse(timeoutReceived.get()); notifyInstructionResultReceived(InstructionResult.newBuilder() .setInstructionId("MyInstructionId") .setSuccess(false) .build()); assertFalse(successReceived.get()); assertTrue(failureReceived.get()); assertFalse(timeoutReceived.get()); }
@Override public InstructionResultSource onInstructionResultFor(String instructionId) { return new InstructionResultSource() { @Override public void subscribe(ResultSubscriber resultSubscriber, Duration timeout) { Function<String, List<ResultSubscriber>> copyOnWriteArrayList = id -> new CopyOnWriteArrayList<>(); List<ResultSubscriber> subscribers = subscribersMap.computeIfAbsent(instructionId, copyOnWriteArrayList); subscribers.add(new TimeoutResultSubscriber(resultSubscriber, timeout, subscribers::remove)); } @Override public Duration defaultTimeout() { return Duration.ofSeconds(resultTimeout); } }; }
InstructionResultSourceFactory implements InstructionResultSource.Factory { @Override public InstructionResultSource onInstructionResultFor(String instructionId) { return new InstructionResultSource() { @Override public void subscribe(ResultSubscriber resultSubscriber, Duration timeout) { Function<String, List<ResultSubscriber>> copyOnWriteArrayList = id -> new CopyOnWriteArrayList<>(); List<ResultSubscriber> subscribers = subscribersMap.computeIfAbsent(instructionId, copyOnWriteArrayList); subscribers.add(new TimeoutResultSubscriber(resultSubscriber, timeout, subscribers::remove)); } @Override public Duration defaultTimeout() { return Duration.ofSeconds(resultTimeout); } }; } }
InstructionResultSourceFactory implements InstructionResultSource.Factory { @Override public InstructionResultSource onInstructionResultFor(String instructionId) { return new InstructionResultSource() { @Override public void subscribe(ResultSubscriber resultSubscriber, Duration timeout) { Function<String, List<ResultSubscriber>> copyOnWriteArrayList = id -> new CopyOnWriteArrayList<>(); List<ResultSubscriber> subscribers = subscribersMap.computeIfAbsent(instructionId, copyOnWriteArrayList); subscribers.add(new TimeoutResultSubscriber(resultSubscriber, timeout, subscribers::remove)); } @Override public Duration defaultTimeout() { return Duration.ofSeconds(resultTimeout); } }; } @Autowired  InstructionResultSourceFactory(PlatformService platformService,
                                          @Value("${axoniq.axonserver.instruction.result.timeout:10}") int resultTimeout);  InstructionResultSourceFactory(Consumer<Consumer<InstructionResult>> registration,
                                          int resultTimeout); }
InstructionResultSourceFactory implements InstructionResultSource.Factory { @Override public InstructionResultSource onInstructionResultFor(String instructionId) { return new InstructionResultSource() { @Override public void subscribe(ResultSubscriber resultSubscriber, Duration timeout) { Function<String, List<ResultSubscriber>> copyOnWriteArrayList = id -> new CopyOnWriteArrayList<>(); List<ResultSubscriber> subscribers = subscribersMap.computeIfAbsent(instructionId, copyOnWriteArrayList); subscribers.add(new TimeoutResultSubscriber(resultSubscriber, timeout, subscribers::remove)); } @Override public Duration defaultTimeout() { return Duration.ofSeconds(resultTimeout); } }; } @Autowired  InstructionResultSourceFactory(PlatformService platformService,
                                          @Value("${axoniq.axonserver.instruction.result.timeout:10}") int resultTimeout);  InstructionResultSourceFactory(Consumer<Consumer<InstructionResult>> registration,
                                          int resultTimeout); @Override InstructionResultSource onInstructionResultFor(String instructionId); }
InstructionResultSourceFactory implements InstructionResultSource.Factory { @Override public InstructionResultSource onInstructionResultFor(String instructionId) { return new InstructionResultSource() { @Override public void subscribe(ResultSubscriber resultSubscriber, Duration timeout) { Function<String, List<ResultSubscriber>> copyOnWriteArrayList = id -> new CopyOnWriteArrayList<>(); List<ResultSubscriber> subscribers = subscribersMap.computeIfAbsent(instructionId, copyOnWriteArrayList); subscribers.add(new TimeoutResultSubscriber(resultSubscriber, timeout, subscribers::remove)); } @Override public Duration defaultTimeout() { return Duration.ofSeconds(resultTimeout); } }; } @Autowired  InstructionResultSourceFactory(PlatformService platformService,
                                          @Value("${axoniq.axonserver.instruction.result.timeout:10}") int resultTimeout);  InstructionResultSourceFactory(Consumer<Consumer<InstructionResult>> registration,
                                          int resultTimeout); @Override InstructionResultSource onInstructionResultFor(String instructionId);  }