Add tests for XrayUDPStorage / Reporter.

pom.xml

@@ -54,6 +54,10 @@
     <!-- This allows you to test feature branches with jitpack -->
     <armeria.groupId>com.linecorp.armeria</armeria.groupId>
     <armeria.version>0.98.3</armeria.version>
+
+    <!-- Generally best to match this with armeria's dependency -->
+    <netty.version>4.1.46.Final</netty.version>
+
     <zipkin.version>2.20.0</zipkin.version>
     <zipkin-reporter.version>2.12.1</zipkin-reporter.version>
     <brave.version>5.9.1</brave.version>
@@ -132,6 +136,13 @@
         <type>pom</type>
         <scope>import</scope>
       </dependency>
+      <dependency>
+        <groupId>io.netty</groupId>
+        <artifactId>netty-bom</artifactId>
+        <version>${netty.version}</version>
+        <type>pom</type>
+        <scope>import</scope>
+      </dependency>
       <dependency>
         <groupId>io.zipkin.zipkin2</groupId>
         <artifactId>zipkin</artifactId>

reporter-xray-udp/pom.xml

@@ -39,5 +39,15 @@
       <groupId>io.zipkin.reporter2</groupId>
       <artifactId>zipkin-reporter</artifactId>
     </dependency>
+    <dependency>
+      <groupId>io.netty</groupId>
+      <artifactId>netty-transport</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.zipkin.zipkin2</groupId>
+      <artifactId>zipkin-tests</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 </project>

reporter-xray-udp/src/test/java/zipkin2/reporter/xray_udp/XRayUDPReporterTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 The OpenZipkin Authors
+ * Copyright 2016-2020 The OpenZipkin Authors
  *
  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
  * in compliance with the License. You may obtain a copy of the License at
@@ -13,4 +13,79 @@
  */
 package zipkin2.reporter.xray_udp;
 
-public class XRayUDPReporterTest {}
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBufUtil;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.DatagramPacket;
+import io.netty.channel.socket.nio.NioDatagramChannel;
+import java.net.InetSocketAddress;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import zipkin2.TestObjects;
+import zipkin2.storage.xray_udp.InternalStorageAccess;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class XRayUDPReporterTest {
+
+  private static EventLoopGroup workerGroup;
+  private static Channel serverChannel;
+  private static BlockingQueue<byte[]> receivedPayloads;
+
+  private static XRayUDPReporter reporter;
+
+  @BeforeClass
+  public static void startServer() {
+    workerGroup = new NioEventLoopGroup();
+    receivedPayloads = new LinkedBlockingQueue<>();
+
+    Bootstrap bootstrap = new Bootstrap();
+    bootstrap.group(workerGroup)
+        .channel(NioDatagramChannel.class)
+        .handler(new ChannelInitializer<NioDatagramChannel>() {
+          @Override protected void initChannel(NioDatagramChannel channel) {
+            channel.pipeline().addLast(new SimpleChannelInboundHandler<DatagramPacket>() {
+              @Override protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket msg) {
+                receivedPayloads.add(ByteBufUtil.getBytes(msg.content()));
+              }
+            });
+          }
+        });
+
+    serverChannel = bootstrap.bind(0).syncUninterruptibly().channel();
+
+    // TODO(anuraaga): Consider changing return type of create to XRayUdpReporter so it's more
+    // obvious the type is Closeable.
+    reporter = (XRayUDPReporter) XRayUDPReporter
+        .create("localhost:" + ((InetSocketAddress) serverChannel.localAddress()).getPort());
+  }
+
+  @AfterClass
+  public static void stopServer() {
+    reporter.close();
+    serverChannel.close().syncUninterruptibly();
+    workerGroup.shutdownGracefully();
+  }
+
+  @Before
+  public void setUp() {
+    receivedPayloads.clear();
+  }
+
+  @Test
+  public void sendSingleSpan() throws Exception {
+    reporter.report(TestObjects.CLIENT_SPAN);
+    assertThat(receivedPayloads.take())
+        .containsExactly(InternalStorageAccess.encode(TestObjects.CLIENT_SPAN));
+    assertThat(receivedPayloads).isEmpty();
+  }
+}

