@Test public void shouldProcessEvents() throws InterruptedException, ExecutionException, TimeoutException { reset(handlers); for (final ResourceUpdateHandler handler : handlers) { when(handler.canHandle(event)).thenReturn(true); } final CompletableFuture<Void> processed = controller.onEventInternal(EventBus.Type.CHANGE_EVENT, JsonUtils.toString(event)); processed.get(1, TimeUnit.SECONDS); for (final ResourceUpdateHandler handler : handlers) { verify(handler).process(event); } }
CompletableFuture<Void> onEventInternal(final String event, final String data) { if (!running.get()) { return CompletableFuture.completedFuture(null); } if (!Objects.equals(event, EventBus.Type.CHANGE_EVENT)) { return CompletableFuture.completedFuture(null); } final ChangeEvent changeEvent; try { changeEvent = JsonUtils.reader().forType(ChangeEvent.class).readValue(data); } catch (final IOException e) { LOGGER.error("Error while processing change-event {}", data, e); final CompletableFuture<Void> errored = new CompletableFuture<>(); errored.completeExceptionally(e); return errored; } final CompletableFuture<Void> done = new CompletableFuture<>(); if (changeEvent != null) { scheduler.execute(() -> run(changeEvent, done)); } return done; }
ResourceUpdateController { CompletableFuture<Void> onEventInternal(final String event, final String data) { if (!running.get()) { return CompletableFuture.completedFuture(null); } if (!Objects.equals(event, EventBus.Type.CHANGE_EVENT)) { return CompletableFuture.completedFuture(null); } final ChangeEvent changeEvent; try { changeEvent = JsonUtils.reader().forType(ChangeEvent.class).readValue(data); } catch (final IOException e) { LOGGER.error("Error while processing change-event {}", data, e); final CompletableFuture<Void> errored = new CompletableFuture<>(); errored.completeExceptionally(e); return errored; } final CompletableFuture<Void> done = new CompletableFuture<>(); if (changeEvent != null) { scheduler.execute(() -> run(changeEvent, done)); } return done; } }
ResourceUpdateController { CompletableFuture<Void> onEventInternal(final String event, final String data) { if (!running.get()) { return CompletableFuture.completedFuture(null); } if (!Objects.equals(event, EventBus.Type.CHANGE_EVENT)) { return CompletableFuture.completedFuture(null); } final ChangeEvent changeEvent; try { changeEvent = JsonUtils.reader().forType(ChangeEvent.class).readValue(data); } catch (final IOException e) { LOGGER.error("Error while processing change-event {}", data, e); final CompletableFuture<Void> errored = new CompletableFuture<>(); errored.completeExceptionally(e); return errored; } final CompletableFuture<Void> done = new CompletableFuture<>(); if (changeEvent != null) { scheduler.execute(() -> run(changeEvent, done)); } return done; } @Autowired  ResourceUpdateController(final ResourceUpdateConfiguration configuration, final EventBus eventBus, final List<ResourceUpdateHandler> handlers); }
ResourceUpdateController { CompletableFuture<Void> onEventInternal(final String event, final String data) { if (!running.get()) { return CompletableFuture.completedFuture(null); } if (!Objects.equals(event, EventBus.Type.CHANGE_EVENT)) { return CompletableFuture.completedFuture(null); } final ChangeEvent changeEvent; try { changeEvent = JsonUtils.reader().forType(ChangeEvent.class).readValue(data); } catch (final IOException e) { LOGGER.error("Error while processing change-event {}", data, e); final CompletableFuture<Void> errored = new CompletableFuture<>(); errored.completeExceptionally(e); return errored; } final CompletableFuture<Void> done = new CompletableFuture<>(); if (changeEvent != null) { scheduler.execute(() -> run(changeEvent, done)); } return done; } @Autowired  ResourceUpdateController(final ResourceUpdateConfiguration configuration, final EventBus eventBus, final List<ResourceUpdateHandler> handlers); void start(); void stop(); }
ResourceUpdateController { CompletableFuture<Void> onEventInternal(final String event, final String data) { if (!running.get()) { return CompletableFuture.completedFuture(null); } if (!Objects.equals(event, EventBus.Type.CHANGE_EVENT)) { return CompletableFuture.completedFuture(null); } final ChangeEvent changeEvent; try { changeEvent = JsonUtils.reader().forType(ChangeEvent.class).readValue(data); } catch (final IOException e) { LOGGER.error("Error while processing change-event {}", data, e); final CompletableFuture<Void> errored = new CompletableFuture<>(); errored.completeExceptionally(e); return errored; } final CompletableFuture<Void> done = new CompletableFuture<>(); if (changeEvent != null) { scheduler.execute(() -> run(changeEvent, done)); } return done; } @Autowired  ResourceUpdateController(final ResourceUpdateConfiguration configuration, final EventBus eventBus, final List<ResourceUpdateHandler> handlers); void start(); void stop();  }