@Test void testMessages(){ EventProcessorLatencyMonitor testSubject = new EventProcessorLatencyMonitor(); EventMessage<?> firstEventMessage = mock(EventMessage.class); when(firstEventMessage.getTimestamp()).thenReturn(Instant.ofEpochMilli(0)); EventMessage<?> secondEventMessage = mock(EventMessage.class); when(secondEventMessage.getTimestamp()).thenReturn(Instant.ofEpochMilli(1000)); Map<? super EventMessage<?>, MessageMonitor.MonitorCallback> callbacks = testSubject.onMessagesIngested(Arrays.asList(firstEventMessage, secondEventMessage)); callbacks.get(firstEventMessage).reportSuccess(); Map<String, Metric> metricSet = testSubject.getMetrics(); Gauge<Long> latency = (Gauge<Long>) metricSet.get("latency"); assertEquals(1000, latency.getValue(), 0); }
@Override public Map<String, Metric> getMetrics() { long lastProcessedTime = this.lastProcessedTime.longValue(); long lastReceivedTime = this.lastReceivedTime.longValue(); long processTime; if(lastReceivedTime == -1 || lastProcessedTime == -1){ processTime = 0; } else { processTime = lastReceivedTime - lastProcessedTime; } Map<String, Metric> metrics = new HashMap<>(); metrics.put("latency", (Gauge<Long>) () -> processTime); return metrics; }
EventProcessorLatencyMonitor implements MessageMonitor<EventMessage<?>>, MetricSet { @Override public Map<String, Metric> getMetrics() { long lastProcessedTime = this.lastProcessedTime.longValue(); long lastReceivedTime = this.lastReceivedTime.longValue(); long processTime; if(lastReceivedTime == -1 || lastProcessedTime == -1){ processTime = 0; } else { processTime = lastReceivedTime - lastProcessedTime; } Map<String, Metric> metrics = new HashMap<>(); metrics.put("latency", (Gauge<Long>) () -> processTime); return metrics; } }
EventProcessorLatencyMonitor implements MessageMonitor<EventMessage<?>>, MetricSet { @Override public Map<String, Metric> getMetrics() { long lastProcessedTime = this.lastProcessedTime.longValue(); long lastReceivedTime = this.lastReceivedTime.longValue(); long processTime; if(lastReceivedTime == -1 || lastProcessedTime == -1){ processTime = 0; } else { processTime = lastReceivedTime - lastProcessedTime; } Map<String, Metric> metrics = new HashMap<>(); metrics.put("latency", (Gauge<Long>) () -> processTime); return metrics; }  }
EventProcessorLatencyMonitor implements MessageMonitor<EventMessage<?>>, MetricSet { @Override public Map<String, Metric> getMetrics() { long lastProcessedTime = this.lastProcessedTime.longValue(); long lastReceivedTime = this.lastReceivedTime.longValue(); long processTime; if(lastReceivedTime == -1 || lastProcessedTime == -1){ processTime = 0; } else { processTime = lastReceivedTime - lastProcessedTime; } Map<String, Metric> metrics = new HashMap<>(); metrics.put("latency", (Gauge<Long>) () -> processTime); return metrics; }  @Override MonitorCallback onMessageIngested(EventMessage<?> message); @Override Map<String, Metric> getMetrics(); }
EventProcessorLatencyMonitor implements MessageMonitor<EventMessage<?>>, MetricSet { @Override public Map<String, Metric> getMetrics() { long lastProcessedTime = this.lastProcessedTime.longValue(); long lastReceivedTime = this.lastReceivedTime.longValue(); long processTime; if(lastReceivedTime == -1 || lastProcessedTime == -1){ processTime = 0; } else { processTime = lastReceivedTime - lastProcessedTime; } Map<String, Metric> metrics = new HashMap<>(); metrics.put("latency", (Gauge<Long>) () -> processTime); return metrics; }  @Override MonitorCallback onMessageIngested(EventMessage<?> message); @Override Map<String, Metric> getMetrics();  }