path: root/sca-java-1.x/branches/sca-java-1.6.2/modules/binding-ws-axis2/src/main/java/org/apache/tuscany/sca/binding/ws/axis2/jms/JMSConnectionFactory.java

/*
 * 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.tuscany.sca.binding.ws.axis2.jms;

import java.util.HashMap;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.Map;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.Topic;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NameNotFoundException;
import javax.naming.NamingException;

import org.apache.axis2.addressing.EndpointReference;
import org.apache.axis2.transport.jms.JMSConstants;
import org.apache.axis2.transport.jms.JMSUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.tuscany.sca.work.WorkScheduler;

/**
 * Encapsulate a JMS Connection factory definition within an Axis2.xml
 * <p/>
 * More than one JMS connection factory could be defined within an Axis2 XML
 * specifying the JMSListener as the transportReceiver.
 * <p/>
 * These connection factories are created at the initialization of the
 * transportReceiver, and any service interested in using any of these could
 * specify the name of the factory and the destination through Parameters named
 * JMSConstants.CONFAC_PARAM and JMSConstants.DEST_PARAM as shown below.
 * <p/>
 * <parameter name="transport.jms.ConnectionFactory" locked="true">myQueueConnectionFactory</parameter>
 * <parameter name="transport.jms.Destination" locked="true">TestQueue</parameter>
 * <p/>
 * If a connection factory is defined by a parameter named
 * JMSConstants.DEFAULT_CONFAC_NAME in the Axis2 XML, services which does not
 * explicitly specify a connection factory will be defaulted to it - if it is
 * defined in the Axis2 configuration.
 * <p/>
 * e.g.
 * <transportReceiver name="jms" class="org.apache.axis2.transport.jms.JMSListener">
 * <parameter name="myTopicConnectionFactory" locked="false">
 * <parameter name="java.naming.factory.initial" locked="false">org.apache.activemq.jndi.ActiveMQInitialContextFactory</parameter>
 * <parameter name="java.naming.provider.url" locked="false">tcp://localhost:61616</parameter>
 * <parameter name="transport.jms.ConnectionFactoryJNDIName" locked="false">TopicConnectionFactory</parameter>
 * <parameter name="transport.jms.Destination" locked="false">myTopicOne, myTopicTwo</parameter>
 * </parameter>
 * <parameter name="myQueueConnectionFactory" locked="false">
 * <parameter name="java.naming.factory.initial" locked="false">org.apache.activemq.jndi.ActiveMQInitialContextFactory</parameter>
 * <parameter name="java.naming.provider.url" locked="false">tcp://localhost:61616</parameter>
 * <parameter name="transport.jms.ConnectionFactoryJNDIName" locked="false">QueueConnectionFactory</parameter>
 * <parameter name="transport.jms.Destination" locked="false">myQueueOne, myQueueTwo</parameter>
 * </parameter>
 * <parameter name="default" locked="false">
 * <parameter name="java.naming.factory.initial" locked="false">org.apache.activemq.jndi.ActiveMQInitialContextFactory</parameter>
 * <parameter name="java.naming.provider.url" locked="false">tcp://localhost:61616</parameter>
 * <parameter name="transport.jms.ConnectionFactoryJNDIName" locked="false">ConnectionFactory</parameter>
 * <parameter name="transport.jms.Destination" locked="false">myDestinationOne, myDestinationTwo</parameter>
 * </parameter>
 * </transportReceiver>
 */
public class JMSConnectionFactory {

    private static final Log log = LogFactory.getLog(JMSConnectionFactory.class);

    /**
     * The name used for the connection factory definition within Axis2
     */
    private String name = null;
    /**
     * The JNDI name of the actual connection factory
     */
    private String jndiName = null;
    /**
     * The JNDI name of the actual connection factory username
     */
    private String jndiUser = null;
    /**
     * The JNDI name of the actual connection factory password
     */
    private String jndiPass = null;
    /**
     * Map of destination JNDI names to service names
     */
    private Map serviceJNDINameMapping = null;
    /**
     * Map of destinations to service names
     */
    private Map serviceDestinationMapping = null;
    /**
     * The JMS Sessions listening for messages
     */
    private Map jmsSessions = null;
    /**
     * Properties of the connection factory
     */
    private Hashtable properties = null;
    /**
     * The JNDI Context used
     */
    private Context context = null;
    /**
     * The actual ConnectionFactory instance held within
     */
    private ConnectionFactory conFactory = null;
    /**
     * The JMS Connection is opened.
     */
    private Connection connection = null;
    /**
     * The JMS Message receiver for this connection factory
     */
    private JMSMessageReceiver msgRcvr = null;
    /**
     * The actual password for the connection factory after retrieval from JNDI.
     * If this is not supplied, the OS username will be used by default
     */
    private String user = null;
    /**
     * The actual password for the connection factory after retrieval from JNDI.
     * If this is not supplied, the OS credentials will be used by default
     */
    private String pass = null;
    
