GRPCExporter extends MetricFormatter implements MetricValuesExportService, IConsumer<ExportData> { @Override public void consume(List<ExportData> data) { if (data.size() == 0) { return; } GRPCStreamStatus status = new GRPCStreamStatus(); StreamObserver<ExportMetricValue> streamObserver = exportServiceFutureStub.withDeadlineAfter( 10, TimeUnit.SECONDS) .export( new StreamObserver<ExportResponse>() { @Override public void onNext( ExportResponse response) { } @Override public void onError( Throwable throwable) { status.done(); } @Override public void onCompleted() { status.done(); } }); AtomicInteger exportNum = new AtomicInteger(); data.forEach(row -> { ExportMetricValue.Builder builder = ExportMetricValue.newBuilder(); Metrics metrics = row.getMetrics(); if (metrics instanceof LongValueHolder) { long value = ((LongValueHolder) metrics).getValue(); builder.setLongValue(value); builder.setType(ValueType.LONG); } else if (metrics instanceof IntValueHolder) { long value = ((IntValueHolder) metrics).getValue(); builder.setLongValue(value); builder.setType(ValueType.LONG); } else if (metrics instanceof DoubleValueHolder) { double value = ((DoubleValueHolder) metrics).getValue(); builder.setDoubleValue(value); builder.setType(ValueType.DOUBLE); } else if (metrics instanceof MultiIntValuesHolder) { int[] values = ((MultiIntValuesHolder) metrics).getValues(); for (int value : values) { builder.addLongValues(value); } builder.setType(ValueType.MULTI_LONG); } else { return; } MetricsMetaInfo meta = row.getMeta(); builder.setMetricName(meta.getMetricsName()); String entityName = getEntityName(meta); if (entityName == null) { return; } builder.setEntityName(entityName); builder.setEntityId(meta.getId()); builder.setTimeBucket(metrics.getTimeBucket()); streamObserver.onNext(builder.build()); exportNum.getAndIncrement(); }); streamObserver.onCompleted(); long sleepTime = 0; long cycle = 100L; while (!status.isDone()) { try { sleepTime += cycle; Thread.sleep(cycle); } catch (InterruptedException e) { } if (sleepTime > 2000L) { LOGGER.warn( "Export {} metrics to {}:{}, wait {} milliseconds.", exportNum.get(), setting.getTargetHost(), setting .getTargetPort(), sleepTime ); cycle = 2000L; } } LOGGER.debug( "Exported {} metrics to {}:{} in {} milliseconds.", exportNum.get(), setting.getTargetHost(), setting .getTargetPort(), sleepTime); } GRPCExporter(GRPCExporterSetting setting); @Override void export(ExportEvent event); void initSubscriptionList(); @Override void init(); @Override void consume(List<ExportData> data); @Override void onError(List<ExportData> data, Throwable t); @Override void onExit();  }
@Test public void consume() { exporter.consume(dataList()); exporter.consume(Collections.emptyList()); }