AsyncMetricObserver extends BaseMetricObserver { public void stop() { stopUpdateThread = true; processingThread.interrupt(); } AsyncMetricObserver(
      String name,
      MetricObserver observer,
      int queueSize,
      long expireTime);  AsyncMetricObserver(String name, MetricObserver observer);  AsyncMetricObserver(String name, MetricObserver observer, int queueSize); void updateImpl(List<Metric> metrics); void stop();  }
@Test public void testUpdate() throws Exception { MemoryMetricObserver mmo = new MemoryMetricObserver("mem", 50); AsyncMetricObserver amo = new AsyncMetricObserver("async", mmo, 50); amo.update(mkList(1)); amo.update(mkList(2)); amo.update(mkList(3)); amo.update(mkList(4)); Thread.sleep(1000); assertEquals(mmo.getObservations(), UnmodifiableList.of(mkList(1), mkList(2), mkList(3), mkList(4))); amo.stop(); }
@Test public void testExceedQueueSize() throws Exception { MemoryMetricObserver mmo = new MemoryMetricObserver("mem", 50); MetricObserver smo = new SlowMetricObserver(mmo, 500L); AsyncMetricObserver amo = new AsyncMetricObserver("async", smo, 2); amo.update(mkList(1)); amo.update(mkList(2)); amo.update(mkList(3)); amo.update(mkList(4)); Thread.sleep(4000); List<List<Metric>> observations = mmo.getObservations(); assertTrue(observations.size() >= 2); int last = observations.size() - 1; assertEquals(observations.get(last - 1), mkList(3)); assertEquals(observations.get(last), mkList(4)); amo.stop(); }
@Test public void testExpiration() throws Exception { MemoryMetricObserver mmo = new MemoryMetricObserver("mem", 50); MetricObserver smo = new SlowMetricObserver(mmo, 500L); AsyncMetricObserver amo = new AsyncMetricObserver("async", smo, 50, 250); amo.update(mkList(1)); amo.update(mkList(2)); amo.update(mkList(3)); amo.update(mkList(4)); Thread.sleep(4000); List<List<Metric>> observations = mmo.getObservations(); assertEquals(observations.size(), 1); assertEquals(observations.get(0), mkList(1)); amo.stop(); }
@Test public void testFailedUpdate() throws Exception { MetricObserver fmo = new FailingMetricObserver(); AsyncMetricObserver amo = new AsyncMetricObserver("async", fmo, 50, 250); amo.update(mkList(1)); amo.update(mkList(1)); amo.update(mkList(1)); amo.update(mkList(1)); amo.update(mkList(1)); amo.update(mkList(1)); Thread.sleep(1000); amo.stop(); }