package io.latent.storm.rabbitmq;

import backtype.storm.spout.Scheme;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import io.latent.storm.rabbitmq.config.ConsumerConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.List;
import java.util.Map;

/**
 * A simple RabbitMQ spout that emits an anchored tuple stream (on the default stream). This can be used with
 * Storm's guaranteed message processing.
 * 
 * @author peter@latent.io
 */
public class RabbitMQSpout extends BaseRichSpout {
  private final MessageScheme scheme;
  private final Declarator declarator;

  private transient Logger logger;
  private transient RabbitMQConsumer consumer;
  private transient SpoutOutputCollector collector;

  public RabbitMQSpout(Scheme scheme) {
    this(MessageScheme.Builder.from(scheme), new Declarator.NoOp());
  }

  public RabbitMQSpout(Scheme scheme, Declarator declarator) {
    this(MessageScheme.Builder.from(scheme), declarator);
  }

  public RabbitMQSpout(MessageScheme scheme, Declarator declarator) {
    this.scheme = scheme;
    this.declarator = declarator;
  }

  @Override
  public void open(final Map config,
                   final TopologyContext context,
                   final SpoutOutputCollector spoutOutputCollector) {
    ConsumerConfig consumerConfig = ConsumerConfig.getFromStormConfig(config);
    ErrorReporter reporter = new ErrorReporter() {
      @Override
      public void reportError(Throwable error) {
        spoutOutputCollector.reportError(error);
      }
    };
    consumer = loadConsumer(declarator, reporter, consumerConfig);
    scheme.open(config, context);
    consumer.open();
    logger = LoggerFactory.getLogger(RabbitMQSpout.class);
    collector = spoutOutputCollector;
  }

  protected RabbitMQConsumer loadConsumer(Declarator declarator,
                                          ErrorReporter reporter,
                                          ConsumerConfig config) {
    return new RabbitMQConsumer(config.getConnectionConfig(),
                                config.getPrefetchCount(),
                                config.getQueueName(),
                                config.isRequeueOnFail(),
                                declarator,
                                reporter);
  }

  @Override
  public void close() {
    consumer.close();
    scheme.close();
    super.close();
  }

  @Override
  public void nextTuple() {
    Message message;
    while ((message = consumer.nextMessage()) != Message.NONE) {
      List<Object> tuple = extractTuple(message);
      if (!tuple.isEmpty()) {
        emit(tuple, message, collector);
      }
    }
  }

  protected List<Integer> emit(List<Object> tuple,
                               Message message,
                               SpoutOutputCollector spoutOutputCollector) {
    return spoutOutputCollector.emit(tuple, getDeliveryTag(message));
  }

  private List<Object> extractTuple(Message message) {
    long deliveryTag = getDeliveryTag(message);
    try {
      List<Object> tuple = scheme.deserialize(message);
      if (tuple != null && !tuple.isEmpty()) {
        return tuple;
      }
      String errorMsg = "Deserialization error for msgId " + deliveryTag;
      logger.warn(errorMsg);
      collector.reportError(new Exception(errorMsg));
    } catch (Exception e) {
      logger.warn("Deserialization error for msgId " + deliveryTag, e);
      collector.reportError(e);
    }
    // get the malformed message out of the way by dead-lettering (if dead-lettering is configured) and move on
    consumer.deadLetter(deliveryTag);
    return Collections.emptyList();
  }

  @Override
  public void ack(Object msgId) {
    if (msgId instanceof Long) consumer.ack((Long) msgId);
  }

  @Override
  public void fail(Object msgId) {
    if (msgId instanceof Long) consumer.fail((Long) msgId);
  }

  @Override
  public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
    outputFieldsDeclarer.declare(scheme.getOutputFields());
  }

  protected long getDeliveryTag(Message message) {
    return ((Message.DeliveredMessage) message).getDeliveryTag();
  }
}

@@ -906,16 +906,43 @@
 ector;%0A%0A
+  private boolean active;%0A%0A
   public
@@ -969,24 +969,24 @@
 e scheme) %7B%0A
-
     this(Mes
@@ -1951,24 +1951,43 @@
 tCollector;%0A
+    active = true;%0A
   %7D%0A%0A  prote
@@ -2641,24 +2641,49 @@
 xtTuple() %7B%0A
+    if (!active) return;%0A
     Message 
@@ -4187,24 +4187,24 @@
 Declarer) %7B%0A
-
     outputFi
@@ -4252,24 +4252,203 @@
 ds());%0A  %7D%0A%0A
+  @Override%0A  public void deactivate()%0A  %7B%0A    super.deactivate();%0A    active = false;%0A  %7D%0A%0A  @Override%0A  public void activate()%0A  %7B%0A    super.activate();%0A    active = true;%0A  %7D%0A%0A
   protected 

support for activate/deactivate in RabbitMQSpout. when a spout is deactivated, this implementation does not close the connection to rabbitmq. The reasoning behind this decision is that disconnecting from rabbitmq will cause all tuples that in flight to be put back into the queue and replayed later upon reconnection. That may or may not be ok. By not disconnecting from rabbitmq intentionally upon deactivation, we can allow tuples in flight to finish processing all the way through. We just won't emit any new tuples to be processed by Storm from the spout('s nextTuple method). The downside of this that once in-flight tuples complete processing, the spout will still maintain the connection to rabbitmq. Also the rabbitmq server will stream more messages upto the configured rabbitmq prefetch count to the spout (they just won't be emitted into storm's processing pipeline by the spout).