@Test public void testBasicDispatch() { in.put(keyA, digestA); in.flush(); timeMillis.set(101L); subject.run(); assertThat(pointOut).hasSize(1); assertThat(blockedOut).hasSize(0); assertThat(backingStore).isEmpty(); ReportPoint point = pointOut.get(0); TestUtils.testKeyPointMatch(keyA, point); }
@Override public void run() { try { AtomicInteger dispatchedCount = new AtomicInteger(0); long startMillis = System.currentTimeMillis(); digestsSize.set(digests.size()); Iterator<HistogramKey> index = digests.getRipeDigestsIterator(this.clock); while (index.hasNext()) { digests.compute(index.next(), (k, v) -> { if (v == null) { index.remove(); return null; } if (histogramDisabled.get()) { featureDisabledLogger.info("Histogram feature is not enabled on the server!"); dispatchErrorCounter.inc(); } else { try { ReportPoint out = HistogramUtils.pointFromKeyAndDigest(k, v); output.report(out); dispatchCounter.inc(); } catch (Exception e) { dispatchErrorCounter.inc(); logger.log(Level.SEVERE, "Failed dispatching entry " + k, e); } } index.remove(); dispatchedCount.incrementAndGet(); return null; }); if (dispatchLimit != null && dispatchedCount.get() >= dispatchLimit) break; } dispatchProcessTime.inc(System.currentTimeMillis() - startMillis); } catch (Exception e) { logger.log(Level.SEVERE, "PointHandlerDispatcher error", e); } }
PointHandlerDispatcher implements Runnable { @Override public void run() { try { AtomicInteger dispatchedCount = new AtomicInteger(0); long startMillis = System.currentTimeMillis(); digestsSize.set(digests.size()); Iterator<HistogramKey> index = digests.getRipeDigestsIterator(this.clock); while (index.hasNext()) { digests.compute(index.next(), (k, v) -> { if (v == null) { index.remove(); return null; } if (histogramDisabled.get()) { featureDisabledLogger.info("Histogram feature is not enabled on the server!"); dispatchErrorCounter.inc(); } else { try { ReportPoint out = HistogramUtils.pointFromKeyAndDigest(k, v); output.report(out); dispatchCounter.inc(); } catch (Exception e) { dispatchErrorCounter.inc(); logger.log(Level.SEVERE, "Failed dispatching entry " + k, e); } } index.remove(); dispatchedCount.incrementAndGet(); return null; }); if (dispatchLimit != null && dispatchedCount.get() >= dispatchLimit) break; } dispatchProcessTime.inc(System.currentTimeMillis() - startMillis); } catch (Exception e) { logger.log(Level.SEVERE, "PointHandlerDispatcher error", e); } } }
PointHandlerDispatcher implements Runnable { @Override public void run() { try { AtomicInteger dispatchedCount = new AtomicInteger(0); long startMillis = System.currentTimeMillis(); digestsSize.set(digests.size()); Iterator<HistogramKey> index = digests.getRipeDigestsIterator(this.clock); while (index.hasNext()) { digests.compute(index.next(), (k, v) -> { if (v == null) { index.remove(); return null; } if (histogramDisabled.get()) { featureDisabledLogger.info("Histogram feature is not enabled on the server!"); dispatchErrorCounter.inc(); } else { try { ReportPoint out = HistogramUtils.pointFromKeyAndDigest(k, v); output.report(out); dispatchCounter.inc(); } catch (Exception e) { dispatchErrorCounter.inc(); logger.log(Level.SEVERE, "Failed dispatching entry " + k, e); } } index.remove(); dispatchedCount.incrementAndGet(); return null; }); if (dispatchLimit != null && dispatchedCount.get() >= dispatchLimit) break; } dispatchProcessTime.inc(System.currentTimeMillis() - startMillis); } catch (Exception e) { logger.log(Level.SEVERE, "PointHandlerDispatcher error", e); } } PointHandlerDispatcher(Accumulator digests,
                                ReportableEntityHandler<ReportPoint, String> output,
                                TimeProvider clock,
                                Supplier<Boolean> histogramDisabled,
                                @Nullable Integer dispatchLimit,
                                @Nullable Granularity granularity); }
PointHandlerDispatcher implements Runnable { @Override public void run() { try { AtomicInteger dispatchedCount = new AtomicInteger(0); long startMillis = System.currentTimeMillis(); digestsSize.set(digests.size()); Iterator<HistogramKey> index = digests.getRipeDigestsIterator(this.clock); while (index.hasNext()) { digests.compute(index.next(), (k, v) -> { if (v == null) { index.remove(); return null; } if (histogramDisabled.get()) { featureDisabledLogger.info("Histogram feature is not enabled on the server!"); dispatchErrorCounter.inc(); } else { try { ReportPoint out = HistogramUtils.pointFromKeyAndDigest(k, v); output.report(out); dispatchCounter.inc(); } catch (Exception e) { dispatchErrorCounter.inc(); logger.log(Level.SEVERE, "Failed dispatching entry " + k, e); } } index.remove(); dispatchedCount.incrementAndGet(); return null; }); if (dispatchLimit != null && dispatchedCount.get() >= dispatchLimit) break; } dispatchProcessTime.inc(System.currentTimeMillis() - startMillis); } catch (Exception e) { logger.log(Level.SEVERE, "PointHandlerDispatcher error", e); } } PointHandlerDispatcher(Accumulator digests,
                                ReportableEntityHandler<ReportPoint, String> output,
                                TimeProvider clock,
                                Supplier<Boolean> histogramDisabled,
                                @Nullable Integer dispatchLimit,
                                @Nullable Granularity granularity); @Override void run(); }
PointHandlerDispatcher implements Runnable { @Override public void run() { try { AtomicInteger dispatchedCount = new AtomicInteger(0); long startMillis = System.currentTimeMillis(); digestsSize.set(digests.size()); Iterator<HistogramKey> index = digests.getRipeDigestsIterator(this.clock); while (index.hasNext()) { digests.compute(index.next(), (k, v) -> { if (v == null) { index.remove(); return null; } if (histogramDisabled.get()) { featureDisabledLogger.info("Histogram feature is not enabled on the server!"); dispatchErrorCounter.inc(); } else { try { ReportPoint out = HistogramUtils.pointFromKeyAndDigest(k, v); output.report(out); dispatchCounter.inc(); } catch (Exception e) { dispatchErrorCounter.inc(); logger.log(Level.SEVERE, "Failed dispatching entry " + k, e); } } index.remove(); dispatchedCount.incrementAndGet(); return null; }); if (dispatchLimit != null && dispatchedCount.get() >= dispatchLimit) break; } dispatchProcessTime.inc(System.currentTimeMillis() - startMillis); } catch (Exception e) { logger.log(Level.SEVERE, "PointHandlerDispatcher error", e); } } PointHandlerDispatcher(Accumulator digests,
                                ReportableEntityHandler<ReportPoint, String> output,
                                TimeProvider clock,
                                Supplier<Boolean> histogramDisabled,
                                @Nullable Integer dispatchLimit,
                                @Nullable Granularity granularity); @Override void run();  }