Merge pull request #109 from spotify/serializer

Revert sendMetrics

modules/pubsub/src/main/java/com/spotify/ffwd/pubsub/PubsubPluginSink.java

@@ -42,6 +42,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.UUID;
 import java.util.concurrent.Executor;
 import lombok.extern.slf4j.Slf4j;
 
@@ -101,20 +102,32 @@ public void onSuccess(String messageId) { }
     }, executorService);
   }
 
-  // The pubsub plugin only supports sending batches of metrics.
   @Override
   public AsyncFuture<Void> sendMetrics(Collection<Metric> metrics) {
-    log.info("Sending {} metrics", metrics.size());
-
-    int size = 0;
-    try {
-      final ByteString m = ByteString.copyFrom(serializer.serialize(metrics));
-      publishPubSub(m);
-      size += m.size();
-    } catch (Exception e) {
-      log.error("Failed to serialize batch of metrics {}", e);
+    final UUID traceId = UUID.randomUUID();
+    log.info("{}: Start sending metrics", traceId);
+
+    for (Metric metric : metrics) {
+      try {
+        final ApiFuture<String> publish = publisher.publish(PubsubMessage.newBuilder()
+          .setData(ByteString.copyFrom(serializer.serialize(metric)))
+          .build()
+        );
+        ApiFutures.addCallback(publish, new ApiFutureCallback<String>() {
+          @Override
+          public void onFailure(Throwable t) {
+            log.error("Failed sending metrics {}", t.getMessage());
+          }
+
+          @Override
+          public void onSuccess(String messageId) { }
+
+        }, executorService);
+      } catch (Exception e) {
+        log.error("Failed to publish metric {}", e);
+      }
     }
-    log.debug("Total size of sent metrics {} bytes", size);
+    log.info("{}: Finished sending metrics", traceId);
     return async.resolved();
   }