/*
    * Copyright (C) The Apache Software Foundation. All rights reserved.
    *
    * This software is published under the terms of the Apache Software License
    * version 1.1, a copy of which has been included with this distribution in
    * the LICENSE file.
    * 
    * $Id: BridgeMDO.java 155459 2005-02-26 13:24:44Z dirkv $
    */
  package org.apache.commons.messagelet;
  
  import java.util.Enumeration;
  
  import javax.jms.BytesMessage;
  import javax.jms.Destination;
  import javax.jms.JMSException;
  import javax.jms.MapMessage;
  import javax.jms.Message;
  import javax.jms.ObjectMessage;
  import javax.jms.StreamMessage;
  import javax.jms.TextMessage;
  import javax.servlet.ServletException;
  
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  import org.apache.commons.messenger.Messenger;
  
  
  /** <p><code>BridgeMDO</code> is an MDO which implements a JMS bridge
   * from one JMS destination and connection to another.
   * This allows messages to be consumed on one destination and sent to 
   * another JMS destination, using possibly different JMS providers.
   * For example this can be used to bridge from SpiritWave to MQSeries. 
   * </p>
   * <p>
   * This class is a useful base class to other possible bridge implementations
   * such as 2 phase commit bridges or bridges with some complex transformation etc.
   * This class has a number of Factory and Strategy methods to further customize
   * the acknowledgement and transaction management, the message construction, 
   * transformation and how to handle message headers etc.
   * </p>
   *
   * @author <a href="mailto:jstrachan@apache.org">James Strachan</a>
   * @version $Revision: 155459 $
   */
  public class BridgeMDO extends MessengerMDO {
  
      /** Logger */
      private static final Log log = LogFactory.getLog(BridgeMDO.class);
    
      /** the Messenger used to output messages */
      private Messenger outputMessenger;
        
      /** the Destination output messages will be sent to */
      private Destination outputDestination;
  
      /** the name of the messenger to use for output */      
      private String outputConnection;
      
      /** the name of the destination to use */
      private String outputSubject;
      
      /** the buffer size used for ByteMessage and StreamMessage copying */
      private int bufferSize = 32 * 1024;
  
      /** should this MDO work in transacted mode */
      private boolean transacted = false;
      
      
      public BridgeMDO() {
      }
      
      public void init() throws ServletException {
          try {
              Messenger messenger = getMessenger();
              Messenger outputMessenger = getOutputMessenger();
              
              if ( messenger == null ) {
                  throw new ServletException( "No input Messenger is defined for this Bridge" );
              }
              if ( outputMessenger == null ) {
                  throw new ServletException( "No output Messenger is defined for this Bridge" );
              }
              
              // enable transacted mode 
              boolean tran1 = messenger.getSessionFactory().isTransacted();
              boolean tran2 = outputMessenger.getSessionFactory().isTransacted();
              
              if ( tran1 != tran2 ) {
                  throw new ServletException( 
                      "Both the input and output Messenger must have the same transacted mode. "
                      + "Input is: " + tran1 + " output is: " + tran2 
                  );
              }
              transacted = tran1;
              
              // use client acknowledgement
              
              // ### This should be specified in the Messenger.xml file
             //messenger.getSessionFactory().setAcknowledgeMode( Session.CLIENT_ACKNOWLEDGE );     
             //outputMessenger.getSessionFactory().setAcknowledgeMode( Session.CLIENT_ACKNOWLEDGE );     
 
             validateOutputDestination();
             
         }
         catch (JMSException e) {
             log.error( "Caught exception trying to configure the transacted, client acknowledge modes of the JMS connections" );
             log.error( "Caught: " + e, e);
             throw new ServletException( "Caught exception trying to configure the transacted, client acknowledge modes of the JMS connections" + e, e);
         }
     }
     
     // MessageListener interface
     //-------------------------------------------------------------------------
     public void onMessage(Message message) {
         Messenger messenger = getMessenger();
         
         try {
             Message outputMessage = createOutputMessage(message);
             if ( outputMessage != null ) {
                 Destination destination = getOutputDestination();
                 
                 if ( log.isDebugEnabled() ) {
                     log.debug( "Sending message to: " + destination );
                 }
                 
                 getOutputMessenger().send( destination, outputMessage );
             }
             acknowledge(message);
             acknowledge(outputMessage);
             commit();
         }
         catch (Exception e) {
             log.error("Could not send message due to exception", e);
             rollback();
         }
     }
     
     
     // Properties
     //-------------------------------------------------------------------------
     
     /** 
      * @return true if this MDO should work in transacted mode
      */
     public boolean isTransacted() {
         return transacted;
     }
 
     /**
      * Sets whether this MDO should work in transacted mode
      */    
     public void setTransacted(boolean transacted) {
         this.transacted = transacted;
     }
     
         
     public String getOutputConnection() {
         return outputConnection;
     }
 
     /**
      * Sets the connection name (messenger instance) to use
      * to output messages
      */    
     public void setOutputConnection(String outputConnection) {
         this.outputConnection = outputConnection;
     }
     
     public String getOutputSubject() {
         return outputSubject;
     }
     
     /** 
      * Sets the subject (i.e. destination name) to send messages to
      */
     public void setOutputSubject(String outputSubject) {
         this.outputSubject = outputSubject;
     }
     
     /**
      * Gets the Messenger used to output messages 
      */
     public Messenger getOutputMessenger() throws JMSException {
         if ( outputMessenger == null ) {
             String name = getOutputConnection();
             if ( name != null ) {
                 outputMessenger = getMessengerManager().getMessenger( name );
             }
             else {
                 // default to the input messenger
                 outputMessenger = getMessenger();
             }
         }
         return outputMessenger;
     }
     
     /**
      * Sets the Messenger used to output messages 
      */
     public void setOutputMessenger(Messenger outputMessenger) {
         this.outputMessenger = outputMessenger;
     }
     
     /**
      * Gets the Destination output messages will be sent to
      */
     public Destination getOutputDestination() throws JMSException {
         if ( outputDestination == null ) {
             String subject = getOutputSubject();
             if ( subject == null ) {
                 throw new JMSException( "A bridge must have an outputSubject defined!" );
             }
             outputDestination = getOutputMessenger().getDestination( subject );
         }
         return outputDestination;
     }
     
     /**
      * Sets the Destination output messages will be sent to
      */
     public void setOutputDestination(Destination outputDestination) {
         this.outputDestination = outputDestination;
     }
     
     /**
      * Gets the buffer size used for ByteMessage and StreamMessage copying
      */
     public int getBufferSize() {
         return bufferSize;
     }
 
     /**
      * Sets the buffer size used for ByteMessage and StreamMessage copying
      */
     public void setBufferSize(int bufferSize) {
         this.bufferSize = bufferSize;
     }
     
     
     // Implementation methods
     //-------------------------------------------------------------------------
 
     /**
      * Strategy method to perform a commit() on both the incoming Messenger and the
      * output Messenger.
      */
     protected void commit() throws JMSException {
         if ( transacted ) {
             Messenger outputMessenger = getOutputMessenger();
             Messenger inputMessenger = getMessenger();
             
             if ( outputMessenger != inputMessenger ) {
                 outputMessenger.commit();
             }
             inputMessenger.commit();
         }
     }
 
     /**
      * Strategy method to perform a rollback() on both the incoming Messenger and the
      * output Messenger.
      */
     protected void rollback() {
         if ( transacted ) {
             try {
                 Messenger outputMessenger = getOutputMessenger();
                 Messenger inputMessenger = getMessenger();
                 
                 if ( outputMessenger != inputMessenger ) {
                         outputMessenger.rollback();
                 }
             }
             catch (Exception e) {
                 log.error( "Caught exception rolling back the output messenger: " + e, e );
             }
             
             try {
                 getMessenger().rollback();
             }
             catch (Exception e) {
                 log.error( "Caught exception rolling back the input messenger: " + e, e );
             }
         }
     }
 
     
     /**
      * Factory method to create an output message given an input message.
      * Derived classes could override this method to perform any kind of 
      * Message transformation.
      */
     protected Message createOutputMessage(Message inputMessage) throws JMSException {
         Message outputMessage = null;
         
         if ( inputMessage instanceof TextMessage ) {
             outputMessage = createOutputTextMessage( (TextMessage) inputMessage );
         }
         else if ( inputMessage instanceof ObjectMessage ) {
             outputMessage = createOutputObjectMessage( (ObjectMessage) inputMessage );
         }
         else if ( inputMessage instanceof MapMessage ) {
             outputMessage = createOutputMapMessage( (MapMessage) inputMessage );
         }
         else if ( inputMessage instanceof BytesMessage ) {
             outputMessage = createOutputBytesMessage( (BytesMessage) inputMessage );
         }
         else if ( inputMessage instanceof StreamMessage ) {
             outputMessage = createOutputStreamMessage( (StreamMessage) inputMessage );
         }
         else {
             outputMessage = getOutputMessenger().createMessage();
         }
         
         processMessageHeaders(inputMessage, outputMessage);
         
         return outputMessage;
     }
         
     /**
      * Factory method to create ObjectMessage 
      * Derived classes could override this method to perform any kind of 
      * Message transformation.
      */
     protected ObjectMessage createOutputObjectMessage(ObjectMessage inputMessage) throws JMSException {
         return getOutputMessenger().createObjectMessage( inputMessage.getObject() );
     }
     
     /**
      * Factory method to create TextMessage 
      * Derived classes could override this method to perform any kind of 
      * Message transformation.
      */
     protected TextMessage createOutputTextMessage(TextMessage inputMessage) throws JMSException {
         return getOutputMessenger().createTextMessage( inputMessage.getText() );
     }
     
     /**
      * Factory method to create MapMessage 
      * Derived classes could override this method to perform any kind of 
      * Message transformation.
      */
     protected MapMessage createOutputMapMessage(MapMessage inputMessage) throws JMSException {
         MapMessage answer = getOutputMessenger().createMapMessage();
         
         // copy across all values
         for ( Enumeration e = inputMessage.getMapNames(); e.hasMoreElements(); ) {
             String name = (String) e.nextElement();
             Object value = inputMessage.getObject( name );
             answer.setObject( name, value );
         }
         return answer;
     }
     
     /**
      * Factory method to create BytesMessage 
      * Derived classes could override this method to perform any kind of 
      * Message transformation.
      */
     protected BytesMessage createOutputBytesMessage(BytesMessage inputMessage) throws JMSException {
         BytesMessage answer = getOutputMessenger().createBytesMessage();
         
         // copy across all data
         byte[] buffer = new byte[bufferSize];
         while (true ) {
             int size = inputMessage.readBytes( buffer );
             if ( size <= 0 ) {
                 break;
             }
             answer.writeBytes( buffer, 0, size );
             if ( size < bufferSize ) {
                 break;
             }
         }
         return answer;
     }
     
     /**
      * Factory method to create StreamMessage 
      * Derived classes could override this method to perform any kind of 
      * Message transformation.
      */
     protected StreamMessage createOutputStreamMessage(StreamMessage inputMessage) throws JMSException {
         StreamMessage answer = getOutputMessenger().createStreamMessage();
         
         // copy across all data
         byte[] buffer = new byte[bufferSize];
         while (true ) {
             int size = inputMessage.readBytes( buffer );
             if ( size <= 0 ) {
                 break;
             }
             answer.writeBytes( buffer, 0, size );
             if ( size < bufferSize ) {
                 break;
             }
         }
         return answer;
     }
     
     
     
     /**
      * Strategy method to add any headers required on the output message.
      * Derived classes could override this method to perform any kind of 
      * header processing, such as copying the correlation ID, copying all
      * headers or adding some new custom headers etc.
      */
     protected void processMessageHeaders(Message inputMessage, Message outputMessage) throws JMSException {
     }
      
     /**
      * Strategy method to allow different derived classes to acknowledge
      * messages differently, such as to disable acknowledgements
      */
     protected void acknowledge(Message message) throws JMSException {
         message.acknowledge();
     }            
     
     /**
      * Validates that there is a valid output destintation that we can use.
      * Derivations of this class could use multiple destinations
      */
     protected void validateOutputDestination() throws JMSException, ServletException {
         if ( getOutputDestination() == null ) {
             throw new ServletException( "No output Destination is defined for this Bridge" );
         }
     }
 }