@Test void testMultithreadedCapacity() { TestClock testClock = new TestClock(); CapacityMonitor testSubject = new CapacityMonitor(1, TimeUnit.SECONDS, testClock); EventMessage<Object> foo = asEventMessage("foo"); EventMessage<Object> bar = asEventMessage("bar"); Map<? super Message<?>, MessageMonitor.MonitorCallback> callbacks = testSubject .onMessagesIngested(Arrays.asList(foo, bar)); testClock.increase(1000); callbacks.get(foo).reportSuccess(); callbacks.get(bar).reportFailure(null); Map<String, Metric> metricSet = testSubject.getMetrics(); Gauge<Double> capacityGauge = (Gauge<Double>) metricSet.get("capacity"); assertEquals(2, capacityGauge.getValue(), 0); }
@Override public Map<String, Metric> getMetrics() { Map<String, Metric> metrics = new HashMap<>(); metrics.put("capacity", capacity); return metrics; }
CapacityMonitor implements MessageMonitor<Message<?>>, MetricSet { @Override public Map<String, Metric> getMetrics() { Map<String, Metric> metrics = new HashMap<>(); metrics.put("capacity", capacity); return metrics; } }
CapacityMonitor implements MessageMonitor<Message<?>>, MetricSet { @Override public Map<String, Metric> getMetrics() { Map<String, Metric> metrics = new HashMap<>(); metrics.put("capacity", capacity); return metrics; } CapacityMonitor();  CapacityMonitor(long window, TimeUnit timeUnit);  CapacityMonitor(long window, TimeUnit timeUnit, Clock clock); }
CapacityMonitor implements MessageMonitor<Message<?>>, MetricSet { @Override public Map<String, Metric> getMetrics() { Map<String, Metric> metrics = new HashMap<>(); metrics.put("capacity", capacity); return metrics; } CapacityMonitor();  CapacityMonitor(long window, TimeUnit timeUnit);  CapacityMonitor(long window, TimeUnit timeUnit, Clock clock); @Override MonitorCallback onMessageIngested(Message<?> message); @Override Map<String, Metric> getMetrics(); }
CapacityMonitor implements MessageMonitor<Message<?>>, MetricSet { @Override public Map<String, Metric> getMetrics() { Map<String, Metric> metrics = new HashMap<>(); metrics.put("capacity", capacity); return metrics; } CapacityMonitor();  CapacityMonitor(long window, TimeUnit timeUnit);  CapacityMonitor(long window, TimeUnit timeUnit, Clock clock); @Override MonitorCallback onMessageIngested(Message<?> message); @Override Map<String, Metric> getMetrics();  }