    private WorkScheduler workScheduler;
    private boolean consumerRunning;

    /**
     * Create a JMSConnectionFactory for the given Axis2 name and JNDI name
     *
     * @param name     the local Axis2 name of the connection factory
     * @param jndiName the JNDI name of the actual connection factory used
     */
    JMSConnectionFactory(String name, String jndiName, WorkScheduler workScheduler) {
        this.name = name;
        this.jndiName = jndiName;
        serviceJNDINameMapping = new HashMap();
        serviceDestinationMapping = new HashMap();
        properties = new Hashtable();
        jmsSessions = new HashMap();
        this.workScheduler = workScheduler;
    }

    /**
     * Create a JMSConnectionFactory for the given Axis2 name
     *
     * @param name the local Axis2 name of the connection factory
     */
    JMSConnectionFactory(String name, WorkScheduler workScheduler) {
        this(name, null, workScheduler);
    }

    /**
     * Connect to the actual JMS connection factory specified by the JNDI name
     *
     * @throws NamingException if the connection factory cannot be found
     */
    public void connect() throws NamingException {
        if (context == null) {
            createInitialContext();
        }
        conFactory = (ConnectionFactory) context.lookup(jndiName);

        if (jndiUser != null)
                user = (String ) context.lookup(jndiUser);
        
        if (jndiPass != null)
                pass = (String ) context.lookup(jndiPass);
        
        log.debug("Connected to the actual connection factory : " + jndiName);
    }

    /**
     * Creates the initial context using the set properties
     *
     * @throws NamingException
     */
    private void createInitialContext() throws NamingException {
        context = new InitialContext(properties);
    }

    /**
     * Set the JNDI connection factory name
     *
     * @param jndiName
     */
    public void setJndiName(String jndiName) {
        this.jndiName = jndiName;
    }

    /**
     * Get the JNDI name of the actual factory username
     *
     * @return the jndi name of the actual connection factory username
     */
        public void setJndiUser(String jndiUser) {
                this.jndiUser = jndiUser;
        }   
    
    /**
     * Get the JNDI name of the actual factory password
     *
     * @return the jndi name of the actual connection factory password
     */
        public void setJndiPass(String jndiPass) {
                this.jndiPass = jndiPass;
        }

    /**
     * Add a listen destination on this connection factory on behalf of the given service
     *
     * @param destinationJndi destination JNDI name
     * @param serviceName     the service to which it belongs
     */
    public void addDestination(String destinationJndi, String serviceName) {

        serviceJNDINameMapping.put(destinationJndi, serviceName);
        String destinationName = getDestinationName(destinationJndi);

        if (destinationName == null) {
            log.warn("JMS Destination with JNDI name : " + destinationJndi + " does not exist");

            Connection con = null;
            try {
                if ((jndiUser == null) || (jndiPass == null)){
                        // Use the OS username and credentials
                    con = conFactory.createConnection();                        
                } else{
                        // use an explicit username and password
                    con = conFactory.createConnection(user, pass);                                              
                }
                Session session = con.createSession(false, Session.AUTO_ACKNOWLEDGE);
                Queue queue = session.createQueue(destinationJndi);
                destinationName = queue.getQueueName();
                log.warn("JMS Destination with JNDI name : " + destinationJndi + " created");

            } catch (JMSException e) {
                log.error("Unable to create a Destination with JNDI name : " + destinationJndi, e);
                // mark service as faulty
                JMSUtils.markServiceAsFaulty(
                    (String) serviceJNDINameMapping.get(destinationJndi),
                    "Error looking up JMS destination : " + destinationJndi,
                    msgRcvr.getAxisConf().getAxisConfiguration());

            } finally {
                if (con != null) {
                    try {
                        con.close();
                    } catch (JMSException ignore) {}
                }
            }
        }
        
        serviceDestinationMapping.put(destinationName, serviceName);
        log.info("Mapping JNDI name : " + destinationJndi + " and JMS Destination name : " +
            destinationName + " against service : " + serviceName);
    }

