@Test public void onExecption() throws InterruptedException { AtomicInteger triggers = new AtomicInteger(); WebsocketProcessorEventsSource testSubject = new WebsocketProcessorEventsSource(() -> { if (triggers.getAndIncrement() == 5) { throw new IllegalArgumentException("Failed"); } }); ScheduledFuture<?> scheduler = Executors.newSingleThreadScheduledExecutor() .scheduleAtFixedRate(testSubject::applyIfUpdates, 50, 50, TimeUnit.MILLISECONDS); for (int run = 0; run < 10; run++) { IntStream.range(0, 1000).parallel().forEach(i -> testSubject .on(new EventProcessorEvents.EventProcessorStatusUpdate(null))); Thread.sleep(100); } assertEquals(10, triggers.get()); scheduler.cancel(false); }
@EventListener public void on(EventProcessorStatusUpdate event) { updates.set(true); }
WebsocketProcessorEventsSource { @EventListener public void on(EventProcessorStatusUpdate event) { updates.set(true); } }
WebsocketProcessorEventsSource { @EventListener public void on(EventProcessorStatusUpdate event) { updates.set(true); } @Autowired  WebsocketProcessorEventsSource(SimpMessagingTemplate websocket);  WebsocketProcessorEventsSource(Runnable updatesConsumer); }
WebsocketProcessorEventsSource { @EventListener public void on(EventProcessorStatusUpdate event) { updates.set(true); } @Autowired  WebsocketProcessorEventsSource(SimpMessagingTemplate websocket);  WebsocketProcessorEventsSource(Runnable updatesConsumer); @Scheduled(initialDelayString = "${axoniq.axonserver.websocket-update.initial-delay:10000}", fixedRateString = "${axoniq.axonserver.websocket-update.rate:1000}") void applyIfUpdates(); @EventListener void on(EventProcessorStatusUpdate event); }
WebsocketProcessorEventsSource { @EventListener public void on(EventProcessorStatusUpdate event) { updates.set(true); } @Autowired  WebsocketProcessorEventsSource(SimpMessagingTemplate websocket);  WebsocketProcessorEventsSource(Runnable updatesConsumer); @Scheduled(initialDelayString = "${axoniq.axonserver.websocket-update.initial-delay:10000}", fixedRateString = "${axoniq.axonserver.websocket-update.rate:1000}") void applyIfUpdates(); @EventListener void on(EventProcessorStatusUpdate event);  }