@Test public void reportFailuresAreRecoveredWithoutBatching() throws Exception { HttpReporter reporter = initReporter(Builder::disableSpanBatching); verifySpanSendFailuresAreRecovered(reporter, false); }
private PublisherSource.Processor<Span, Span> initReporter(final Builder builder, final HttpClient client) { final PublisherSource.Processor<Span, Span> buffer; SpanBytesEncoder spanEncoder = builder.codec.spanBytesEncoder(); final BufferAllocator allocator = client.executionContext().bufferAllocator(); final Publisher<Buffer> spans; if (builder.disableBatching) { buffer = newPublisherProcessorDropHeadOnOverflow(builder.maxConcurrentReports); spans = fromSource(buffer).map(span -> allocator.wrap(spanEncoder.encode(span))); } else { buffer = newPublisherProcessorDropHeadOnOverflow(builder.batchSizeHint * builder.maxConcurrentReports); spans = fromSource(buffer) .buffer(forCountOrTime(builder.batchSizeHint, builder.maxBatchDuration, () -> new ListAccumulator(builder.batchSizeHint), client.executionContext().executor())) .filter(accumulate -> !accumulate.isEmpty()) .map(bufferedSpans -> allocator.wrap(spanEncoder.encodeList(bufferedSpans))); } final CompletableSource.Processor spansTerminated = newCompletableProcessor(); toSource(spans.flatMapCompletable(encodedSpansReporter(client, builder.codec), builder.maxConcurrentReports)) .subscribe(spansTerminated); closeable.prepend(toAsyncCloseable(graceful -> { closeInitiated = true; try { buffer.onComplete(); } catch (Throwable t) { LOGGER.error("Failed to dispose request buffer. Ignoring.", t); } return graceful ? fromSource(spansTerminated) : completed(); })); return buffer; }
HttpReporter extends Component implements Reporter<Span>, AsyncCloseable { private PublisherSource.Processor<Span, Span> initReporter(final Builder builder, final HttpClient client) { final PublisherSource.Processor<Span, Span> buffer; SpanBytesEncoder spanEncoder = builder.codec.spanBytesEncoder(); final BufferAllocator allocator = client.executionContext().bufferAllocator(); final Publisher<Buffer> spans; if (builder.disableBatching) { buffer = newPublisherProcessorDropHeadOnOverflow(builder.maxConcurrentReports); spans = fromSource(buffer).map(span -> allocator.wrap(spanEncoder.encode(span))); } else { buffer = newPublisherProcessorDropHeadOnOverflow(builder.batchSizeHint * builder.maxConcurrentReports); spans = fromSource(buffer) .buffer(forCountOrTime(builder.batchSizeHint, builder.maxBatchDuration, () -> new ListAccumulator(builder.batchSizeHint), client.executionContext().executor())) .filter(accumulate -> !accumulate.isEmpty()) .map(bufferedSpans -> allocator.wrap(spanEncoder.encodeList(bufferedSpans))); } final CompletableSource.Processor spansTerminated = newCompletableProcessor(); toSource(spans.flatMapCompletable(encodedSpansReporter(client, builder.codec), builder.maxConcurrentReports)) .subscribe(spansTerminated); closeable.prepend(toAsyncCloseable(graceful -> { closeInitiated = true; try { buffer.onComplete(); } catch (Throwable t) { LOGGER.error("Failed to dispose request buffer. Ignoring.", t); } return graceful ? fromSource(spansTerminated) : completed(); })); return buffer; } }
HttpReporter extends Component implements Reporter<Span>, AsyncCloseable { private PublisherSource.Processor<Span, Span> initReporter(final Builder builder, final HttpClient client) { final PublisherSource.Processor<Span, Span> buffer; SpanBytesEncoder spanEncoder = builder.codec.spanBytesEncoder(); final BufferAllocator allocator = client.executionContext().bufferAllocator(); final Publisher<Buffer> spans; if (builder.disableBatching) { buffer = newPublisherProcessorDropHeadOnOverflow(builder.maxConcurrentReports); spans = fromSource(buffer).map(span -> allocator.wrap(spanEncoder.encode(span))); } else { buffer = newPublisherProcessorDropHeadOnOverflow(builder.batchSizeHint * builder.maxConcurrentReports); spans = fromSource(buffer) .buffer(forCountOrTime(builder.batchSizeHint, builder.maxBatchDuration, () -> new ListAccumulator(builder.batchSizeHint), client.executionContext().executor())) .filter(accumulate -> !accumulate.isEmpty()) .map(bufferedSpans -> allocator.wrap(spanEncoder.encodeList(bufferedSpans))); } final CompletableSource.Processor spansTerminated = newCompletableProcessor(); toSource(spans.flatMapCompletable(encodedSpansReporter(client, builder.codec), builder.maxConcurrentReports)) .subscribe(spansTerminated); closeable.prepend(toAsyncCloseable(graceful -> { closeInitiated = true; try { buffer.onComplete(); } catch (Throwable t) { LOGGER.error("Failed to dispose request buffer. Ignoring.", t); } return graceful ? fromSource(spansTerminated) : completed(); })); return buffer; } private  HttpReporter(final Builder builder); }
HttpReporter extends Component implements Reporter<Span>, AsyncCloseable { private PublisherSource.Processor<Span, Span> initReporter(final Builder builder, final HttpClient client) { final PublisherSource.Processor<Span, Span> buffer; SpanBytesEncoder spanEncoder = builder.codec.spanBytesEncoder(); final BufferAllocator allocator = client.executionContext().bufferAllocator(); final Publisher<Buffer> spans; if (builder.disableBatching) { buffer = newPublisherProcessorDropHeadOnOverflow(builder.maxConcurrentReports); spans = fromSource(buffer).map(span -> allocator.wrap(spanEncoder.encode(span))); } else { buffer = newPublisherProcessorDropHeadOnOverflow(builder.batchSizeHint * builder.maxConcurrentReports); spans = fromSource(buffer) .buffer(forCountOrTime(builder.batchSizeHint, builder.maxBatchDuration, () -> new ListAccumulator(builder.batchSizeHint), client.executionContext().executor())) .filter(accumulate -> !accumulate.isEmpty()) .map(bufferedSpans -> allocator.wrap(spanEncoder.encodeList(bufferedSpans))); } final CompletableSource.Processor spansTerminated = newCompletableProcessor(); toSource(spans.flatMapCompletable(encodedSpansReporter(client, builder.codec), builder.maxConcurrentReports)) .subscribe(spansTerminated); closeable.prepend(toAsyncCloseable(graceful -> { closeInitiated = true; try { buffer.onComplete(); } catch (Throwable t) { LOGGER.error("Failed to dispose request buffer. Ignoring.", t); } return graceful ? fromSource(spansTerminated) : completed(); })); return buffer; } private  HttpReporter(final Builder builder); @Override CheckResult check(); @Override void report(final Span span); @Override void close(); @Override Completable closeAsync(); @Override Completable closeAsyncGracefully(); }
HttpReporter extends Component implements Reporter<Span>, AsyncCloseable { private PublisherSource.Processor<Span, Span> initReporter(final Builder builder, final HttpClient client) { final PublisherSource.Processor<Span, Span> buffer; SpanBytesEncoder spanEncoder = builder.codec.spanBytesEncoder(); final BufferAllocator allocator = client.executionContext().bufferAllocator(); final Publisher<Buffer> spans; if (builder.disableBatching) { buffer = newPublisherProcessorDropHeadOnOverflow(builder.maxConcurrentReports); spans = fromSource(buffer).map(span -> allocator.wrap(spanEncoder.encode(span))); } else { buffer = newPublisherProcessorDropHeadOnOverflow(builder.batchSizeHint * builder.maxConcurrentReports); spans = fromSource(buffer) .buffer(forCountOrTime(builder.batchSizeHint, builder.maxBatchDuration, () -> new ListAccumulator(builder.batchSizeHint), client.executionContext().executor())) .filter(accumulate -> !accumulate.isEmpty()) .map(bufferedSpans -> allocator.wrap(spanEncoder.encodeList(bufferedSpans))); } final CompletableSource.Processor spansTerminated = newCompletableProcessor(); toSource(spans.flatMapCompletable(encodedSpansReporter(client, builder.codec), builder.maxConcurrentReports)) .subscribe(spansTerminated); closeable.prepend(toAsyncCloseable(graceful -> { closeInitiated = true; try { buffer.onComplete(); } catch (Throwable t) { LOGGER.error("Failed to dispose request buffer. Ignoring.", t); } return graceful ? fromSource(spansTerminated) : completed(); })); return buffer; } private  HttpReporter(final Builder builder); @Override CheckResult check(); @Override void report(final Span span); @Override void close(); @Override Completable closeAsync(); @Override Completable closeAsyncGracefully();  }