reporter-xray-udp/src/test/java/zipkin2/storage/xray_udp/InternalStorageAccess.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2016-2020 The OpenZipkin Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package zipkin2.storage.xray_udp;
+
+import zipkin2.Span;
+
+/**
+ * Classpath accessor to expose package-private methods in the storage.
+ */
+public final class InternalStorageAccess {
+  /** Encode a span for XRay, */
+  public static byte[] encode(Span span) {
+    return UDPMessageEncoder.encode(span);
+  }
+}

storage-xray-udp/pom.xml

@@ -46,5 +46,15 @@
       <version>2.4.0</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>io.netty</groupId>
+      <artifactId>netty-transport</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.zipkin.zipkin2</groupId>
+      <artifactId>zipkin-tests</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 </project>

storage-xray-udp/src/test/java/zipkin2/storage/xray_udp/XRayUDPStorageTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 The OpenZipkin Authors
+ * Copyright 2016-2020 The OpenZipkin Authors
  *
  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
  * in compliance with the License. You may obtain a copy of the License at
@@ -13,4 +13,108 @@
  */
 package zipkin2.storage.xray_udp;
 
-public class XRayUDPStorageTest {}
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBufUtil;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.DatagramPacket;
+import io.netty.channel.socket.nio.NioDatagramChannel;
+import java.net.InetSocketAddress;
+import java.util.Collections;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import zipkin2.Span;
+import zipkin2.TestObjects;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+public class XRayUDPStorageTest {
+
+  private static EventLoopGroup workerGroup;
+  private static Channel serverChannel;
+  private static BlockingQueue<byte[]> receivedPayloads;
+
+  private static XRayUDPStorage storage;
+
+  @BeforeClass
+  public static void startServer() {
+    workerGroup = new NioEventLoopGroup();
+    receivedPayloads = new LinkedBlockingQueue<>();
+
+    Bootstrap bootstrap = new Bootstrap();
+    bootstrap.group(workerGroup)
+        .channel(NioDatagramChannel.class)
+        .handler(new ChannelInitializer<NioDatagramChannel>() {
+          @Override protected void initChannel(NioDatagramChannel channel) {
+            channel.pipeline().addLast(new SimpleChannelInboundHandler<DatagramPacket>() {
+              @Override protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket msg) {
+                receivedPayloads.add(ByteBufUtil.getBytes(msg.content()));
+              }
+            });
+          }
+        });
+
+    serverChannel = bootstrap.bind(0).syncUninterruptibly().channel();
+
+    storage = XRayUDPStorage.newBuilder()
+        .address("localhost:" + ((InetSocketAddress) serverChannel.localAddress()).getPort())
+        .build();
+  }
+
+  @AfterClass
+  public static void stopServer() {
+    storage.close();
+    serverChannel.close().syncUninterruptibly();
+    workerGroup.shutdownGracefully();
+  }
+
+  @Before
+  public void setUp() {
+    receivedPayloads.clear();
+  }
+
+  @Test
+  public void sendTrace() throws Exception {
+    storage.accept(TestObjects.TRACE).execute();
+    for (Span span : TestObjects.TRACE) {
+      byte[] received = receivedPayloads.take();
+      assertThat(received).containsExactly(UDPMessageEncoder.encode(span));
+    }
+    assertThat(receivedPayloads).isEmpty();
+  }
+
+  @Test
+  public void sendSingleSpan() throws Exception {
+    storage.accept(Collections.singletonList(TestObjects.CLIENT_SPAN)).execute();
+    assertThat(receivedPayloads.take())
+        .containsExactly(UDPMessageEncoder.encode(TestObjects.CLIENT_SPAN));
+    assertThat(receivedPayloads).isEmpty();
+  }
+
+  @Test
+  public void sendNoSpans() throws Exception {
+    storage.accept(Collections.emptyList()).execute();
+    // Give some time for any potential bugs to get sent to the server.
+    Thread.sleep(100);
+    assertThat(receivedPayloads).isEmpty();
+  }
+
+  @Test
+  public void sendAfterClose() throws Exception {
+    XRayUDPStorage storage = XRayUDPStorage.newBuilder()
+        .address("localhost:" + ((InetSocketAddress)serverChannel.localAddress()).getPort())
+        .build();
+    storage.close();
+    assertThatThrownBy(() -> storage.accept(TestObjects.TRACE))
+        .isInstanceOf(IllegalStateException.class);
+  }
+}