Decouples v2 types from v1 types, in preparation of a new package (op…

…enzipkin#1726)

This moves all the code needed for v2 types into zipkin.internal.v2
without any references to code outside the package. To reduce
duplication, "v1" code can reference v2 types (as they need to in order
to covert for example). Once this is in, we can consider refactoring out
a v2 module.

benchmarks/src/main/java/zipkin/benchmarks/CodecBenchmarks.java

@@ -41,8 +41,7 @@
 import zipkin.Codec;
 import zipkin.Endpoint;
 import zipkin.internal.v2.Span;
-import zipkin.internal.v2.codec.BytesDecoder;
-import zipkin.internal.v2.codec.BytesEncoder;
+import zipkin.internal.v2.codec.SpanBytesCodec;
 
 /**
  * This compares the speed of the bundled java codec with the approach used in the scala
@@ -157,28 +156,28 @@ public byte[] writeClientSpan_thrift_libthrift() throws TException {
   }
 
   static final byte[] span2Json = read("/span2.json");
-  static final Span span2 = BytesDecoder.JSON.decode(span2Json);
+  static final Span span2 = SpanBytesCodec.JSON.decode(span2Json);
   static final List<Span> tenSpan2s = Collections.nCopies(10, span2);
-  static final byte[] tenSpan2sJson = BytesEncoder.JSON.encodeList(tenSpan2s);
+  static final byte[] tenSpan2sJson = SpanBytesCodec.JSON.encodeList(tenSpan2s);
 
   @Benchmark
   public Span readClientSpan_json_span2() {
-    return BytesDecoder.JSON.decode(span2Json);
+    return SpanBytesCodec.JSON.decode(span2Json);
   }
 
   @Benchmark
   public List<Span> readTenClientSpans_json_span2() {
-    return BytesDecoder.JSON.decodeList(tenSpan2sJson);
+    return SpanBytesCodec.JSON.decodeList(tenSpan2sJson);
   }
 
   @Benchmark
   public byte[] writeClientSpan_json_span2() {
-    return BytesEncoder.JSON.encode(span2);
+    return SpanBytesCodec.JSON.encode(span2);
   }
 
   @Benchmark
   public byte[] writeTenClientSpans_json_span2() {
-    return BytesEncoder.JSON.encodeList(tenSpan2s);
+    return SpanBytesCodec.JSON.encodeList(tenSpan2s);
   }
 
   static final byte[] rpcSpanJson = read("/span-rpc.json");

benchmarks/src/main/java/zipkin/benchmarks/Span2ConverterBenchmarks.java

@@ -38,7 +38,7 @@
 import zipkin.internal.V2SpanConverter;
 import zipkin.internal.Util;
 
-import static zipkin.internal.V2SpanConverter.convert;
+import static zipkin.internal.V2SpanConverter.toEndpoint;
 
 @Measurement(iterations = 5, time = 1)
 @Warmup(iterations = 10, time = 1)
@@ -97,8 +97,8 @@ public class Span2ConverterBenchmarks {
     .name("get")
     .kind(Span.Kind.SERVER)
     .shared(true)
-    .localEndpoint(convert(backend))
-    .remoteEndpoint(convert(frontend))
+    .localEndpoint(toEndpoint(backend))
+    .remoteEndpoint(toEndpoint(frontend))
     .timestamp(1472470996250000L)
     .duration(100000L)
     .putTag(TraceKeys.HTTP_PATH, "/backend")

pom.xml

@@ -323,6 +323,12 @@
         <version>${okio.version}</version>
       </dependency>
 
+      <dependency>
+        <groupId>com.squareup.moshi</groupId>
+        <artifactId>moshi</artifactId>
+        <version>1.5.0</version>
+      </dependency>
+
       <dependency>
         <groupId>com.google.auto.value</groupId>
         <artifactId>auto-value</artifactId>

zipkin-collector/kafka/src/test/java/zipkin/collector/kafka/KafkaCollectorTest.java

@@ -32,7 +32,7 @@
 import zipkin.collector.kafka.KafkaCollector.Builder;
 import zipkin.internal.ApplyTimestampAndDuration;
 import zipkin.internal.V2SpanConverter;
-import zipkin.internal.v2.codec.BytesEncoder;
+import zipkin.internal.v2.codec.SpanBytesCodec;
 import zipkin.storage.AsyncSpanConsumer;
 import zipkin.storage.AsyncSpanStore;
 import zipkin.storage.SpanStore;
@@ -146,7 +146,7 @@ public void messageWithMultipleSpans_json2() throws Exception {
       ApplyTimestampAndDuration.apply(LOTS_OF_SPANS[1])
     );
 
-    byte[] message = BytesEncoder.JSON.encodeList(asList(
+    byte[] message = SpanBytesCodec.JSON.encodeList(asList(
       V2SpanConverter.fromSpan(spans.get(0)).get(0),
       V2SpanConverter.fromSpan(spans.get(1)).get(0)
     ));

zipkin-collector/kafka10/src/test/java/zipkin/collector/kafka10/KafkaCollectorTest.java

@@ -39,7 +39,7 @@
 import zipkin.collector.kafka10.KafkaCollector.Builder;
 import zipkin.internal.ApplyTimestampAndDuration;
 import zipkin.internal.V2SpanConverter;
-import zipkin.internal.v2.codec.BytesEncoder;
+import zipkin.internal.v2.codec.SpanBytesCodec;
 import zipkin.storage.AsyncSpanConsumer;
 import zipkin.storage.AsyncSpanStore;
 import zipkin.storage.SpanStore;
@@ -200,7 +200,7 @@ public void messageWithMultipleSpans_json2() throws Exception {
       ApplyTimestampAndDuration.apply(LOTS_OF_SPANS[1])
     );
 
-    byte[] message = BytesEncoder.JSON.encodeList(asList(
+    byte[] message = SpanBytesCodec.JSON.encodeList(asList(
       V2SpanConverter.fromSpan(spans.get(0)).get(0),
       V2SpanConverter.fromSpan(spans.get(1)).get(0)
     ));

zipkin-junit/pom.xml

@@ -42,6 +42,11 @@
       <artifactId>mockwebserver</artifactId>
     </dependency>
 
+    <dependency>
+      <groupId>com.squareup.moshi</groupId>
+      <artifactId>moshi</artifactId>
+    </dependency>
+
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>

zipkin-junit/src/main/java/zipkin/junit/ZipkinDispatcher.java

@@ -31,7 +31,8 @@
 import zipkin.collector.CollectorMetrics;
 import zipkin.internal.V2JsonSpanDecoder;
 import zipkin.internal.V2StorageComponent;
-import zipkin.internal.v2.codec.BytesEncoder;
+import zipkin.internal.v2.codec.DependencyLinkBytesCodec;
+import zipkin.internal.v2.codec.SpanBytesCodec;
 import zipkin.internal.v2.internal.Platform;
 import zipkin.storage.Callback;
 import zipkin.storage.QueryRequest;
@@ -124,20 +125,18 @@ MockResponse queryV2(HttpUrl url) throws IOException {
     } else if (url.encodedPath().equals("/api/v2/dependencies")) {
       Long endTs = maybeLong(url.queryParameter("endTs"));
       Long lookback = maybeLong(url.queryParameter("lookback"));
-      List<DependencyLink> result = store2.getDependencies(
+      List<zipkin.internal.v2.DependencyLink> result = store2.getDependencies(
         endTs != null ? endTs : System.currentTimeMillis(),
         lookback != null ? lookback : DEFAULT_LOOKBACK
       ).execute();
-      return jsonResponse(Codec.JSON.writeDependencyLinks(result));
+      return jsonResponse(DependencyLinkBytesCodec.JSON.encodeList(result));
     } else if (url.encodedPath().equals("/api/v2/traces")) {
       List<List<zipkin.internal.v2.Span>> traces = store2.getTraces(toQueryRequest2(url)).execute();
-      return jsonResponse(BytesEncoder.JSON.encodeNestedList(traces));
+      return jsonResponse(SpanBytesCodec.JSON.encodeNestedList(traces));
     } else if (url.encodedPath().startsWith("/api/v2/trace/")) {
       String traceIdHex = url.encodedPath().replace("/api/v2/trace/", "");
       List<zipkin.internal.v2.Span> trace = store2.getTrace(normalizeTraceId(traceIdHex)).execute();
-      if (!trace.isEmpty()) {
-        return jsonResponse(BytesEncoder.JSON.encodeList(trace));
-      }
+      if (!trace.isEmpty()) return jsonResponse(SpanBytesCodec.JSON.encodeList(trace));
     }
     return new MockResponse().setResponseCode(404);
   }

zipkin-junit/src/test/java/zipkin/junit/ZipkinRuleTest.java

@@ -32,7 +32,7 @@
 import zipkin.Span;
 import zipkin.internal.ApplyTimestampAndDuration;
 import zipkin.internal.V2SpanConverter;
-import zipkin.internal.v2.codec.BytesEncoder;
+import zipkin.internal.v2.codec.SpanBytesCodec;
 
 import static java.lang.String.format;
 import static java.util.Arrays.asList;
@@ -68,7 +68,7 @@ public void getTraces_storedViaPostVersion2() throws IOException {
       ApplyTimestampAndDuration.apply(LOTS_OF_SPANS[1])
     );
 
-    byte[] message = BytesEncoder.JSON.encodeList(asList(
+    byte[] message = SpanBytesCodec.JSON.encodeList(asList(
       V2SpanConverter.fromSpan(spans.get(0)).get(0),
       V2SpanConverter.fromSpan(spans.get(1)).get(0)
     ));

zipkin-junit/src/test/java/zipkin/junit/v2/HttpV2SpanConsumer.java

@@ -19,9 +19,8 @@
 import okhttp3.OkHttpClient;
 import okhttp3.Request;
 import okhttp3.RequestBody;
-import okio.Buffer;
 import zipkin.internal.v2.Span;
-import zipkin.internal.v2.codec.BytesEncoder;
+import zipkin.internal.v2.codec.SpanBytesCodec;
 import zipkin.internal.v2.storage.SpanConsumer;
 
 /** Implements the span consumer interface by forwarding requests over http. */
@@ -33,16 +32,10 @@ final class HttpV2SpanConsumer implements SpanConsumer {
   }
 
   @Override public zipkin.internal.v2.Call<Void> accept(List<Span> spans) {
-    Buffer json = new Buffer();
-    json.writeByte('[');
-    for (int i = 0, length = spans.size(); i < length; ) {
-      json.write(BytesEncoder.JSON.encode(spans.get(i)));
-      if (++i < length) json.writeByte(',');
-    }
-    json.writeByte(']');
+    byte[] json = SpanBytesCodec.JSON.encodeList(spans);
     return factory.newCall(new Request.Builder()
         .url(factory.baseUrl.resolve("/api/v2/spans"))
-        .post(RequestBody.create(MediaType.parse("application/json"), json.readByteArray())).build(),
+        .post(RequestBody.create(MediaType.parse("application/json"), json)).build(),
       b -> null /* void */
     );
   }

zipkin-junit/src/test/java/zipkin/junit/v2/HttpV2SpanStore.java

@@ -13,22 +13,28 @@
  */
 package zipkin.junit.v2;
 
+import com.squareup.moshi.JsonAdapter;
+import com.squareup.moshi.Moshi;
+import com.squareup.moshi.Types;
 import java.util.Collections;
 import java.util.List;
 import javax.annotation.Nullable;
 import okhttp3.HttpUrl;
 import okhttp3.OkHttpClient;
 import okhttp3.Request;
-import zipkin.Codec;
-import zipkin.DependencyLink;
 import zipkin.internal.v2.Call;
+import zipkin.internal.v2.DependencyLink;
 import zipkin.internal.v2.Span;
-import zipkin.internal.v2.codec.BytesDecoder;
+import zipkin.internal.v2.codec.DependencyLinkBytesCodec;
+import zipkin.internal.v2.codec.SpanBytesCodec;
 import zipkin.internal.v2.storage.QueryRequest;
 import zipkin.internal.v2.storage.SpanStore;
 
 /** Implements the span store interface by forwarding requests over http. */
 final class HttpV2SpanStore implements SpanStore {
+  static final JsonAdapter<List<String>> STRING_LIST_ADAPTER =
+    new Moshi.Builder().build().adapter(Types.newParameterizedType(List.class, String.class));
+
   final HttpV2Call.Factory factory;
 
   HttpV2SpanStore(OkHttpClient client, HttpUrl baseUrl) {
@@ -46,13 +52,13 @@ final class HttpV2SpanStore implements SpanStore {
     maybeAddQueryParam(url, "lookback", request.lookback());
     maybeAddQueryParam(url, "limit", request.limit());
     return factory.newCall(new Request.Builder().url(url.build()).build(),
-      content -> BytesDecoder.JSON.decodeNestedList(content.readByteArray()));
+      content -> SpanBytesCodec.JSON.decodeNestedList(content.readByteArray()));
   }
 
   @Override public Call<List<Span>> getTrace(String traceId) {
     return factory.newCall(new Request.Builder()
       .url(factory.baseUrl.resolve("/api/v2/trace/" + Span.normalizeTraceId(traceId)))
-      .build(), content -> BytesDecoder.JSON.decodeList(content.readByteArray()))
+      .build(), content -> SpanBytesCodec.JSON.decodeList(content.readByteArray()))
       .handleError(((error, callback) -> {
         if (error instanceof HttpException && ((HttpException) error).code == 404) {
           callback.onSuccess(Collections.emptyList());
@@ -66,20 +72,20 @@ final class HttpV2SpanStore implements SpanStore {
   public Call<List<String>> getServiceNames() {
     return factory.newCall(new Request.Builder()
       .url(factory.baseUrl.resolve("/api/v2/services"))
-      .build(), content -> Codec.JSON.readStrings(content.readByteArray()));
+      .build(), STRING_LIST_ADAPTER::fromJson);
   }
 
   @Override
   public Call<List<String>> getSpanNames(String serviceName) {
     return factory.newCall(new Request.Builder()
       .url(factory.baseUrl.resolve("/api/v2/spans?serviceName=" + serviceName))
-      .build(), content -> Codec.JSON.readStrings(content.readByteArray()));
+      .build(), STRING_LIST_ADAPTER::fromJson);
   }
 
   @Override public Call<List<DependencyLink>> getDependencies(long endTs, long lookback) {
     return factory.newCall(new Request.Builder()
       .url(factory.baseUrl.resolve("/api/v2/dependencies?endTs=" + endTs + "&lookback=" + lookback))
-      .build(), content -> Codec.JSON.readDependencyLinks(content.readByteArray()));
+      .build(), content -> DependencyLinkBytesCodec.JSON.decodeList(content.readByteArray()));
   }
 
   void maybeAddQueryParam(HttpUrl.Builder builder, String name, @Nullable Object value) {

zipkin-server/src/main/java/zipkin/server/ZipkinQueryApiV2.java

@@ -32,12 +32,12 @@
 import org.springframework.web.bind.annotation.ResponseStatus;
 import org.springframework.web.bind.annotation.RestController;
 import org.springframework.web.context.request.WebRequest;
-import zipkin.Codec;
-import zipkin.DependencyLink;
 import zipkin.internal.V2StorageComponent;
 import zipkin.internal.v2.Call;
+import zipkin.internal.v2.DependencyLink;
 import zipkin.internal.v2.Span;
-import zipkin.internal.v2.codec.BytesEncoder;
+import zipkin.internal.v2.codec.DependencyLinkBytesCodec;
+import zipkin.internal.v2.codec.SpanBytesCodec;
 import zipkin.internal.v2.storage.QueryRequest;
 import zipkin.storage.StorageComponent;
 
@@ -83,7 +83,7 @@ public byte[] getDependencies(
 
     Call<List<DependencyLink>> call = storage.v2SpanStore()
       .getDependencies(endTs, lookback != null ? lookback : defaultLookback);
-    return Codec.JSON.writeDependencyLinks(call.execute());
+    return DependencyLinkBytesCodec.JSON.encodeList(call.execute());
   }
 
   @RequestMapping(value = "/services", method = RequestMethod.GET)
@@ -128,7 +128,7 @@ public String getTraces(
       .limit(limit).build();
 
     List<List<Span>> traces = storage.v2SpanStore().getTraces(queryRequest).execute();
-    return new String(BytesEncoder.JSON.encodeNestedList(traces), UTF_8);
+    return new String(SpanBytesCodec.JSON.encodeNestedList(traces), UTF_8);
   }
 
   @RequestMapping(value = "/trace/{traceIdHex}", method = RequestMethod.GET, produces = APPLICATION_JSON_VALUE)
@@ -137,7 +137,7 @@ public String getTrace(@PathVariable String traceIdHex, WebRequest request) thro
 
     List<Span> trace = storage.v2SpanStore().getTrace(traceIdHex).execute();
     if (trace.isEmpty()) throw new TraceNotFoundException(traceIdHex);
-    return new String(BytesEncoder.JSON.encodeList(trace), UTF_8);
+    return new String(SpanBytesCodec.JSON.encodeList(trace), UTF_8);
   }
 
   @ExceptionHandler(Version2StorageNotConfigured.class)

zipkin-server/src/test/java/zipkin/server/ZipkinServerIntegrationTest.java

@@ -38,7 +38,7 @@
 import zipkin.internal.ApplyTimestampAndDuration;
 import zipkin.internal.V2InMemoryStorage;
 import zipkin.internal.V2SpanConverter;
-import zipkin.internal.v2.codec.BytesEncoder;
+import zipkin.internal.v2.codec.SpanBytesCodec;
 
 import static java.lang.String.format;
 import static java.util.Arrays.asList;
@@ -93,7 +93,7 @@ public void writeSpans_noContentTypeIsJson() throws Exception {
   public void writeSpans_version2() throws Exception {
     Span span = ApplyTimestampAndDuration.apply(LOTS_OF_SPANS[0]);
 
-    byte[] message = BytesEncoder.JSON.encodeList(asList(
+    byte[] message = SpanBytesCodec.JSON.encodeList(asList(
       V2SpanConverter.fromSpan(span).get(0)
     ));
 

zipkin-storage/elasticsearch-http/pom.xml

@@ -49,7 +49,6 @@
     <dependency>
       <groupId>com.squareup.moshi</groupId>
       <artifactId>moshi</artifactId>
-      <version>1.5.0</version>
     </dependency>
 
     <dependency>

zipkin-storage/elasticsearch-http/src/main/java/zipkin/moshi/JsonReaders.java

@@ -16,7 +16,9 @@
 import com.squareup.moshi.JsonReader;
 import java.io.EOFException;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.LinkedHashSet;
+import java.util.List;
 import java.util.Set;
 import javax.annotation.Nullable;
 
@@ -69,10 +71,10 @@ public static JsonReader enterPath(JsonReader reader, String path) throws IOExce
     return null;
   }
 
-  public static Set<String> collectValuesNamed(JsonReader reader, String name) throws IOException {
+  public static List<String> collectValuesNamed(JsonReader reader, String name) throws IOException {
     Set<String> result = new LinkedHashSet<>();
     visitObject(reader, name, result);
-    return result;
+    return new ArrayList<>(result);
   }
 
   static void visitObject(JsonReader reader, String name, Set<String> result) throws IOException {