@Test public void testInvokeDoesNotCacheUntrackableResponse() throws Exception { messageHandler = new OOOMessageHandlerImpl<>(msg -> false, () -> {}); InvokeContext context = new DummyContext(new DummyClientSourceId(1), 25, 18); EntityMessage message = mock(EntityMessage.class); EntityResponse response1 = mock(EntityResponse.class); EntityResponse entityResponse1 = messageHandler.invoke(context, message, (ctxt, msg) -> response1); assertThat(entityResponse1, is(response1)); EntityResponse response2 = mock(EntityResponse.class); EntityResponse entityResponse2 = messageHandler.invoke(context, message, (ctxt, msg) -> response2); assertThat(entityResponse2, is(response2)); }
@Override public R invoke(InvokeContext context, M message, BiFunction<InvokeContext, M, R> invokeFunction) throws EntityUserException { if (trackerPolicy.test(message) && context.isValidClientInformation()) { ClientSourceId clientId = context.getClientSource(); Tracker<M, R> messageTracker = clientMessageTracker.getTracker(clientId); messageTracker.reconcile(context.getOldestTransactionId()); if (checkDuplicates.get()) { if (messageTracker.wasReconciled(context.getCurrentTransactionId())) { return null; } R response = messageTracker.getTrackedValue(context.getCurrentTransactionId()); if (response != null) { return response; } } R response = invokeFunction.apply(context, message); messageTracker.track(trackid.incrementAndGet(), context.getCurrentTransactionId(), message, response); return response; } else { return invokeFunction.apply(context, message); } }
OOOMessageHandlerImpl implements OOOMessageHandler<M, R> { @Override public R invoke(InvokeContext context, M message, BiFunction<InvokeContext, M, R> invokeFunction) throws EntityUserException { if (trackerPolicy.test(message) && context.isValidClientInformation()) { ClientSourceId clientId = context.getClientSource(); Tracker<M, R> messageTracker = clientMessageTracker.getTracker(clientId); messageTracker.reconcile(context.getOldestTransactionId()); if (checkDuplicates.get()) { if (messageTracker.wasReconciled(context.getCurrentTransactionId())) { return null; } R response = messageTracker.getTrackedValue(context.getCurrentTransactionId()); if (response != null) { return response; } } R response = invokeFunction.apply(context, message); messageTracker.track(trackid.incrementAndGet(), context.getCurrentTransactionId(), message, response); return response; } else { return invokeFunction.apply(context, message); } } }
OOOMessageHandlerImpl implements OOOMessageHandler<M, R> { @Override public R invoke(InvokeContext context, M message, BiFunction<InvokeContext, M, R> invokeFunction) throws EntityUserException { if (trackerPolicy.test(message) && context.isValidClientInformation()) { ClientSourceId clientId = context.getClientSource(); Tracker<M, R> messageTracker = clientMessageTracker.getTracker(clientId); messageTracker.reconcile(context.getOldestTransactionId()); if (checkDuplicates.get()) { if (messageTracker.wasReconciled(context.getCurrentTransactionId())) { return null; } R response = messageTracker.getTrackedValue(context.getCurrentTransactionId()); if (response != null) { return response; } } R response = invokeFunction.apply(context, message); messageTracker.track(trackid.incrementAndGet(), context.getCurrentTransactionId(), message, response); return response; } else { return invokeFunction.apply(context, message); } } OOOMessageHandlerImpl(Predicate<M> trackerPolicy, DestroyCallback callback); }
OOOMessageHandlerImpl implements OOOMessageHandler<M, R> { @Override public R invoke(InvokeContext context, M message, BiFunction<InvokeContext, M, R> invokeFunction) throws EntityUserException { if (trackerPolicy.test(message) && context.isValidClientInformation()) { ClientSourceId clientId = context.getClientSource(); Tracker<M, R> messageTracker = clientMessageTracker.getTracker(clientId); messageTracker.reconcile(context.getOldestTransactionId()); if (checkDuplicates.get()) { if (messageTracker.wasReconciled(context.getCurrentTransactionId())) { return null; } R response = messageTracker.getTrackedValue(context.getCurrentTransactionId()); if (response != null) { return response; } } R response = invokeFunction.apply(context, message); messageTracker.track(trackid.incrementAndGet(), context.getCurrentTransactionId(), message, response); return response; } else { return invokeFunction.apply(context, message); } } 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 R invoke(InvokeContext context, M message, BiFunction<InvokeContext, M, R> invokeFunction) throws EntityUserException { if (trackerPolicy.test(message) && context.isValidClientInformation()) { ClientSourceId clientId = context.getClientSource(); Tracker<M, R> messageTracker = clientMessageTracker.getTracker(clientId); messageTracker.reconcile(context.getOldestTransactionId()); if (checkDuplicates.get()) { if (messageTracker.wasReconciled(context.getCurrentTransactionId())) { return null; } R response = messageTracker.getTrackedValue(context.getCurrentTransactionId()); if (response != null) { return response; } } R response = invokeFunction.apply(context, message); messageTracker.track(trackid.incrementAndGet(), context.getCurrentTransactionId(), message, response); return response; } else { return invokeFunction.apply(context, message); } } 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);  }