/**
 * InboundMessageHandlerBean.java
 * Version: 1.0
 * Date:    2004/10/15
 * @author Selvan V. Rajan
 * @version 1.0 
 */
package com.acme.intg;

import javax.ejb.*;
import javax.jms.*;
import weblogic.ejb.*;

import com.bea.wli.bpm.proxy.JpdProxy;
import com.bea.wli.bpm.proxy.JpdProxyException;
import com.bea.wli.bpm.proxy.JpdProxySession; 


import java.text.SimpleDateFormat;
import java.util.Date;
import weblogic.ejbgen.Constants;
import weblogic.jndi.*;
import weblogic.logging.NonCatalogLogger;

import com.acme.logging.AppLogger;
import com.acme.exception.AcmeException;
import com.acme.util.Parameters;
import com.acme.client.CreditCardProcess;
import com.bea.data.RawData;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import javax.rmi.PortableRemoteObject;

/**
 * This message-driven bean ...
 * @author Selvan V. Rajan
 * @version 1.0 
 * @ejbgen:message-driven clients-on-same-server="false" 
 *   max-beans-in-free-pool="30"
 *   ejb-name = InboundMessageHandler
 *   default-transaction="NotSupported"
 *   destination-jndi-name = CCInboundQueue
 *   destination-type = javax.jms.Queue
 *   initial-beans-in-free-pool="20"
 *
 */