    /**
     * Remove listen destination on this connection factory
     *
     * @param destinationJndi the JMS destination to be removed
     * @throws if an error occurs trying to stop listening for messages before removal
     */
    public void removeDestination(String destinationJndi) throws JMSException {
        // find and save provider specific Destination name before we delete
        String providerSpecificDestination = getDestinationName(destinationJndi);
        stoplistenOnDestination(destinationJndi);
        serviceJNDINameMapping.remove(destinationJndi);
        if (providerSpecificDestination != null) {
            serviceDestinationMapping.remove(providerSpecificDestination);
        }
    }

    /**
     * Add a property to the connection factory
     *
     * @param key
     * @param value
     */
    public void addProperty(String key, String value) {
        properties.put(key, value);
    }

    /**
     * Return the name of the connection factory
     *
     * @return the Axis2 name of this factory
     */
    public String getName() {
        return name;
    }

    /**
     * Get the JNDI name of the actual factory
     *
     * @return the jndi name of the actual connection factory
     */
    public String getJndiName() {
        return jndiName;
    }
    
    /**
     * Get the JNDI name of the actual factory username
     *
     * @return the jndi name of the actual connection factory username
     */
        public String getJndiUser() {
                return jndiUser;
        }
    
    /**
     * Get the JNDI name of the actual factory password
     *
     * @return the jndi name of the actual connection factory password
     */
    public String getJndiPass() {
                return jndiPass;
        }
  
    
    /**
     * This is the real password for the connection factory after the JNDI lookup
     *
     * @return the real password for the connection factory after the JNDI lookup
     */
        public String getPass() {
                return pass;
        }

    /**
     * This is the real username for the connection factory after the JNDI lookup
     *
     * @return the eal username for the connection factory after the JNDI lookup
     */
        public String getUser() {
                return user;
        }

        /**
     * Get the actual underlying connection factory
     *
     * @return actual connection factory
     */
    public ConnectionFactory getConFactory() {
        return conFactory;
    }

    /**
     * Get the list of destinations (JNDI) associated with this connection factory
     *
     * @return destinations to service maping
     */
    public Map getDestinations() {
        return serviceJNDINameMapping;
    }

    /**
     * Get the connection factory properties
     *
     * @return properties
     */
    public Hashtable getProperties() {
        return properties;
    }

    /**
     * Begin listening for messages on the list of destinations associated
     * with this connection factory. (Called during Axis2 initialization of
     * the Transport receivers)
     *
     * @param msgRcvr the message receiver which will process received messages
     * @throws JMSException on exceptions
     */
    public void listen(JMSMessageReceiver msgRcvr) throws JMSException {

        // save a reference to the message receiver
        this.msgRcvr = msgRcvr;

        log.debug("Connection factory : " + name + " initializing...");

        if (conFactory == null || context == null) {
            handleException(
                    "Connection factory must be 'connected' before listening");
        } else {
            try {
                if ((jndiUser == null) || (jndiPass == null)){
                        // User the OS username and credentials
                    connection = conFactory.createConnection();                         
                } else{
                        // use an explicit username and password
                        connection = conFactory.createConnection(user, pass);                                                   
                }
            } catch (JMSException e) {
                handleException("Error creating a JMS connection using the " +
                        "factory : " + jndiName, e);
            }
        }

        Iterator iter = serviceJNDINameMapping.keySet().iterator();
        while (iter.hasNext()) {
            String destinationJndi = (String) iter.next();
            listenOnDestination(destinationJndi);
        }

        // start the connection
        if (!consumerRunning) {
            connection.start();
        }
        log.info("Connection factory : " + name + " initialized...");
    }

    /**
     * Listen on the given destination from this connection factory. Used to
     * start listening on a destination associated with a newly deployed service
     *
     * @param destinationJndi the JMS destination to listen on
     * @throws JMSException on exception
     */
    public void listenOnDestination(String destinationJndi) throws JMSException {
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Destination destination = null;
        try {
            Object o = context.lookup(destinationJndi);
            destination = (Destination) o;

        } catch (NameNotFoundException e) {
            log.warn("Cannot find destination : " + destinationJndi +
                    " Creating a Queue with this name");
            destination = session.createQueue(destinationJndi);

        } catch (NamingException e) {
            log.warn("Error looking up destination : " + destinationJndi, e);
            // mark service as faulty
            JMSUtils.markServiceAsFaulty(
                    (String) serviceJNDINameMapping.get(destinationJndi),
                    "Error looking up JMS destination : " + destinationJndi,
                    this.msgRcvr.getAxisConf().getAxisConfiguration());
        }

        MessageConsumer consumer = session.createConsumer(destination);
//        consumer.setMessageListener(this.msgRcvr); replace with new Tuscany method:
        registerMessageReceiver(consumer, this.msgRcvr);
        jmsSessions.put(destinationJndi, session);
    }

