Add distribution support to FastForward client

ffwd-client/pom.xml

@@ -32,6 +32,21 @@
             <artifactId>protoc</artifactId>
             <type>pom</type>
         </dependency>
+        <!-- Test Dependencies -->
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-api</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-engine</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-junit-jupiter</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
-
 </project>

ffwd-client/src/main/java/com/spotify/ffwd/FastForward.java

@@ -1,8 +1,8 @@
 /*-
  * -\-\-
- * FastForward Client
+ * FastForward Java Client
  * --
- * Copyright (C) 2016 - 2019 Spotify AB
+ * Copyright (C) 2016 - 2020 Spotify AB
  * --
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -31,10 +31,31 @@
 
 public class FastForward {
 
+  /**
+   * If you are using an older version, please use Version.V0.
+   * This variable will eventually be removed.
+   */
+  @Deprecated()
   public static final int LATEST_VERSION = 0;
+
   public static final String DEFAULT_HOST = "localhost";
   public static final int DEFAULT_PORT = 19091;
 
+  public enum Version {
+    V0(0),
+    V1(1);
+    private final int version;
+
+    Version(final int version) {
+      this.version = version;
+    }
+
+    public int getVersion() {
+      return version;
+    }
+
+  }
+
 
   public static FastForward setup() throws UnknownHostException, SocketException {
     return setup(InetAddress.getByName(DEFAULT_HOST), DEFAULT_PORT);
@@ -57,6 +78,7 @@ public static FastForward setup(InetAddress addr) throws SocketException {
    * Initialization method for a FastForward client.
    *
    * @return A new instance of a FastForward client.
+   *
    * @throws SocketException If a datagram socket cannot be created.
    */
   public static FastForward setup(InetAddress addr, int port) throws SocketException {
@@ -74,15 +96,26 @@ private FastForward(InetAddress addr, int port, DatagramSocket socket) {
     this.socket = socket;
   }
 
+  protected FastForward() throws UnknownHostException, SocketException {
+    this.addr = InetAddress.getByName(DEFAULT_HOST);
+    this.port = DEFAULT_PORT;
+    this.socket = new DatagramSocket();
+  }
+
+
   public void send(Metric metric) throws IOException {
-    sendFrame(metric.serialize());
+    sendFrame(metric.serialize(), Version.V0);
   }
 
-  private void sendFrame(byte[] bytes) throws IOException {
+  public void send(com.spotify.ffwd.v1.Metric metric) throws IOException {
+    sendFrame(metric.serialize(), Version.V1);
+  }
+
+  void sendFrame(byte[] bytes, Version v) throws IOException {
     final ByteBuffer buffer = ByteBuffer.allocate(bytes.length + 8);
     buffer.order(ByteOrder.BIG_ENDIAN);
 
-    buffer.putInt(LATEST_VERSION);
+    buffer.putInt(v.getVersion());
     buffer.putInt(buffer.capacity());
     buffer.put(bytes);
     buffer.rewind();
@@ -100,4 +133,8 @@ public static Metric metric(String key) {
     return new Metric().key(key);
   }
 
+  public static com.spotify.ffwd.v1.Metric metricV1(String key) {
+    return new com.spotify.ffwd.v1.Metric().key(key);
+  }
+
 }

ffwd-client/src/main/java/com/spotify/ffwd/v1/Metric.java

@@ -18,24 +18,6 @@
  * -/-/-
  */
 
-/*
- * FastForward Client
- *   --
- *   Copyright (C) 2016 - 2019 Spotify AB
- *   --
- *   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 com.spotify.ffwd.v1;
 
 import com.google.protobuf.ByteString;
@@ -67,7 +49,7 @@ public Metric() {
     this.has = 0;
     this.time = 0;
     this.key = null;
-    this.value = null;
+    this.value = Value.doubleValue(0);
     this.host = null;
     this.tags = new ArrayList<>();
     this.attributes = new HashMap<>();
@@ -80,7 +62,7 @@ public Metric(
     this.has = has;
     this.time = time;
     this.key = key;
-    this.value = value;
+    this.value = (value == null) ? Value.doubleValue(0) : value;
     this.host = host;
     this.tags = tags;
     this.attributes = attributes;
@@ -103,7 +85,8 @@ public Metric key(String key) {
   }
 
   public Metric value(Value value) {
-    return new Metric(set(VALUE), time, key, value, host, tags, attributes);
+    Value newValue = (value == null) ? Value.doubleValue(0) : value;
+    return new Metric(set(VALUE), time, key, newValue, host, tags, attributes);
   }
 
   public Metric host(String host) {
@@ -149,10 +132,12 @@ public byte[] serialize() {
         builder.setValue(Protocol1.Value.newBuilder().setDoubleValue(doubleValue.getValue()));
       } else if (value instanceof Value.DistributionValue) {
         Value.DistributionValue distributionValue = (Value.DistributionValue) value;
-        ByteString byteString = ByteString.copyFrom(distributionValue.getValue());
+        ByteString byteString = ByteString.copyFrom(distributionValue.getValue().array());
         builder.setValue(Protocol1.Value.newBuilder().setDistributionValue(byteString));
+      } else {
+        throw new IllegalArgumentException("Failed to identify distribution type : [" + value
+                                          + "]");
       }
-
     }
 
     if (test(HOST)) {

ffwd-client/src/main/java/com/spotify/ffwd/v1/Value.java

@@ -21,12 +21,16 @@
 package com.spotify.ffwd.v1;
 
 import com.google.auto.value.AutoValue;
-
 import java.nio.ByteBuffer;
 import java.util.function.Function;
 
 
-
+/**
+ * The actual value of the data point.
+ * Currently we support two value types:
+ * Double and Distribution.
+ *
+ */
 public abstract class Value {
 
   Value() {}

ffwd-client/src/main/proto/protocol1.proto

@@ -37,6 +37,7 @@ message Metric {
 
 }
 
+// Actual data point value. We currently support 2 types, distribution and double.
 message Value {
     oneof value {
         double double_value = 1;

ffwd-client/src/test/java/com/spotify/ffwd/FastForwardTest.java

@@ -0,0 +1,93 @@
+/*-
+ * -\-\-
+ * FastForward Java Client
+ * --
+ * Copyright (C) 2016 - 2020 Spotify AB
+ * --
+ * 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 com.spotify.ffwd;
+
+import com.spotify.ffwd.v1.Metric;
+import java.io.IOException;
+import java.net.SocketException;
+import java.net.UnknownHostException;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+import org.mockito.Spy;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.times;
+
+
+public class FastForwardTest {
+   private static final String KEY = "key";
+
+
+   @Spy
+   private FastForward client;
+
+   private ArgumentCaptor<byte []> inputCaptor;
+   private ArgumentCaptor<FastForward.Version> versionCaptor;
+
+   @BeforeEach
+   public void before() throws IOException{
+     MockitoAnnotations.initMocks(this);
+     inputCaptor = ArgumentCaptor.forClass(byte[].class);
+     versionCaptor = ArgumentCaptor.forClass(FastForward.Version.class);
+     Mockito.doNothing().when(client).sendFrame(inputCaptor.capture(), versionCaptor.capture());
+   }
+
+  @Test
+  public void testSendV1() throws SocketException, UnknownHostException , IOException {
+    final  Metric metricV1 = FastForward.metricV1(KEY);
+    client.send(metricV1);
+    Mockito.verify(client,times(1)).sendFrame(Mockito.any(byte[].class) ,
+        Mockito.any(FastForward.Version.class));
+    assertArrayEquals(metricV1.serialize(), inputCaptor.getValue());
+    assertEquals(FastForward.Version.V1,versionCaptor.getValue());
+  }
+
+   @Test
+   public void testSendV0() throws SocketException, UnknownHostException , IOException {
+     final  com.spotify.ffwd.Metric metricV0 = FastForward.metric(KEY);
+     client.send(metricV0);
+     Mockito.verify(client,times(1)).sendFrame(Mockito.any(byte[].class) ,
+         Mockito.any(FastForward.Version.class));
+     assertArrayEquals(metricV0.serialize(), inputCaptor.getValue());
+     assertEquals(FastForward.Version.V0,versionCaptor.getValue());
+   }
+
+   @Test
+   public void testMetricV0(){
+     com.spotify.ffwd.Metric metricV0 = FastForward.metric(KEY);
+     assertEquals(com.spotify.ffwd.Metric.class, metricV0.getClass());
+     assertEquals(KEY,metricV0.getKey());
+   }
+
+   @Test
+   public void testMetricV1(){
+     Metric metricV1 = FastForward.metricV1(KEY);
+     assertEquals(Metric.class, metricV1.getClass());
+     assertEquals(KEY,metricV1.getKey());
+   }
+
+
+}