ScribeSpanConsumer implements Scribe.AsyncIface { @Override public void Log(List<LogEntry> messages, AsyncMethodCallback<ResultCode> resultHandler) { metrics.incrementMessages(); List<Span> spans = new ArrayList<>(); int byteCount = 0; try { for (LogEntry logEntry : messages) { if (!category.equals(logEntry.category)) continue; byte[] bytes = logEntry.message.getBytes(StandardCharsets.ISO_8859_1); bytes = Base64.getMimeDecoder().decode(bytes); byteCount += bytes.length; spans.add(SpanBytesDecoder.THRIFT.decodeOne(bytes)); } } catch (RuntimeException e) { metrics.incrementMessagesDropped(); resultHandler.onError(e); return; } finally { metrics.incrementBytes(byteCount); } collector.accept(spans, new Callback<Void>() { @Override public void onSuccess(Void value) { resultHandler.onComplete(ResultCode.OK); } @Override public void onError(Throwable t) { Exception error = t instanceof Exception ? (Exception) t : new RuntimeException(t); resultHandler.onError(error); } }, CommonPools.blockingTaskExecutor()); } ScribeSpanConsumer(Collector collector, CollectorMetrics metrics, String category); @Override void Log(List<LogEntry> messages, AsyncMethodCallback<ResultCode> resultHandler);  }
@Test void malformedDataIsDropped() { ScribeSpanConsumer scribe = newScribeSpanConsumer("zipkin", consumer); LogEntry entry = new LogEntry(); entry.category = "zipkin"; entry.message = "notbase64"; CaptureAsyncMethodCallback callback = new CaptureAsyncMethodCallback(); scribe.Log(asList(entry), callback); assertThat(callback.error).isInstanceOf(IllegalArgumentException.class); await().untilAsserted(() -> assertThat(scribeMetrics.messages()).isEqualTo(1)); assertThat(scribeMetrics.messagesDropped()).isEqualTo(1); assertThat(scribeMetrics.bytes()).isZero(); assertThat(scribeMetrics.spans()).isZero(); assertThat(scribeMetrics.spansDropped()).isZero(); }
@Test void consumerExceptionBeforeCallbackDoesntSetFutureException() { consumer = (input) -> { throw new NullPointerException("endpoint was null"); }; ScribeSpanConsumer scribe = newScribeSpanConsumer("zipkin", consumer); LogEntry entry = new LogEntry(); entry.category = "zipkin"; entry.message = encodedSpan; CaptureAsyncMethodCallback callback = new CaptureAsyncMethodCallback(); scribe.Log(asList(entry), callback); assertThat(callback.error).isNull(); await().untilAsserted(() -> assertThat(scribeMetrics.messages()).isEqualTo(1)); assertThat(scribeMetrics.messagesDropped()).isZero(); assertThat(scribeMetrics.bytes()).isEqualTo(bytes.length); assertThat(scribeMetrics.spans()).isEqualTo(1); assertThat(scribeMetrics.spansDropped()).isEqualTo(1); }