Introduces SpanRecord type and clarifies a PendingSpan concept

Before, we had some elements that weren't well organized about not-yet-
reported spans. Formerly, there was `MutableSpan`, `MutableSpanMap`, and
`Recorder`. These were internal and replaced by still internal types of:

* SpanRecord: mutable type which can become a Zipkin span.
* PendingSpan: SpanRecord and a clock appropriate for a trace context.
* PendingSpans: repository allowing lookup and removal by trace context.

These are clarifications and optimizations of code that existed before.
Notably, this removes the need to re-build a zipkin span to customize it
before reporting, or to need a zipkin span at all when reporting to
metrics or other aggregations. Future work on that will happen in later
pull requests. There is no public api change involved as all work is
currently internal.

brave/src/main/java/brave/RealScopedSpan.java

@@ -1,6 +1,8 @@
 package brave;
 
-import brave.internal.recorder.Recorder;
+import brave.Tracing.SpanReporter;
+import brave.internal.recorder.PendingSpanRecords;
+import brave.internal.recorder.SpanRecord;
 import brave.propagation.CurrentTraceContext.Scope;
 import brave.propagation.TraceContext;
 
@@ -9,13 +11,27 @@ final class RealScopedSpan extends ScopedSpan {
 
   final TraceContext context;
   final Scope scope;
-  final Recorder recorder;
+  final SpanRecord record;
+  final Clock clock;
+  final PendingSpanRecords pendingSpanRecords;
+  final SpanReporter spanReporter;
   final ErrorParser errorParser;
 
-  RealScopedSpan(TraceContext context, Scope scope, Recorder recorder, ErrorParser errorParser) {
+  RealScopedSpan(
+      TraceContext context,
+      Scope scope,
+      SpanRecord record,
+      Clock clock,
+      PendingSpanRecords pendingSpanRecords,
+      SpanReporter spanReporter,
+      ErrorParser errorParser
+  ) {
     this.context = context;
     this.scope = scope;
-    this.recorder = recorder;
+    this.pendingSpanRecords = pendingSpanRecords;
+    this.record = record;
+    this.clock = clock;
+    this.spanReporter = spanReporter;
     this.errorParser = errorParser;
   }
 
@@ -28,12 +44,12 @@ final class RealScopedSpan extends ScopedSpan {
   }
 
   @Override public ScopedSpan annotate(String value) {
-    recorder.annotate(context, value);
+    record.annotate(clock.currentTimeMicroseconds(), value);
     return this;
   }
 
   @Override public ScopedSpan tag(String key, String value) {
-    recorder.tag(context, key, value);
+    record.tag(key, value);
     return this;
   }
 
@@ -44,7 +60,9 @@ final class RealScopedSpan extends ScopedSpan {
 
   @Override public void finish() {
     scope.close();
-    recorder.finish(context);
+    if (!pendingSpanRecords.remove(context)) return; // don't double-report
+    spanReporter.report(context, record);
+    record.finishTimestamp(clock.currentTimeMicroseconds());
   }
 
   @Override public boolean equals(Object o) {

brave/src/main/java/brave/RealSpan.java

@@ -1,21 +1,34 @@
 package brave;
 
-import brave.internal.recorder.Recorder;
+import brave.Tracing.SpanReporter;
+import brave.internal.recorder.PendingSpanRecords;
+import brave.internal.recorder.SpanRecord;
 import brave.propagation.TraceContext;
 import zipkin2.Endpoint;
 
 /** This wraps the public api and guards access to a mutable span. */
 final class RealSpan extends Span {
 
   final TraceContext context;
-  final Recorder recorder;
+  final PendingSpanRecords pendingSpanRecords;
+  final SpanRecord record;
+  final Clock clock;
+  final SpanReporter spanReporter;
   final ErrorParser errorParser;
   final RealSpanCustomizer customizer;
 
-  RealSpan(TraceContext context, Recorder recorder, ErrorParser errorParser) {
+  RealSpan(TraceContext context,
+      PendingSpanRecords pendingSpanRecords,
+      SpanRecord record,
+      Clock clock,
+      SpanReporter spanReporter,
+      ErrorParser errorParser) {
     this.context = context;
-    this.recorder = recorder;
-    this.customizer = new RealSpanCustomizer(context, recorder);
+    this.pendingSpanRecords = pendingSpanRecords;
+    this.record = record;
+    this.clock = clock;
+    this.customizer = new RealSpanCustomizer(context, record, clock);
+    this.spanReporter = spanReporter;
     this.errorParser = errorParser;
   }
 
@@ -28,41 +41,73 @@ final class RealSpan extends Span {
   }
 
   @Override public SpanCustomizer customizer() {
-    return new RealSpanCustomizer(context, recorder);
+    return new RealSpanCustomizer(context, record, clock);
   }
 
   @Override public Span start() {
-    recorder.start(context());
-    return this;
+    return start(clock.currentTimeMicroseconds());
   }
 
   @Override public Span start(long timestamp) {
-    recorder.start(context(), timestamp);
+    synchronized (record) {
+      record.startTimestamp(timestamp);
+    }
     return this;
   }
 
   @Override public Span name(String name) {
-    recorder.name(context(), name);
+    synchronized (record) {
+      record.name(name);
+    }
     return this;
   }
 
   @Override public Span kind(Kind kind) {
-    recorder.kind(context(), kind);
+    synchronized (record) {
+      record.kind(kind);
+    }
     return this;
   }
 
   @Override public Span annotate(String value) {
-    recorder.annotate(context(), value);
-    return this;
+    return annotate(clock.currentTimeMicroseconds(), value);
   }
 
   @Override public Span annotate(long timestamp, String value) {
-    recorder.annotate(context(), timestamp, value);
+    // Modern instrumentation should not send annotations such as this, but we leniently
+    // accept them rather than fail. This for example allows old bridges like to Brave v3 to work
+    if ("cs".equals(value)) {
+      synchronized (record) {
+        record.kind(Span.Kind.CLIENT);
+        record.startTimestamp(timestamp);
+      }
+    } else if ("sr".equals(value)) {
+      synchronized (record) {
+        record.kind(Span.Kind.SERVER);
+        record.startTimestamp(timestamp);
+      }
+    } else if ("cr".equals(value)) {
+      synchronized (record) {
+        record.kind(Span.Kind.CLIENT);
+      }
+      finish(timestamp);
+    } else if ("ss".equals(value)) {
+      synchronized (record) {
+        record.kind(Span.Kind.SERVER);
+      }
+      finish(timestamp);
+    } else {
+      synchronized (record) {
+        record.annotate(timestamp, value);
+      }
+    }
     return this;
   }
 
   @Override public Span tag(String key, String value) {
-    recorder.tag(context(), key, value);
+    synchronized (record) {
+      record.tag(key, value);
+    }
     return this;
   }
 
@@ -72,24 +117,31 @@ final class RealSpan extends Span {
   }
 
   @Override public Span remoteEndpoint(Endpoint remoteEndpoint) {
-    recorder.remoteEndpoint(context(), remoteEndpoint);
+    synchronized (record) {
+      record.remoteEndpoint(remoteEndpoint);
+    }
     return this;
   }
 
   @Override public void finish() {
-    recorder.finish(context());
+    finish(clock.currentTimeMicroseconds());
   }
 
   @Override public void finish(long timestamp) {
-    recorder.finish(context(), timestamp);
+    if (!pendingSpanRecords.remove(context)) return;
+    synchronized (record) {
+      record.finishTimestamp(timestamp);
+    }
+    spanReporter.report(context, record);
   }
 
   @Override public void abandon() {
-    recorder.abandon(context());
+    pendingSpanRecords.remove(context);
   }
 
   @Override public void flush() {
-    recorder.flush(context());
+    abandon();
+    spanReporter.report(context, record);
   }
 
   @Override public String toString() {

brave/src/main/java/brave/RealSpanCustomizer.java

@@ -1,31 +1,40 @@
 package brave;
 
-import brave.internal.recorder.Recorder;
+import brave.internal.recorder.SpanRecord;
 import brave.propagation.TraceContext;
 
 /** This wraps the public api and guards access to a mutable span. */
 final class RealSpanCustomizer implements SpanCustomizer {
 
   final TraceContext context;
-  final Recorder recorder;
+  final SpanRecord record;
+  final Clock clock;
 
-  RealSpanCustomizer(TraceContext context, Recorder recorder) {
+  RealSpanCustomizer(TraceContext context, SpanRecord record, Clock clock) {
     this.context = context;
-    this.recorder = recorder;
+    this.record = record;
+    this.clock = clock;
   }
 
   @Override public SpanCustomizer name(String name) {
-    recorder.name(context, name);
+    synchronized (record) {
+      record.name(name);
+    }
     return this;
   }
 
   @Override public SpanCustomizer annotate(String value) {
-    recorder.annotate(context, value);
+    long timestamp = clock.currentTimeMicroseconds();
+    synchronized (record) {
+      record.annotate(timestamp, value);
+    }
     return this;
   }
 
   @Override public SpanCustomizer tag(String key, String value) {
-    recorder.tag(context, key, value);
+    synchronized (record) {
+      record.tag(key, value);
+    }
     return this;
   }