IdempotentEventFilter implements Predicate<Event> { public static Predicate<Event> using(HazelcastInstance hazelcastInstance, String eventWindowName) { return new IdempotentEventFilter(hazelcastInstance.getMap(eventWindowName)); } private  IdempotentEventFilter(Map<SerializableEventIdentity, Boolean> eventsWindow); static Config configureCache(String eventWindowName, int timeToLiveSeconds, Config config); static Predicate<Event> using(HazelcastInstance hazelcastInstance, String eventWindowName); @Override boolean test(Event event);  }
@Test public void discardsDuplicatesWithinEventWindow() throws InterruptedException, ExecutionException { TestEvents eventHandler = mock(TestEvents.class); EventOutChannel filteredChannel = DispatchingEventOutChannel .toHandler(TestEvents.class, eventHandler) .filter(IdempotentEventFilter.using(hazelcastInstance, "eventWindow")); TestEvents proxy = EventEmittingProxy.proxying(filteredChannel, TestEvents.class); final StreamTimestamp timestamp = StreamTimestamp.now(); final String aggregateId = "id1"; proxy.testEvent(timestamp, aggregateId, "foo"); proxy.testEvent(timestamp, aggregateId, "bar"); verify(eventHandler).testEvent(timestamp, aggregateId, "foo"); verify(eventHandler, never()).testEvent(timestamp, aggregateId, "bar"); }
@Ignore("flickering") @Test public void allowsDuplicatesOutsideEventWindow() throws InterruptedException, ExecutionException { TestEvents eventHandler = mock(TestEvents.class); EventOutChannel filteredChannel = DispatchingEventOutChannel .toHandler(TestEvents.class, eventHandler) .filter(IdempotentEventFilter.using(hazelcastInstance, "eventWindow")); TestEvents proxy = EventEmittingProxy.proxying(filteredChannel, TestEvents.class); final StreamTimestamp timestamp = StreamTimestamp.now(); final String aggregateId = "id"; proxy.testEvent(timestamp, aggregateId, "foo"); try { Thread.sleep(1500); } catch (InterruptedException e) { throw new RuntimeException(e); } proxy.testEvent(timestamp, aggregateId, "bar"); verify(eventHandler).testEvent(timestamp, aggregateId, "foo"); verify(eventHandler).testEvent(timestamp, aggregateId, "bar"); }