    private void registerMessageReceiver(final MessageConsumer consumer, final JMSMessageReceiver messageReceiver) throws JMSException {

        try {

            consumer.setMessageListener(messageReceiver);

        } catch (javax.jms.JMSException e) {

            // setMessageListener not allowed in JEE container so use Tuscany threads

            connection.start();
            consumerRunning = true;
            
            workScheduler.scheduleWork(new Runnable() {

                public void run() {
                    try {
                        while (consumerRunning) {
                            final Message msg = consumer.receive();
                            if (msg != null) {
                                workScheduler.scheduleWork(new Runnable() {
                                    public void run() {
                                        try {
                                            messageReceiver.onMessage(msg);
                                        } catch (Exception e) {
                                            log.error("Exception on message receiver thread", e);
                                        }
                                    }
                                });
                            }
                        }
                    } catch (Exception e) {
                        log.error("Exception on consumer receive thread", e);
                    }
                }
            });
        }
    }

    /**
     * Stop listening on the given destination - for undeployment of services
     *
     * @param destinationJndi the JNDI name of the JMS destination
     * @throws JMSException on exception
     */
    private void stoplistenOnDestination(String destinationJndi) throws JMSException {
        ((Session) jmsSessions.get(destinationJndi)).close();
    }

    /**
     * Return the service name using this destination
     *
     * @param destination the destination name
     * @return the service which uses the given destination, or null
     */
    public String getServiceNameForDestination(String destination) {

        return (String) serviceJNDINameMapping.get(destination);
    }

    /**
     * Close all connections, sessions etc.. and stop this connection factory
     */
    public void stop() {
        try {
            consumerRunning = false;
            connection.close();
        } catch (JMSException e) {
            log.warn("Error shutting down connection factory : " + name, e);
        }
    }

    /**
     * Return the provider specific Destination name if any for the destination with the given
     * JNDI name
     * @param destinationJndi the JNDI name of the destination
     * @return the provider specific Destination name or null if cannot be found
     */
    public String getDestinationName(String destinationJndi) {
        try {
            Destination destination = (Destination) context.lookup(destinationJndi);
            if (destination != null && destination instanceof Queue) {
                return ((Queue) destination).getQueueName();
            } else if (destination != null && destination instanceof Topic) {
                return ((Topic) destination).getTopicName();
            }
        } catch (JMSException e) {
            log.warn("Error reading provider specific JMS destination name for destination " +
                "with JNDI name : " + destinationJndi, e);
        } catch (NamingException e) {
            log.warn("Error looking up destination with JNDI name : " + destinationJndi +
                " to map its corresponding provider specific Destination name");
        }
        return null;
    }

    /**
     * Return the EPR for the JMS Destination with the given JNDI name and using
     * this connection factory
     * @param destination the JNDI name of the JMS Destionation
     * @return the EPR
     */
    public EndpointReference getEPRForDestination(String destination) {

        StringBuffer sb = new StringBuffer();
        sb.append(JMSConstants.JMS_PREFIX).append(destination);
        sb.append("?").append(JMSConstants.CONFAC_JNDI_NAME_PARAM).
                append("=").append(getJndiName());
        Iterator props = getProperties().keySet().iterator();
        while (props.hasNext()) {
            String key = (String) props.next();
            String value = (String) getProperties().get(key);
            sb.append("&").append(key).append("=").append(value);
        }

        return new EndpointReference(sb.toString());
    }

    public String getServiceByDestination(String destinationName) {
        return (String) serviceDestinationMapping.get(destinationName);
    }

    private void handleException(String msg) throws AxisJMSException {
        log.error(msg);
        throw new AxisJMSException(msg);
    }

    private void handleException(String msg, Exception e) throws AxisJMSException {
        log.error(msg, e);
        throw new AxisJMSException(msg, e);
    }
}