@Test void testMessages(){ MessageCountingMonitor testSubject = new MessageCountingMonitor(); EventMessage<Object> foo = asEventMessage("foo"); EventMessage<Object> bar = asEventMessage("bar"); EventMessage<Object> baz = asEventMessage("baz"); Map<? super Message<?>, MessageMonitor.MonitorCallback> callbacks = testSubject.onMessagesIngested(Arrays.asList(foo, bar, baz)); callbacks.get(foo).reportSuccess(); callbacks.get(bar).reportFailure(null); callbacks.get(baz).reportIgnored(); Map<String, Metric> metricSet = testSubject.getMetrics(); Counter ingestedCounter = (Counter) metricSet.get("ingestedCounter"); Counter processedCounter = (Counter) metricSet.get("processedCounter"); Counter successCounter = (Counter) metricSet.get("successCounter"); Counter failureCounter = (Counter) metricSet.get("failureCounter"); Counter ignoredCounter = (Counter) metricSet.get("ignoredCounter"); assertEquals(3, ingestedCounter.getCount()); assertEquals(2, processedCounter.getCount()); assertEquals(1, successCounter.getCount()); assertEquals(1, failureCounter.getCount()); assertEquals(1, ignoredCounter.getCount()); }
@Override public Map<String, Metric> getMetrics() { Map<String, Metric> metricSet = new HashMap<>(); metricSet.put("ingestedCounter", ingestedCounter); metricSet.put("processedCounter", processedCounter); metricSet.put("successCounter", successCounter); metricSet.put("failureCounter", failureCounter); metricSet.put("ignoredCounter", ignoredCounter); return metricSet; }
MessageCountingMonitor implements MessageMonitor<Message<?>>, MetricSet { @Override public Map<String, Metric> getMetrics() { Map<String, Metric> metricSet = new HashMap<>(); metricSet.put("ingestedCounter", ingestedCounter); metricSet.put("processedCounter", processedCounter); metricSet.put("successCounter", successCounter); metricSet.put("failureCounter", failureCounter); metricSet.put("ignoredCounter", ignoredCounter); return metricSet; } }
MessageCountingMonitor implements MessageMonitor<Message<?>>, MetricSet { @Override public Map<String, Metric> getMetrics() { Map<String, Metric> metricSet = new HashMap<>(); metricSet.put("ingestedCounter", ingestedCounter); metricSet.put("processedCounter", processedCounter); metricSet.put("successCounter", successCounter); metricSet.put("failureCounter", failureCounter); metricSet.put("ignoredCounter", ignoredCounter); return metricSet; }  }
MessageCountingMonitor implements MessageMonitor<Message<?>>, MetricSet { @Override public Map<String, Metric> getMetrics() { Map<String, Metric> metricSet = new HashMap<>(); metricSet.put("ingestedCounter", ingestedCounter); metricSet.put("processedCounter", processedCounter); metricSet.put("successCounter", successCounter); metricSet.put("failureCounter", failureCounter); metricSet.put("ignoredCounter", ignoredCounter); return metricSet; }  @Override MonitorCallback onMessageIngested(Message<?> message); @Override Map<String, Metric> getMetrics(); }
MessageCountingMonitor implements MessageMonitor<Message<?>>, MetricSet { @Override public Map<String, Metric> getMetrics() { Map<String, Metric> metricSet = new HashMap<>(); metricSet.put("ingestedCounter", ingestedCounter); metricSet.put("processedCounter", processedCounter); metricSet.put("successCounter", successCounter); metricSet.put("failureCounter", failureCounter); metricSet.put("ignoredCounter", ignoredCounter); return metricSet; }  @Override MonitorCallback onMessageIngested(Message<?> message); @Override Map<String, Metric> getMetrics();  }