@Test public void sendMessage() throws Exception { JsonNode value = mapper.createObjectNode(); Payload<JsonNode> payload = new Payload<JsonNode>((byte) 0, 2, value); Record record = new Record(0, 1L, 3L, payload); Message<JsonNode> message = new Message<>(0, 1L, 2, 3L, value); doReturn(value).when(messageFunction).apply(payload); doReturn(message).when(metrics).record(eq(MESSAGE), argThat(new ArgMatcher<Supplier<Message<JsonNode>>>() {})); doNothing().when(underTest).sendEvent(message); underTest.handleIncomingEvent(new Request(1L)); underTest.sendMessage(record); verify(underTest).sendEvent(message); assertThat(underTest.getRequested(), is(0L)); @SuppressWarnings("unchecked") ArgumentCaptor<Supplier<Message<JsonNode>>> captor = ArgumentCaptor.forClass(Supplier.class); verify(metrics).record(eq(MESSAGE), captor.capture()); Message<JsonNode> message2 = captor.getValue().get(); assertThat(message2, is(message)); }
@VisibleForTesting void sendMessage(Record record) { Message<JsonNode> message = metrics.record(MESSAGE, () -> { Payload<JsonNode> payload = record.getPayload(); return new Message<>(record.getPartition(), record.getOffset(), payload.getSchemaVersion(), record.getTimestampMs(), messageFunction.apply(payload)); }); sendEvent(message); metrics.markMessageLatency(message.getPartition(), message.getTimestampMs()); requested--; if (log.isDebugEnabled() && requested == 0) { log.debug("Completed sending all requested messages"); } }
OfframpServiceV2 implements OfframpService { @VisibleForTesting void sendMessage(Record record) { Message<JsonNode> message = metrics.record(MESSAGE, () -> { Payload<JsonNode> payload = record.getPayload(); return new Message<>(record.getPartition(), record.getOffset(), payload.getSchemaVersion(), record.getTimestampMs(), messageFunction.apply(payload)); }); sendEvent(message); metrics.markMessageLatency(message.getPartition(), message.getTimestampMs()); requested--; if (log.isDebugEnabled() && requested == 0) { log.debug("Completed sending all requested messages"); } } }
OfframpServiceV2 implements OfframpService { @VisibleForTesting void sendMessage(Record record) { Message<JsonNode> message = metrics.record(MESSAGE, () -> { Payload<JsonNode> payload = record.getPayload(); return new Message<>(record.getPartition(), record.getOffset(), payload.getSchemaVersion(), record.getTimestampMs(), messageFunction.apply(payload)); }); sendEvent(message); metrics.markMessageLatency(message.getPartition(), message.getTimestampMs()); requested--; if (log.isDebugEnabled() && requested == 0) { log.debug("Completed sending all requested messages"); } }  }
OfframpServiceV2 implements OfframpService { @VisibleForTesting void sendMessage(Record record) { Message<JsonNode> message = metrics.record(MESSAGE, () -> { Payload<JsonNode> payload = record.getPayload(); return new Message<>(record.getPartition(), record.getOffset(), payload.getSchemaVersion(), record.getTimestampMs(), messageFunction.apply(payload)); }); sendEvent(message); metrics.markMessageLatency(message.getPartition(), message.getTimestampMs()); requested--; if (log.isDebugEnabled() && requested == 0) { log.debug("Completed sending all requested messages"); } }  @Override void run(); @Override void close(); @Override void onEvent(String event); }
OfframpServiceV2 implements OfframpService { @VisibleForTesting void sendMessage(Record record) { Message<JsonNode> message = metrics.record(MESSAGE, () -> { Payload<JsonNode> payload = record.getPayload(); return new Message<>(record.getPartition(), record.getOffset(), payload.getSchemaVersion(), record.getTimestampMs(), messageFunction.apply(payload)); }); sendEvent(message); metrics.markMessageLatency(message.getPartition(), message.getTimestampMs()); requested--; if (log.isDebugEnabled() && requested == 0) { log.debug("Completed sending all requested messages"); } }  @Override void run(); @Override void close(); @Override void onEvent(String event);  }