@Test public void testHandleEvent() throws Exception { EventCollector collector = new EventCollector(); assertThat(collector.getEvents().count(), equalTo(0l)); collector.handleEvent(event("myTopic")); assertThat(collector.getEvents().count(), equalTo(1l)); assertThat(collector.getEvents().findFirst().get().getTopic(), equalTo("myTopic")); }
@Override public synchronized void handleEvent(Event event) { events.addLast(event); if (events.size() > maxSize) { events.removeFirst(); } consumers.forEach(c -> c.accept(event)); }
EventCollector implements EventHandler { @Override public synchronized void handleEvent(Event event) { events.addLast(event); if (events.size() > maxSize) { events.removeFirst(); } consumers.forEach(c -> c.accept(event)); } }
EventCollector implements EventHandler { @Override public synchronized void handleEvent(Event event) { events.addLast(event); if (events.size() > maxSize) { events.removeFirst(); } consumers.forEach(c -> c.accept(event)); } EventCollector(); }
EventCollector implements EventHandler { @Override public synchronized void handleEvent(Event event) { events.addLast(event); if (events.size() > maxSize) { events.removeFirst(); } consumers.forEach(c -> c.accept(event)); } EventCollector(); @Override synchronized void handleEvent(Event event); Stream<Event> getEvents(); synchronized void addConsumer(Consumer<Event> eventConsumer); synchronized void removeConsumer(Consumer<Event> eventConsumer); }
EventCollector implements EventHandler { @Override public synchronized void handleEvent(Event event) { events.addLast(event); if (events.size() > maxSize) { events.removeFirst(); } consumers.forEach(c -> c.accept(event)); } EventCollector(); @Override synchronized void handleEvent(Event event); Stream<Event> getEvents(); synchronized void addConsumer(Consumer<Event> eventConsumer); synchronized void removeConsumer(Consumer<Event> eventConsumer);  }