jmsclient.java

CRM源码This file describes some issues that should be implemented in future and how it should be imple

/*
 * Copyright 2006-2007 Queplix Corp.
 *
 * Licensed under the Queplix Public License, Version 1.1.1 (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.queplix.com/solutions/commercial-open-source/queplix-public-license/
 *
 * 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.queplix.core.utils.async;

import com.queplix.core.error.GenericSystemException;
import com.queplix.core.utils.log.AbstractLogger;

import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.ObjectMessage;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueSender;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import java.io.Serializable;

/**
 * <p>Abstract class for for sending message to JMS server</p>
 * @author Baranov Andrey [ALB]
 * @version $Revision: 1.1.1.1 $ $Date: 2005/09/12 15:31:18 $
 */

public abstract class JMSClient
    extends AbstractLogger
    implements java.io.Serializable {

    // variables
    private final String queueConnName;
    private final String queueName;

    private transient Queue queue;
    private transient QueueConnectionFactory factory;

    //
    // Constructor
    //
    protected JMSClient( String queueConnName, String queueName ) {
        this.queueConnName = queueConnName;
        this.queueName = queueName;
    }

    //
    // Send message
    //
    protected void send( Serializable message ) {
        send( false, message );
    }

    //
    // Send message
    //
    protected void send( boolean transacted, Serializable message ) {
        send( transacted, DeliveryMode.NON_PERSISTENT, 4, 0, message );
    }

    //
    // Send message
    //
    protected void send( boolean transacted,
                         int deliveryMode,
                         int priority,
                         long timeToLive,
                         Serializable message ) {

        QueueConnection connection = null;
        QueueSession session = null;

        try {
            // get connection factory and queue
            QueueConnectionFactory factory = getConnectionFactory();
            Queue queue = getQueue();

            // create connection
            connection = factory.createQueueConnection();

            // create session and sender
            session = connection.createQueueSession( transacted, Session.AUTO_ACKNOWLEDGE );
            QueueSender sender = session.createSender( queue );

            if( getLogger().isDebugEnabled() ) {
                DEBUG( "============================" );
                DEBUG( "Sender message:" );
                DEBUG( "	Client: " + getClass().getName() );
                DEBUG( "	Queue: " + queueName );
                DEBUG( "	Transacted: " + session.getTransacted() );
                DEBUG( "	Message: " + message );
                DEBUG( "============================" );
            }

            // create message
            ObjectMessage msg = session.createObjectMessage();
            msg.setObject( message );

            // send
            sender.send( msg, deliveryMode, priority, timeToLive );

        } catch( JMSException ex ) {
            ex.printStackTrace();
            destroy();
            throw new GenericSystemException( "JMS exception: " + ex.getMessage(), ex );

        } catch( NamingException ex ) {
            ex.printStackTrace();
            destroy();
            throw new GenericSystemException( "Naming exception: " + ex.getMessage(), ex );

        } catch( Exception ex ) {
            ex.printStackTrace();
            destroy();
            throw new GenericSystemException( "Unknown exception: " + ex.getMessage(), ex );

        } finally {
            if( session != null ) {
                try {
                    session.close();
                } catch( JMSException e ) {}
            }
            if( connection != null ) {
                try {
                    connection.close();
                } catch( JMSException e ) {}
            }
        }
    }

    //
    // Get JMS connection factory
    //
    private synchronized QueueConnectionFactory getConnectionFactory()
        throws JMSException, NamingException {

        if( factory == null ) {
            Context ctx = new InitialContext();
            factory = ( QueueConnectionFactory ) ctx.lookup( queueConnName );
        }

        return factory;
    }

    //
    // Get JMS queue
    //
    private synchronized Queue getQueue()
        throws JMSException, NamingException {

        if( queue == null ) {
            Context ctx = new InitialContext();
            queue = ( Queue ) ctx.lookup( queueName );
        }

        return queue;
    }

    //
    // Destroy client
    //
    private synchronized void destroy() {
        factory = null;
        queue = null;
    }
}