@weblogic.ejbgen.MessageDriven(clientsOnSameServer = Constants.Bool.FALSE, 
                               maxBeansInFreePool = "30", 
                               ejbName = "InboundMessageHandler", 
                               defaultTransaction = weblogic.ejbgen.MessageDriven.DefaultTransaction.NOT_SUPPORTED, 
                               destinationJndiName = "CCInboundQueue", 
	destinationType = "javax.jms.Queue", 
	initialBeansInFreePool = "20"
)
public class InboundMessageHandlerBean
  extends GenericMessageDrivenBean
  implements MessageDrivenBean, MessageListener
{
  public static CreditCardProcess creditCardProcess;
  private String errorMessage;
  private static AppLogger logger = new AppLogger("InboundMessageHandlerBean");
  private static InitialContext initContext=null;
  private static long max=0, min=10000, cuml=0;
  private long curr;
  private static int total=0;
  private static boolean done=false;
  private static Date pStartTime, pEndTime;
  private static SimpleDateFormat timestampFormat = new SimpleDateFormat(Parameters.getSProperty("benchmark.time.format"));
  private static SimpleDateFormat minuteFormat = new SimpleDateFormat(Parameters.getSProperty("benchmark.minute.format"));
  
  public static final String COORPROPNAME="ASYNC_BIN_SEL";

  /**
    * Defines the JNDI context factory.
  */

    public final static String DEFAULT_JNDI_FACTORY = "weblogic.jndi.WLInitialContextFactory";

    /**
     * Defines the JMS connection factory for the queue.
     */

    public final static String DEFAULT_JMS_CONNECTION_FACTORY = "weblogic.jws.jms.QueueConnectionFactory";

    /**
     * Defines the queue.
     */
    public final static String OUTBOUND_QUEUE = "CCOutboundQueue";
    
    private static QueueConnectionFactory qConnectionFactory ;

    private static QueueConnection qConnection ;

    private static QueueSession qSession;

    private static Queue queue ;

    private static QueueSender qSender;

    private static TextMessage txtMessage;
    static {
        try {
            getContext();   
        }catch(Exception e) {}
    }
    
  /**
   * This method is required by the EJB Specification,
   * but is not used by this example.
   */
  public void ejbCreate() throws CreateException {
    try {
        logger.info("InboundMessageHandlerBean.processTextMessage - Calling JPD Proxy factory");
        creditCardProcess = getProxy();
        ((JpdProxySession)creditCardProcess).setConversationID("1234");       
    }catch (JpdProxyException jpe)
    {
      logger.error("InboundMessageHandlerBean.processTextMessage - Failed calling JPD");
      errorMessage = "Failed calling the CreditCardProcess: an error occured in the process definition";

      logger.error(errorMessage + jpe.toString());
      //throw new Exception(errorMessage);
    }   
    
  }
  
  public void ejbRemove() {
    try {
        errorMessage=null;
    } catch(Exception e) {
        
    }    
  }
  
   
  public void onMessage(Message message) {
    Date startTime = new Date();
    if (pStartTime == null)
        pStartTime=startTime;
        
    logger.info("\n\n*******************************************");
    logger.info("InboundMessageHandlerBean.onMessage - Begin");

    // Holds the inbound message for POS transactions sent in text format.
    TextMessage requestMessage = null;
    done=false;
 
    try
    {
      if (message instanceof TextMessage)
      {
        requestMessage = (TextMessage) message;
        processTextMessage(requestMessage);
      }
      else
      {
        // Do not throw an exception since we are in a MDB; return instead.
        logger.error("InboundMessageHandlerBean.onMessage - Invalid JMS message type:\n" +
                                          message.toString());
        return;
      }
    }
    catch (Exception aue)
    {
      // Error was logged, abnormal condition was handled; simply return
      // NOTE: A message-driven cannot throw an exception.
      return;
    }
    if ( !done) 
    {
        Date endTime = new Date();
        pEndTime = endTime;
        logBenchmark(startTime, endTime);
    }

    logger.info("InboundMessageHandlerBean.onMessage - End");
    logger.info("\n\n-----------------------------------------");  
      
  }
  
  private void processTextMessage(TextMessage message)
          throws AcmeException
  {
    logger.info("InboundMessageHandlerBean.processTextMessage - Begin");

    // Holds the required data extracted from the message.
    String request = null;
    String coorid = null;
    
    try
    {
      request = message.getText();
      coorid = message.getStringProperty(COORPROPNAME);
    }
    catch (JMSException je)
    {
      errorMessage = "Failed to extract required data from the inbound message";

      logger.error(errorMessage);
      throw new AcmeException(errorMessage);
    }

    if (isBlankString(request))
    {
      errorMessage = "No data or payload found in the inbound message";

      logger.error(errorMessage);
      throw new AcmeException(errorMessage);
    }

    if ( request.equalsIgnoreCase("done"))
    {
        done=true;
        printResults();
        pStartTime=null;
        pEndTime=null;
        min=10000;
        max=0;
        cuml=0;
        total=0;
        done=true;
        return;   
    }

    try
    {
        logger.info("InboundMessageHandlerBean.processTextMessage - Calling JPD");
        com.bea.data.RawData rawData = RawData.Factory.newInstance(request);
        RawData outRawData = creditCardProcess.clientRequestwithReturn(rawData);
        String outString = outRawData.stringValue();
        try {
            if (initContext == null)
                getContext();

            txtMessage.setStringProperty(COORPROPNAME, coorid);
            txtMessage.setText(outString);
            qSender.send(txtMessage);
            logger.info("Sent message to outbound queue.");

        }catch(Exception e) {
               logger.error("InboundMessageHandlerBean.processTextMessage - EJB Context Error");     
               throw new AcmeException(e.getMessage());
        }
              
        logger.info("InboundMessageHandlerBean.processTextMessage - JPD Returned");
    }
    catch (JpdProxyException jpe)
    {
      logger.error("InboundMessageHandlerBean.processTextMessage - Failed calling JPD");
      errorMessage = "Failed calling the CreditCardProcess: an error occured in the process definition";

      logger.error(errorMessage + jpe.toString());
      throw new AcmeException(errorMessage);
    }

    logger.info("InboundMessageHandlerBean.processTextMessage - End");
  }
  
  private static CreditCardProcess getProxy() throws JpdProxyException
  {
    if (creditCardProcess == null)
        creditCardProcess = (CreditCardProcess) JpdProxy.create(
              CreditCardProcess.class,
              CreditCardProcess.SERVICE_URI);
    else 
      ((JpdProxySession)creditCardProcess).reset();
    return creditCardProcess;
  } 
  
  private boolean isBlankString(String aString)
  {
    return (aString == null || aString.equals(""));
  }
  
  private void printResults() {
    logger.debug("Min: "+(float)min/1000+" Max: "+(float)max/1000+" Total: "+total+" Avg: "+(float)cuml/total/1000);  
    logger.debug("TPS: "+1000/(float)(new Date(pEndTime.getTime()-pStartTime.getTime())).getTime()*total);
  }
  
  private void logBenchmark(Date startTime, Date endTime)
  {
    logger.info("InboundMessageHandlerBean.logBenchmark - Begin");


    Date delta = new Date(endTime.getTime() - startTime.getTime());

    curr = delta.getTime();
    StringBuffer messageBuffer = new StringBuffer();

    
    messageBuffer.append("\n*** Benchmark results:");
    messageBuffer.append(" Start = " + timestampFormat.format(startTime));

    messageBuffer.append(" End = " + timestampFormat.format(endTime));
    messageBuffer.append(" Elapsed = " + minuteFormat.format(delta));
    messageBuffer.append(" ***");

    logger.debug( messageBuffer.toString());
    logger.info("InboundMessageHandlerBean.logBenchmark - End");
    updateStat();
  }
  private void updateStat() {
    synchronized(this) {
        if (curr > max)
            max = curr;
        if (curr < min)
            min = curr;
        cuml=cuml+curr;
        total++;
    }
  }
  
  private static synchronized void getContext() throws Exception {
    if ( initContext == null)
    {    
    try {
        initContext = new InitialContext();
        qConnectionFactory =( QueueConnectionFactory ) initContext.lookup( DEFAULT_JMS_CONNECTION_FACTORY );
        logger.debug("getConnectionFactory Successful");
        qConnection=qConnectionFactory.createQueueConnection();
        logger.debug("getConnection Successful");
        qSession=qConnection.createQueueSession( false, Session.AUTO_ACKNOWLEDGE );
        logger.debug("getSession Successful");
        queue=(Queue)initContext.lookup( OUTBOUND_QUEUE );
        logger.debug("getQueue Successful");
            qConnection.start();
            logger.info( "Creating sender" );
            qSender = qSession.createSender( queue );
            logger.info( "Creating text message" );
            txtMessage = qSession.createTextMessage();
        logger.debug("getContext Successful");
     } 
     catch (Exception ne) {
        logger.error("getContext :"+ne.getMessage());
        throw new EJBException(ne);
     }
    }
  }
 private Object narrow(Object ref, Class c) {
    return PortableRemoteObject.narrow(ref, c);
  }
}


