JConnection.java

/*
 * JOnAS: Java(TM) Open Application Server
 * Copyright (C) 1999 Bull S.A.
 * Contact: jonas-team@objectweb.org
 * 
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or any later version.
 * 
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307
 * USA
 *
 * --------------------------------------------------------------------------
 * $Id: JConnection.java,v 1.11 2004/04/28 15:31:20 durieuxp Exp $
 * --------------------------------------------------------------------------
 */

package org.objectweb.jonas_jms;

import java.util.LinkedList;

import javax.jms.Connection;
import javax.jms.ConnectionConsumer;
import javax.jms.ConnectionMetaData;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.ServerSessionPool;
import javax.jms.Session;
import javax.jms.Topic;
import javax.jms.XAConnection;
import javax.jms.XAConnectionFactory;

import org.objectweb.transaction.jta.TransactionManager;
import org.objectweb.util.monolog.api.BasicLevel;

public class JConnection implements Connection {

    // The XAConnection used in the MOM
    protected XAConnection xac;

    protected boolean closed;
    protected String user;
    protected boolean globaltx = false;
    protected static TransactionManager tm;
    protected JConnectionFactory jcf;
    protected LinkedList sessionlist = new LinkedList();

    // This constant is used to determine connection
    // with an anonymous user in the pool of JConnection.
    protected static final String INTERNAL_USER_NAME = "anInternalNameUsedOnlyByJOnAS";

    protected JConnection(JConnectionFactory jcf, String user)  throws JMSException {
        this.user = user;
        this.jcf = jcf;
        closed = false;
        if (tm == null) {
            tm = JmsManagerImpl.getTransactionManager();
        }
        // Remember if we are inside a global transaction
        // This is to handle a case that is not expected in JMS specs.
        try {
            globaltx = (tm.getTransaction() != null);
        } catch (Exception e) {
        }
    }


    public JConnection(JConnectionFactory jcf, XAConnectionFactory xacf, String user, String passwd) 
        throws JMSException {
        this(jcf, user);
        // Create the underlaying XAConnection
        xac = xacf.createXAConnection(user, passwd);
    }

    public JConnection(JConnectionFactory jcf, XAConnectionFactory xacf) throws JMSException {
        this(jcf, INTERNAL_USER_NAME);
        // Create the underlaying XAConnection
        xac = xacf.createXAConnection();
    } 

    // -----------------------------------------------------------------------
    // internal methods
    // -----------------------------------------------------------------------

    protected synchronized boolean sessionOpen(Session s) {
        if (!closed) {
            sessionlist.add(s);
            return true;
        } else {
            return false;
        }
    }

    protected synchronized void sessionClose(Session s) {
        sessionlist.remove(s);
        if (sessionlist.size() == 0 && closed) {
            notify();
        }
    }

    public String getUser() {
        return user;
    }

    // -----------------------------------------------------------------------
    // Connection implementation
    // -----------------------------------------------------------------------

    public void close() throws JMSException {
        TraceJms.logger.log(BasicLevel.DEBUG, "");
        if (globaltx) {
            // Connection that was open inside a global transaction.
            // Don't wait (to avoid deadlocks) and don't close it now
            // Since this situation is not expected by the specs, we just
            // pool this connection for now, waiting better...
            jcf.freeJConnection(this);
        } else {
            // Wait for all NON transacted sessions to be finished.
            // LATER: Should rollback first all transacted sessions still running.
            synchronized(this) {
                while (sessionlist.size() > 0) {
                    try {
                        wait();
                    } catch (InterruptedException e) {
                        TraceJms.logger.log(BasicLevel.ERROR, "interrupted");
                    }
                }
            }
            closed = true;
            xac.close();
        }
    }
    public void finalClose() throws JMSException { 
        if (!closed) {
            xac.close();
        }
    }
        
    public ConnectionConsumer createConnectionConsumer(Destination destination,
                                                       java.lang.String messageSelector,
                                                       ServerSessionPool sessionPool,
                                                       int maxMessages)
        throws JMSException {
        TraceJms.logger.log(BasicLevel.DEBUG, "");
        return xac.createConnectionConsumer(destination, 
                                            messageSelector,
                                            sessionPool, 
                                            maxMessages); 
    }

    public ConnectionConsumer createDurableConnectionConsumer(Topic topic,
                                                          java.lang.String subscriptionName,
                                                          java.lang.String messageSelector,
                                                          ServerSessionPool sessionPool,
                                                          int maxMessages)
        throws JMSException {
        TraceJms.logger.log(BasicLevel.DEBUG, "");
        return xac.createDurableConnectionConsumer(topic, 
                                                   subscriptionName,
                                                   messageSelector,
                                                   sessionPool, 
                                                   maxMessages);
    }


    public Session createSession(boolean transacted,int acknowledgeMode) throws JMSException {
        TraceJms.logger.log(BasicLevel.DEBUG, "");
        return new JSession(this, xac);
    }


    public String getClientID() throws JMSException {
        TraceJms.logger.log(BasicLevel.DEBUG, "");
        return xac.getClientID();
    }

    public void setClientID(String clientID) throws JMSException {
        TraceJms.logger.log(BasicLevel.DEBUG, "");
        xac.setClientID(clientID);
    }

    public ConnectionMetaData getMetaData() throws JMSException {
        TraceJms.logger.log(BasicLevel.DEBUG, "");
        return xac.getMetaData();
    }

    public ExceptionListener getExceptionListener() throws JMSException {
        TraceJms.logger.log(BasicLevel.DEBUG, "");
        return xac.getExceptionListener();
    }

    public void setExceptionListener(ExceptionListener listener) throws JMSException {
        TraceJms.logger.log(BasicLevel.DEBUG, "");
        xac.setExceptionListener(listener);
    }

    public void start() throws JMSException {
        TraceJms.logger.log(BasicLevel.DEBUG, "");
        xac.start();
    }

    public void stop() throws JMSException {
        TraceJms.logger.log(BasicLevel.DEBUG, "");
        xac.stop();
    }
}

---