/*-
 * -\-\-
 * FastForward Pubsub Module
 * --
 * Copyright (C) 2016 - 2018 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.pubsub;

import com.google.api.core.ApiFuture;
import com.google.api.core.ApiFutureCallback;
import com.google.api.core.ApiFutures;
import com.google.api.gax.rpc.NotFoundException;
import com.google.cloud.pubsub.v1.Publisher;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.inject.Inject;
import com.google.protobuf.ByteString;
import com.google.pubsub.v1.ProjectTopicName;
import com.google.pubsub.v1.PubsubMessage;
import com.spotify.ffwd.model.Batch;
import com.spotify.ffwd.model.Event;
import com.spotify.ffwd.model.Metric;
import com.spotify.ffwd.output.BatchablePluginSink;
import com.spotify.ffwd.output.FakeBatchablePluginSinkBase;
import com.spotify.ffwd.serializer.Serializer;
import eu.toolchain.async.AsyncFramework;
import eu.toolchain.async.AsyncFuture;
import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Executor;
import lombok.extern.slf4j.Slf4j;

/**
 * This output plugin sends metrics to Google pubsub.
 *
 * Notice most of the methods return `async.resolved()`. This is because there's no apparent way to
 * map from `ApiFuture` to `AsyncFuture` and the PluginSink that executes this one calls
 * `collectAndDiscard` on the futures anyway.
 */
@Slf4j
public class PubsubPluginSink extends FakeBatchablePluginSinkBase implements BatchablePluginSink {
  @Inject
  AsyncFramework async;

  @Inject
  Publisher publisher;

  @Inject
  Serializer serializer;

  @Inject
  TopicAdmin topicAdmin;

  @Inject
  ProjectTopicName topicName;

  private final Executor executorService = MoreExecutors.directExecutor();

  @Override
  public boolean isReady() {
    return true;
  }

  @Override
  public void init() { }

  @Override
  public AsyncFuture<Void> sendEvents(Collection<Event> events) {
    log.debug("Sending events is not supported!");
    return async.resolved();
  }

  private void publishPubSub(ByteString bytes) {
    final ApiFuture<String> publish = publisher.publish(
      PubsubMessage.newBuilder().setData(bytes).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);
  }

  // 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);
    }
    log.debug("Total size of sent metrics {} bytes", size);
    return async.resolved();
  }

  @Override
  public AsyncFuture<Void> sendBatches(Collection<Batch> batches) {
    final List<Metric> metrics = convertBatchesToMetrics(batches);
    return sendMetrics(metrics);
  }


  @Override
  public void sendEvent(final Event event) {
    sendEvents(Collections.singletonList(event));
  }

  @Override
  public void sendMetric(final Metric metric) {
    sendMetrics(Collections.singletonList(metric));
  }

  @Override
  public void sendBatch(final Batch batch) {
    sendBatches(Collections.singletonList(batch));
  }

  /**
   * If the service account permissions allow it, check to see if the topic exists. Topic
   * creation is handled outside of this plugin as to limit the privileges given to the producer.
   */
  @Override
  public AsyncFuture<Void> start() {
    log.info("Connecting to topic {}", topicName);
    try {
      topicAdmin.getClient().getTopic(topicName);
      log.info("Topic exists");
    } catch (IOException e) {
      log.error("Topic admin", e);
    } catch (NotFoundException e) {
      log.warn("Topic {} not found or permission issues", topicName);
    }
    return async.resolved();
  }

  @Override
  public AsyncFuture<Void> stop() {
    return async.call(() -> {
      publisher.shutdown();
      return null;
    });
  }
}

@@ -1589,16 +1589,39 @@
 l.List;%0A
+import java.util.UUID;%0A
 import j
@@ -3081,73 +3081,8 @@
  %7D%0A%0A
-  // The pubsub plugin only supports sending batches of metrics.%0A
   @O
@@ -3166,28 +3166,79 @@
     
-log.info(%22S
+final UUID traceId = UUID.randomUUID();%0A    log.info(%22%7B%7D: Start s
 ending
- %7B%7D
  met
@@ -3248,44 +3248,57 @@
 s%22, 
-metrics.size());%0A%0A    int size = 0;%0A
+traceId);%0A%0A    for (Metric metric : metrics) %7B%0A  
     
@@ -3313,29 +3313,108 @@
     
+  
 final 
-ByteString m = 
+ApiFuture%3CString%3E publish = publisher.publish(PubsubMessage.newBuilder()%0A          .setData(
 Byte
@@ -3460,65 +3460,374 @@
 tric
-s));%0A      publishPubSub(m);%0A      size += m.size(
+)))%0A          .build()%0A        );%0A        ApiFutures.addCallback(publish, new ApiFutureCallback%3CString%3E() %7B%0A          @Override%0A          public void onFailure(Throwable t) %7B%0A            log.error(%22Failed sending metrics %7B%7D%22, t.getMessage());%0A          %7D%0A%0A          @Override%0A          public void onSuccess(String messageId) %7B %7D%0A%0A        %7D, executorService
 );%0A    
+  
 %7D ca
@@ -3844,32 +3844,34 @@
 tion e) %7B%0A      
+  
 log.error(%22Faile
@@ -3879,34 +3879,22 @@
  to 
-serialize batch of
+publish
  metric
-s
  %7B%7D%22
@@ -3895,24 +3895,32 @@
 ic %7B%7D%22, e);%0A
+      %7D%0A
     %7D%0A    lo
@@ -3925,57 +3925,52 @@
 log.
-debug(%22Total size of sent metrics %7B%7D bytes%22, size
+info(%22%7B%7D: Finished sending metrics%22, traceId
 );%0A 

Revert sendMetrics