@Test public void testGetTrackedClients() throws Exception { EntityMessage message1 = mock(EntityMessage.class); EntityMessage message2 = mock(EntityMessage.class); EntityMessage message3 = mock(EntityMessage.class); EntityMessage message4 = mock(EntityMessage.class); ToIntFunction<EntityMessage> segmentationStrategy = m -> { if (m == message1 || m == message3) { return 0; } else if (m == message2 || m == message4) { return 1; } else { return -1; } }; Predicate<EntityMessage> trackerPolicy = msg -> { if (msg == message3) { return false; } else { return true; } }; messageHandler = new OOOMessageHandlerImpl<>(trackerPolicy, () -> {}); DummyClientSourceId clientSourceId1 = new DummyClientSourceId(1); DummyClientSourceId clientSourceId2 = new DummyClientSourceId(2); DummyClientSourceId clientSourceId3 = new DummyClientSourceId(3); InvokeContext context1 = new DummyContext(clientSourceId1, 25, 18); InvokeContext context2 = new DummyContext(clientSourceId2, 26, 18); InvokeContext context3 = new DummyContext(clientSourceId3, 27, 18); InvokeContext context4 = new DummyContext(clientSourceId1, 32, 28); messageHandler.invoke(context1, message1, (ctxt, msg) -> mock(EntityResponse.class)); messageHandler.invoke(context2, message2, (ctxt, msg) -> mock(EntityResponse.class)); messageHandler.invoke(context3, message3, (ctxt, msg) -> mock(EntityResponse.class)); messageHandler.invoke(context4, message4, (ctxt, msg) -> mock(EntityResponse.class)); Set<ClientSourceId> clients = messageHandler.getTrackedClients().collect(toSet()); assertThat(clients.size(), is(2)); assertThat(clients.contains(clientSourceId1), is(true)); assertThat(clients.contains(clientSourceId2), is(true)); assertThat(clients.contains(clientSourceId3), is(false)); }
@Override public Stream<ClientSourceId> getTrackedClients() { return clientMessageTracker.getTrackedClients().stream(); }
OOOMessageHandlerImpl implements OOOMessageHandler<M, R> { @Override public Stream<ClientSourceId> getTrackedClients() { return clientMessageTracker.getTrackedClients().stream(); } }
OOOMessageHandlerImpl implements OOOMessageHandler<M, R> { @Override public Stream<ClientSourceId> getTrackedClients() { return clientMessageTracker.getTrackedClients().stream(); } OOOMessageHandlerImpl(Predicate<M> trackerPolicy, DestroyCallback callback); }
OOOMessageHandlerImpl implements OOOMessageHandler<M, R> { @Override public Stream<ClientSourceId> getTrackedClients() { return clientMessageTracker.getTrackedClients().stream(); } OOOMessageHandlerImpl(Predicate<M> trackerPolicy, DestroyCallback callback); @Override R invoke(InvokeContext context, M message, BiFunction<InvokeContext, M, R> invokeFunction); @Override void closeDuplicatesWindow(); @Override R lookupResponse(ClientSourceId src, long txn); @Override void untrackClient(ClientSourceId clientSourceId); @Override Stream<ClientSourceId> getTrackedClients(); @Override Stream<RecordedMessage<M, R>> getRecordedMessages(); @Override void loadRecordedMessages(Stream<RecordedMessage<M, R>> recorded); @Override void destroy(); @Override void addStateTo(StateDumpCollector stateDumper); }
OOOMessageHandlerImpl implements OOOMessageHandler<M, R> { @Override public Stream<ClientSourceId> getTrackedClients() { return clientMessageTracker.getTrackedClients().stream(); } OOOMessageHandlerImpl(Predicate<M> trackerPolicy, DestroyCallback callback); @Override R invoke(InvokeContext context, M message, BiFunction<InvokeContext, M, R> invokeFunction); @Override void closeDuplicatesWindow(); @Override R lookupResponse(ClientSourceId src, long txn); @Override void untrackClient(ClientSourceId clientSourceId); @Override Stream<ClientSourceId> getTrackedClients(); @Override Stream<RecordedMessage<M, R>> getRecordedMessages(); @Override void loadRecordedMessages(Stream<RecordedMessage<M, R>> recorded); @Override void destroy(); @Override void addStateTo(StateDumpCollector stateDumper);  }