/*
 * Copyright (c) 2013 DataTorrent, Inc. ALL Rights Reserved.
 *
 * 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.datatorrent.lib.io.jms;

import com.datatorrent.api.DefaultInputPort;
import com.datatorrent.api.annotation.InputPortFieldAnnotation;

import javax.jms.JMSException;
import javax.jms.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * ActiveMQ output adapter operator with only one input port, which produce data into ActiveMQ message bus.<p><br>
 *
 * <br>
 * Ports:<br>
 * <b>Input</b>: Have only one input port<br>
 * <b>Output</b>: No output port<br>
 * <br>
 * Properties:<br>
 * None<br>
 * <br>
 * Compile time checks:<br>
 * Class derived from this has to implement the abstract method createMessage() <br>
 * <br>
 * Run time checks:<br>
 * None<br>
 * <br>
 * Benchmarks:<br>
 * TBD<br>
 * <br>
 *
 * @since 0.3.2
 */
public abstract class AbstractActiveMQSinglePortOutputOperator<T> extends AbstractActiveMQOutputOperator
{
  private static final Logger logger = LoggerFactory.getLogger(AbstractActiveMQSinglePortOutputOperator.class);
  long countMessages = 0;  // Number of messages produced so far

  /**
   * Convert tuple into JMS message. Tuple can be any Java Object.
   * @param tuple
   * @return Message
   */
  protected abstract Message createMessage(T tuple);

  /**
   * The single input port.
   */
  @InputPortFieldAnnotation(name = "ActiveMQInputPort")
  public final transient DefaultInputPort<T> inputPort = new DefaultInputPort<T>()
  {
    @Override
    public void process(T tuple)
    {
      countMessages++;

      if (countMessages > maxSendMessage && maxSendMessage != 0) {
        if (countMessages == maxSendMessage + 1) {
          logger.warn("Reached maximum send messages of {}", maxSendMessage);
        }
        return; // Stop sending messages after max limit.
      }

      try {
        Message msg = createMessage(tuple);
        getProducer().send(msg);
        //logger.debug("process message {}", tuple.toString());
      }
      catch (JMSException ex) {
        logger.debug(ex.getLocalizedMessage());
      }
    }
  };
}

@@ -1851,204 +1851,72 @@
   * 
-The single input port.%0A   */%0A  @InputPortFieldAnnotation(name = %22ActiveMQInputPort%22)%0A  public final transient DefaultInputPort%3CT%3E inputPort = new DefaultInputPort%3CT%3E()%0A  %7B%0A    @Override%0A    public
+Convert to and send message.%0A   * @param tuple%0A   */%0A  protected
  voi
@@ -1924,16 +1924,21 @@
  process
+Tuple
 (T tuple
@@ -1942,17 +1942,14 @@
 ple)
-%0A 
  
+%0A
   %7B%0A
-  
     
@@ -1966,18 +1966,16 @@
 ges++;%0A%0A
-  
     if (
@@ -2037,18 +2037,16 @@
 %7B%0A      
- 
- 
 if (coun
@@ -2076,26 +2076,24 @@
 sage + 1) %7B%0A
-  
         logg
@@ -2152,26 +2152,24 @@
 ndMessage);%0A
-  
       %7D%0A    
@@ -2170,18 +2170,16 @@
 %7D%0A      
-  
 return; 
@@ -2220,18 +2220,16 @@
  limit.%0A
-  
     %7D%0A%0A 
@@ -2235,18 +2235,14 @@
     
-  
 try %7B%0A
-  
     
@@ -2285,18 +2285,16 @@
 ;%0A      
-  
 getProdu
@@ -2310,18 +2310,16 @@
 d(msg);%0A
-  
       //
@@ -2372,30 +2372,26 @@
 ing());%0A    
-  %7D%0A  
+%7D%0A
     catch (J
@@ -2418,57 +2418,341 @@
     
-  logger.debug(ex.getLocalizedMessage());%0A      %7D
+throw new RuntimeException(%22Failed to send message%22, ex);%0A    %7D%0A  %7D%0A  %0A  /**%0A   * The single input port.%0A   */%0A  @InputPortFieldAnnotation(name = %22ActiveMQInputPort%22)%0A  public final transient DefaultInputPort%3CT%3E inputPort = new DefaultInputPort%3CT%3E()%0A  %7B%0A    @Override%0A    public void process(T tuple)%0A    %7B%0A      processTuple(tuple);
 %0A   

allow derived class to override processing