content: /**
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you 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 org.apache.apex.malhar.contrib.rabbitmq;
  
  import java.io.IOException;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.Map;
  import java.util.Map.Entry;
  import java.util.Set;
  import java.util.concurrent.ArrayBlockingQueue;
  
  import javax.validation.constraints.NotNull;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.apache.apex.malhar.lib.util.KeyValPair;
  import org.apache.apex.malhar.lib.wal.WindowDataManager;
  
  import com.rabbitmq.client.AMQP;
  import com.rabbitmq.client.Channel;
  import com.rabbitmq.client.Connection;
  import com.rabbitmq.client.ConnectionFactory;
  import com.rabbitmq.client.DefaultConsumer;
  import com.rabbitmq.client.Envelope;
  import com.rabbitmq.client.ShutdownSignalException;
  
  import com.datatorrent.api.Context.OperatorContext;
  import com.datatorrent.api.InputOperator;
  import com.datatorrent.api.Operator;
  import com.datatorrent.netlet.util.DTThrowable;
  
  /**
   * This is the base implementation of a RabbitMQ input operator.&nbsp;
   * Subclasses should implement the methods which convert RabbitMQ messages to tuples.
   * <p>
   * Ports:<br>
   * <b>Input</b>: No input port<br>
   * <b>Output</b>: Can have any number of output ports<br>
   * <br>
   * Properties:<br>
   * <b>tuple_blast</b>: Number of tuples emitted in each burst<br>
   * <b>bufferSize</b>: Size of holding buffer<br>
   * <b>host</b>:the address for the consumer to connect to rabbitMQ producer<br>
   * <b>exchange</b>:the exchange for the consumer to connect to rabbitMQ producer<br>
   * <b>exchangeType</b>:the exchangeType for the consumer to connect to rabbitMQ producer<br>
   * <b>routingKey</b>:the routingKey for the consumer to connect to rabbitMQ producer<br>
   * <b>queueName</b>:the queueName for the consumer to connect to rabbitMQ producer<br>
   * <br>
   * Compile time checks:<br>
   * Class derived from this has to implement the abstract method emitTuple() <br>
   * <br>
   * Run time checks:<br>
   * None<br>
   * <br>
   * <b>Benchmarks</b>: Blast as many tuples as possible in inline mode<br>
   * <table border="1" cellspacing=1 cellpadding=1 summary="Benchmark table for AbstractRabbitMQInputOperator&lt;K,V extends Number&gt; operator template">
   * <tr><th>In-Bound</th><th>Out-bound</th><th>Comments</th></tr>
   * <tr><td><b>10 thousand K,V pairs/s</td><td>One tuple per key per window per port</td><td>In-bound rate is the main determinant of performance. Operator can emit about 10 thousand unique (k,v immutable pairs) tuples/sec as RabbitMQ DAG. Tuples are assumed to be
   * immutable. If you use mutable tuples and have lots of keys, the benchmarks may differ</td></tr>
   * </table><br>
   * </p>
   * @displayName Abstract RabbitMQ Input
   * @category Messaging
   * @tags input operator
   *
   * @since 0.3.2
   */
  public abstract class AbstractRabbitMQInputOperator<T> implements
      InputOperator, Operator.ActivationListener<OperatorContext>,
      Operator.CheckpointNotificationListener
  {
    private static final Logger logger = LoggerFactory.getLogger(AbstractRabbitMQInputOperator.class);
    @NotNull
    protected String host;
    protected int port;
    @NotNull
    protected String exchange;
    @NotNull
    protected String exchangeType;
    protected String routingKey = "";
    protected String queueName; // Has to be supplied by client when exchangeType is not "fanout"
    protected transient ConnectionFactory connFactory;
  //  QueueingConsumer consumer = null;
  
    private static final int DEFAULT_BLAST_SIZE = 1000;
    private static final int DEFAULT_BUFFER_SIZE = 1024 * 1024;
    private int tuple_blast = DEFAULT_BLAST_SIZE;
    protected int bufferSize = DEFAULT_BUFFER_SIZE;
  
    protected transient Connection connection;
    protected transient Channel channel;
    protected transient TracingConsumer tracingConsumer;
    protected transient String cTag;
  
    protected transient ArrayBlockingQueue<KeyValPair<Long,byte[]>> holdingBuffer;
    private WindowDataManager windowDataManager;
    protected final transient Map<Long, byte[]> currentWindowRecoveryState;
    private final transient Set<Long> pendingAck;
    private final transient Set<Long> recoveredTags;
    private transient long currentWindowId;
    private transient int operatorContextId;
  
    public AbstractRabbitMQInputOperator()
    {
      currentWindowRecoveryState = new HashMap<Long, byte[]>();
      pendingAck = new HashSet<Long>();
      recoveredTags = new HashSet<Long>();
      windowDataManager = new WindowDataManager.NoopWindowDataManager();
    }
  
  
  /**
   * define a consumer which can asynchronously receive data,
   * and added to holdingBuffer
   */
    public class TracingConsumer extends DefaultConsumer
    {
      public TracingConsumer(Channel ch)
      {
        super(ch);
      }
  
      @Override
      public void handleConsumeOk(String c)
      {
        logger.debug(this + ".handleConsumeOk(" + c + ")");
        super.handleConsumeOk(c);
      }
  
      @Override
      public void handleCancelOk(String c)
      {
        logger.debug(this + ".handleCancelOk(" + c + ")");
        super.handleCancelOk(c);
      }
  
      @Override
      public void handleShutdownSignal(String c, ShutdownSignalException sig)
      {
        logger.debug(this + ".handleShutdownSignal(" + c + ", " + sig + ")");
        super.handleShutdownSignal(c, sig);
      }
  
      @Override
      public void handleDelivery(String consumer_Tag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException
      {
        long tag = envelope.getDeliveryTag();
        if (envelope.isRedeliver() && (recoveredTags.contains(tag) || pendingAck.contains(tag))) {
          if (recoveredTags.contains(tag)) {
            pendingAck.add(tag);
          }
          return;
        }
  
        // Acknowledgements are sent at the end of the window after adding to idempotency manager
        pendingAck.add(tag);
        holdingBuffer.add(new KeyValPair<Long, byte[]>(tag, body));
        logger.debug("Received Async message: {}  buffersize: {} ", new String(body), holdingBuffer.size());
      }
    }
  
    @Override
    public void emitTuples()
    {
      int ntuples = tuple_blast;
      if (ntuples > holdingBuffer.size()) {
        ntuples = holdingBuffer.size();
      }
      for (int i = ntuples; i-- > 0;) {
        KeyValPair<Long, byte[]> message =  holdingBuffer.poll();
        currentWindowRecoveryState.put(message.getKey(), message.getValue());
        emitTuple(message.getValue());
      }
    }
  
    public abstract void emitTuple(byte[] message);
  
    @Override
    public void beginWindow(long windowId)
    {
      currentWindowId = windowId;
      if (windowId <= this.windowDataManager.getLargestCompletedWindow()) {
        replay(windowId);
      }
    }
  
    @SuppressWarnings("unchecked")
    private void replay(long windowId)
    {
      Map<Long, byte[]> recoveredData;
      try {
        recoveredData = (Map<Long, byte[]>)this.windowDataManager.retrieve(windowId);
        if (recoveredData == null) {
          return;
        }
        for (Entry<Long, byte[]>  recoveredEntry : recoveredData.entrySet()) {
          recoveredTags.add(recoveredEntry.getKey());
          emitTuple(recoveredEntry.getValue());
        }
      } catch (IOException e) {
        DTThrowable.rethrow(e);
      }
    }
  
  
    @Override
    public void endWindow()
    {
      //No more messages can be consumed now. so we will call emit tuples once more
      //so that any pending messages can be emitted.
      KeyValPair<Long, byte[]> message;
      while ((message = holdingBuffer.poll()) != null) {
        currentWindowRecoveryState.put(message.getKey(), message.getValue());
        emitTuple(message.getValue());
      }
  
      try {
        this.windowDataManager.save(currentWindowRecoveryState, currentWindowId);
      } catch (IOException e) {
        DTThrowable.rethrow(e);
      }
  
      currentWindowRecoveryState.clear();
  
      for (Long deliveryTag : pendingAck) {
        try {
          channel.basicAck(deliveryTag, false);
        } catch (IOException e) {
          DTThrowable.rethrow(e);
        }
      }
  
      pendingAck.clear();
    }
  
    @Override
    public void setup(OperatorContext context)
    {
      this.operatorContextId = context.getId();
      holdingBuffer = new ArrayBlockingQueue<KeyValPair<Long, byte[]>>(bufferSize);
      this.windowDataManager.setup(context);
    }
  
    @Override
    public void teardown()
    {
      this.windowDataManager.teardown();
    }
  
    @Override
    public void activate(OperatorContext ctx)
    {
      try {
        connFactory = new ConnectionFactory();
        connFactory.setHost(host);
        if (port != 0) {
          connFactory.setPort(port);
        }
  
        connection = connFactory.newConnection();
        channel = connection.createChannel();
  
        channel.exchangeDeclare(exchange, exchangeType);
        boolean resetQueueName = false;
        if (queueName == null) {
          // unique queuename is generated
          // used in case of fanout exchange
          queueName = channel.queueDeclare().getQueue();
          resetQueueName = true;
        } else {
          // user supplied name
          // used in case of direct exchange
          channel.queueDeclare(queueName, true, false, false, null);
        }
  
        channel.queueBind(queueName, exchange, routingKey);
  
  //      consumer = new QueueingConsumer(channel);
  //      channel.basicConsume(queueName, true, consumer);
        tracingConsumer = new TracingConsumer(channel);
        cTag = channel.basicConsume(queueName, false, tracingConsumer);
        if (resetQueueName) {
          queueName = null;
        }
      } catch (IOException ex) {
        throw new RuntimeException("Connection Failure", ex);
      }
    }
  
    @Override
    public void deactivate()
    {
      try {
        channel.close();
        connection.close();
      } catch (IOException ex) {
        logger.debug(ex.toString());
      }
    }
  
    @Override
    public void beforeCheckpoint(long windowId)
    {
    }
  
    @Override
    public void checkpointed(long windowId)
    {
    }
  
    @Override
    public void committed(long windowId)
    {
      try {
        windowDataManager.committed(windowId);
      } catch (IOException e) {
        throw new RuntimeException("committing", e);
      }
    }
  
    public void setTupleBlast(int i)
    {
      this.tuple_blast = i;
    }
  
    public String getHost()
    {
      return host;
    }
  
    public void setHost(String host)
    {
      this.host = host;
    }
  
    public int getPort()
    {
      return port;
    }
  
    public void setPort(int port)
    {
      this.port = port;
    }
  
    public String getExchange()
    {
      return exchange;
    }
  
    public void setExchange(String exchange)
    {
      this.exchange = exchange;
    }
  
    public String getQueueName()
    {
      return queueName;
    }
  
    public void setQueueName(String queueName)
    {
      this.queueName = queueName;
    }
  
    public String getExchangeType()
    {
      return exchangeType;
    }
  
    public void setExchangeType(String exchangeType)
    {
      this.exchangeType = exchangeType;
    }
  
    public String getRoutingKey()
    {
      return routingKey;
    }
  
    public void setRoutingKey(String routingKey)
    {
      this.routingKey = routingKey;
    }
  
    public WindowDataManager getWindowDataManager()
    {
      return windowDataManager;
    }
  
    public void setWindowDataManager(WindowDataManager windowDataManager)
    {
      this.windowDataManager = windowDataManager;
    }
  
  }
  