View Javadoc

    1   /***
    2    *
    3    * Copyright 2004 Protique Ltd
    4    *
    5    * Licensed under the Apache License, Version 2.0 (the "License");
    6    * you may not use this file except in compliance with the License.
    7    * You may obtain a copy of the License at
    8    *
    9    * http://www.apache.org/licenses/LICENSE-2.0
    10   *
    11   * Unless required by applicable law or agreed to in writing, software
    12   * distributed under the License is distributed on an "AS IS" BASIS,
    13   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    14   * See the License for the specific language governing permissions and
    15   * limitations under the License.
    16   *
    17   **/
    18
    19  package org.codehaus.activemq;
    20
    21  import EDU.oswego.cs.dl.util.concurrent.CopyOnWriteArrayList;
    22  import EDU.oswego.cs.dl.util.concurrent.SynchronizedBoolean;
    23  import EDU.oswego.cs.dl.util.concurrent.SynchronizedInt;
    24  import org.apache.commons.logging.Log;
    25  import org.apache.commons.logging.LogFactory;
    26  import org.codehaus.activemq.capacity.CapacityMonitorEvent;
    27  import org.codehaus.activemq.capacity.CapacityMonitorEventListener;
    28  import org.codehaus.activemq.management.JMSConnectionStatsImpl;
    29  import org.codehaus.activemq.management.JMSStatsImpl;
    30  import org.codehaus.activemq.management.StatsCapable;
    31  import org.codehaus.activemq.message.ActiveMQMessage;
    32  import org.codehaus.activemq.message.CapacityInfo;
    33  import org.codehaus.activemq.message.ConnectionInfo;
    34  import org.codehaus.activemq.message.ConsumerInfo;
    35  import org.codehaus.activemq.message.Packet;
    36  import org.codehaus.activemq.message.PacketListener;
    37  import org.codehaus.activemq.message.ProducerInfo;
    38  import org.codehaus.activemq.message.Receipt;
    39  import org.codehaus.activemq.message.SessionInfo;
    40  import org.codehaus.activemq.message.util.MemoryBoundedQueue;
    41  import org.codehaus.activemq.message.util.MemoryBoundedQueueManager;
    42  import org.codehaus.activemq.transport.TransportChannel;
    43  import org.codehaus.activemq.transport.TransportStatusEvent;
    44  import org.codehaus.activemq.transport.TransportStatusEventListener;
    45  import org.codehaus.activemq.util.IdGenerator;
    46
    47  import javax.jms.*;
    48  import javax.jms.IllegalStateException;
    49  import javax.management.j2ee.statistics.Stats;
    50  import java.util.Iterator;
    51
    52  /***
    53   * A <CODE>Connection</CODE> object is a client's active connection to its JMS provider. It typically allocates
    54   * provider resources outside the Java virtual machine (JVM).
    55   * <P>
    56   * Connections support concurrent use.
    57   * <P>
    58   * A connection serves several purposes:
    59   * <UL>
    60   * <LI>It encapsulates an open connection with a JMS provider. It typically represents an open TCP/IP socket between a
    61   * client and the service provider software.
    62   * <LI>Its creation is where client authentication takes place.
    63   * <LI>It can specify a unique client identifier.
    64   * <LI>It provides a <CODE>ConnectionMetaData</CODE> object.
    65   * <LI>It supports an optional <CODE>ExceptionListener</CODE> object.
    66   * </UL>
    67   * <P>
    68   * Because the creation of a connection involves setting up authentication and communication, a connection is a
    69   * relatively heavyweight object. Most clients will do all their messaging with a single connection. Other more advanced
    70   * applications may use several connections. The JMS API does not architect a reason for using multiple connections;
    71   * however, there may be operational reasons for doing so.
    72   * <P>
    73   * A JMS client typically creates a connection, one or more sessions, and a number of message producers and consumers.
    74   * When a connection is created, it is in stopped mode. That means that no messages are being delivered.
    75   * <P>
    76   * It is typical to leave the connection in stopped mode until setup is complete (that is, until all message consumers
    77   * have been created). At that point, the client calls the connection's <CODE>start</CODE> method, and messages begin
    78   * arriving at the connection's consumers. This setup convention minimizes any client confusion that may result from
    79   * asynchronous message delivery while the client is still in the process of setting itself up.
    80   * <P>
    81   * A connection can be started immediately, and the setup can be done afterwards. Clients that do this must be prepared
    82   * to handle asynchronous message delivery while they are still in the process of setting up.
    83   * <P>
    84   * A message producer can send messages while a connection is stopped. <p/>This class is also a <CODE>TopicConnection
    85   * </CODE>. A <CODE>TopicConnection</CODE> object is an active connection to a publish/subscribe JMS provider. A
    86   * client uses a <CODE>TopicConnection</CODE> object to create one or more <CODE>TopicSession</CODE> objects for
    87   * producing and consuming messages.
    88   * <P>
    89   * A <CODE>TopicConnection</CODE> can be used to create a <CODE>TopicSession</CODE>, from which specialized
    90   * topic-related objects can be created. A more general, and recommended approach is to use the <CODE>Connection
    91   * </CODE> object.
    92   * <P>
    93   * <p/><p/>This class is also a <CODE>QueueConnection</CODE>. A A <CODE>QueueConnection</CODE> object is an active
    94   * connection to a point-to-point JMS provider. A client uses a <CODE>QueueConnection</CODE> object to create one or
    95   * more <CODE>QueueSession</CODE> objects for producing and consuming messages.
    96   * <P>
    97   * A <CODE>QueueConnection</CODE> can be used to create a <CODE>QueueSession</CODE>, from which specialized
    98   * queue-related objects can be created. A more general, and recommended, approach is to use the <CODE>Connection
    99   * </CODE> object.
    100  * <P>
    101  * A <CODE>QueueConnection</CODE> cannot be used to create objects specific to the publish/subscribe domain. The
    102  * <CODE>createDurableConnectionConsumer</CODE> method inherits from <CODE>Connection</CODE>, but must throw an
    103  * <CODE>IllegalStateException</CODE> if used from <CODE>QueueConnection</CODE>. // *
    104  *
    105  * @version $Revision: 1.50 $
    106  * @see javax.jms.Connection
    107  * @see javax.jms.ConnectionFactory
    108  * @see javax.jms.QueueConnection
    109  * @see javax.jms.TopicConnection
    110  * @see javax.jms.TopicConnectionFactory
    111  * @see javax.jms.QueueConnection
    112  * @see javax.jms.QueueConnectionFactory
    113  */
    114 public class ActiveMQConnection
    115         implements
    116         Connection,
    117         PacketListener,
    118         ExceptionListener,
    119         TopicConnection,
    120         QueueConnection,
    121         StatsCapable,
    122         CapacityMonitorEventListener,
    123         TransportStatusEventListener {
    124     private static final Log log = LogFactory.getLog(ActiveMQConnection.class);
    125     private static final int DEFAULT_CONNECTION_MEMORY_LIMIT = 10 * 1024 * 1024;
    126     private ActiveMQConnectionFactory factory;
    127     private String userName;
    128     private String password;
    129     protected String clientID;
    130     protected IdGenerator consumerIdGenerator;
    131     private IdGenerator clientIdGenerator;
    132     protected IdGenerator packetIdGenerator;
    133     private IdGenerator sessionIdGenerator;
    134     private TransportChannel transportChannel;
    135     private ExceptionListener exceptionListener;
    136     private CopyOnWriteArrayList sessions;
    137     private CopyOnWriteArrayList messageDispatchers;
    138     private CopyOnWriteArrayList connectionConsumers;
    139     private SynchronizedInt consumerNumberGenerator;
    140     private ActiveMQConnectionMetaData connectionMetaData;
    141     private SynchronizedBoolean closed;
    142     private SynchronizedBoolean started;
    143     private boolean clientIDSet;
    144     private boolean isConnectionInfoSentToBroker;
    145     private boolean isTransportOK;
    146     private long startTime;
    147     private ActiveMQPrefetchPolicy prefetchPolicy;
    148     private JMSConnectionStatsImpl stats;
    149     private JMSStatsImpl factoryStats;
    150     private MemoryBoundedQueueManager boundedQueueManager;
    151     private long flowControlSleepTime = 0;
    152     /***
    153      * Default UserName for the Connection
    154      */
    155     public static final String DEFAULT_USER = "defaultUser";
    156     /***
    157      * Default URL for the ActiveMQ Broker
    158      */
    159     public static final String DEFAULT_URL = "tcp://localhost:61616";
    160     /***
    161      * Default Password for the Connection
    162      */
    163     public static final String DEFAULT_PASSWORD = "defaultPassword";
    164     private boolean userSpecifiedClientID;
    165
    166     /***
    167      * A static helper method to create a new connection
    168      *
    169      * @return an ActiveMQConnection
    170      * @throws JMSException
    171      */
    172     public static ActiveMQConnection makeConnection() throws JMSException {
    173         ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory();
    174         return (ActiveMQConnection) factory.createConnection();
    175     }
    176
    177     /***
    178      * A static helper method to create a new connection
    179      *
    180      * @param uri
    181      * @return and ActiveMQConnection
    182      * @throws JMSException
    183      */
    184     public static ActiveMQConnection makeConnection(String uri) throws JMSException {
    185         ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(uri);
    186         return (ActiveMQConnection) factory.createConnection();
    187     }
    188
    189     /***
    190      * A static helper method to create a new connection
    191      *
    192      * @param user
    193      * @param password
    194      * @param uri
    195      * @return an ActiveMQConnection
    196      * @throws JMSException
    197      */
    198     public static ActiveMQConnection makeConnection(String user, String password, String uri) throws JMSException {
    199         ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(user, password, uri);
    200         return (ActiveMQConnection) factory.createConnection();
    201     }
    202
    203     /***
    204      * Constructs a connection from an existing TransportChannel and user/password.
    205      *
    206      * @param factory
    207      * @param theUserName      the users name
    208      * @param thePassword      the password
    209      * @param transportChannel the transport channel to communicate with the server
    210      * @throws JMSException
    211      */
    212     public ActiveMQConnection(ActiveMQConnectionFactory factory, String theUserName, String thePassword,
    213                               TransportChannel transportChannel) throws JMSException {
    214         this(factory, theUserName, thePassword);
    215         this.transportChannel = transportChannel;
    216         this.transportChannel.setPacketListener(this);
    217         this.transportChannel.setExceptionListener(this);
    218         this.transportChannel.addTransportStatusEventListener(this);
    219         this.transportChannel.start();
    220         this.isTransportOK = true;
    221     }
    222
    223     protected ActiveMQConnection(ActiveMQConnectionFactory factory, String theUserName, String thePassword) {
    224         this.factory = factory;
    225         this.userName = theUserName;
    226         this.password = thePassword;
    227         this.clientIdGenerator = new IdGenerator();
    228         this.packetIdGenerator = new IdGenerator();
    229         this.consumerIdGenerator = new IdGenerator();
    230         this.sessionIdGenerator = new IdGenerator();
    231         this.consumerNumberGenerator = new SynchronizedInt(0);
    232         this.sessions = new CopyOnWriteArrayList();
    233         this.messageDispatchers = new CopyOnWriteArrayList();
    234         this.connectionConsumers = new CopyOnWriteArrayList();
    235         this.connectionMetaData = new ActiveMQConnectionMetaData();
    236         this.closed = new SynchronizedBoolean(false);
    237         this.started = new SynchronizedBoolean(false);
    238         this.startTime = System.currentTimeMillis();
    239         this.prefetchPolicy = new ActiveMQPrefetchPolicy();
    240         this.boundedQueueManager = new MemoryBoundedQueueManager(clientID, DEFAULT_CONNECTION_MEMORY_LIMIT);
    241         this.boundedQueueManager.addCapacityEventListener(this);
    242         boolean transactional = this instanceof XAConnection;
    243         factoryStats = factory.getFactoryStats();
    244         factoryStats.addConnection(this);
    245         stats = new JMSConnectionStatsImpl(sessions, transactional);
    246         factory.onConnectionCreate(this);
    247     }
    248
    249     /***
    250      * @return statistics for this Connection
    251      */
    252     public Stats getStats() {
    253         return stats;
    254     }
    255
    256     /***
    257      * @return a number unique for this connection
    258      */
    259     public JMSConnectionStatsImpl getConnectionStats() {
    260         return stats;
    261     }
    262
    263     /***
    264      * Creates a <CODE>Session</CODE> object.
    265      *
    266      * @param transacted      indicates whether the session is transacted
    267      * @param acknowledgeMode indicates whether the consumer or the client will acknowledge any messages it receives;
    268      *                        ignored if the session is transacted. Legal values are <code>Session.AUTO_ACKNOWLEDGE</code>,
    269      *                        <code>Session.CLIENT_ACKNOWLEDGE</code>, and <code>Session.DUPS_OK_ACKNOWLEDGE</code>.
    270      * @return a newly created session
    271      * @throws JMSException if the <CODE>Connection</CODE> object fails to create a session due to some internal error
    272      *                      or lack of support for the specific transaction and acknowledgement mode.
    273      * @see Session#AUTO_ACKNOWLEDGE
    274      * @see Session#CLIENT_ACKNOWLEDGE
    275      * @see Session#DUPS_OK_ACKNOWLEDGE
    276      * @since 1.1
    277      */
    278     public Session createSession(boolean transacted, int acknowledgeMode) throws JMSException {
    279         checkClosed();
    280         ensureClientIDInitialised();
    281         return new ActiveMQSession(this, (transacted ? Session.SESSION_TRANSACTED : acknowledgeMode));
    282     }
    283
    284     /***
    285      * Gets the client identifier for this connection.
    286      * <P>
    287      * This value is specific to the JMS provider. It is either preconfigured by an administrator in a <CODE>
    288      * ConnectionFactory</CODE> object or assigned dynamically by the application by calling the
    289      * <code>setClientID</code> method.
    290      *
    291      * @return the unique client identifier
    292      * @throws JMSException if the JMS provider fails to return the client ID for this connection due to some internal
    293      *                      error.
    294      */
    295     public String getClientID() throws JMSException {
    296         checkClosed();
    297         return this.clientID;
    298     }
    299
    300     /***
    301      * Sets the client identifier for this connection.
    302      * <P>
    303      * The preferred way to assign a JMS client's client identifier is for it to be configured in a client-specific
    304      * <CODE>ConnectionFactory</CODE> object and transparently assigned to the <CODE>Connection</CODE> object it
    305      * creates.
    306      * <P>
    307      * Alternatively, a client can set a connection's client identifier using a provider-specific value. The facility to
    308      * set a connection's client identifier explicitly is not a mechanism for overriding the identifier that has been
    309      * administratively configured. It is provided for the case where no administratively specified identifier exists.
    310      * If one does exist, an attempt to change it by setting it must throw an <CODE>IllegalStateException</CODE>. If
    311      * a client sets the client identifier explicitly, it must do so immediately after it creates the connection and
    312      * before any other action on the connection is taken. After this point, setting the client identifier is a
    313      * programming error that should throw an <CODE>IllegalStateException</CODE>.
    314      * <P>
    315      * The purpose of the client identifier is to associate a connection and its objects with a state maintained on
    316      * behalf of the client by a provider. The only such state identified by the JMS API is that required to support
    317      * durable subscriptions.
    318      * <P>
    319      * If another connection with the same <code>clientID</code> is already running when this method is called, the
    320      * JMS provider should detect the duplicate ID and throw an <CODE>InvalidClientIDException</CODE>.
    321      *
    322      * @param newClientID the unique client identifier
    323      * @throws JMSException if the JMS provider fails to set the client ID for this connection due to some internal
    324      *                      error.
    325      * @throws javax.jms.InvalidClientIDException
    326      *                      if the JMS client specifies an invalid or duplicate client ID.
    327      * @throws javax.jms.IllegalStateException
    328      *                      if the JMS client attempts to set a connection's client ID at the wrong
    329      *                      time or when it has been administratively configured.
    330      */
    331     public void setClientID(String newClientID) throws JMSException {
    332         if (this.clientIDSet) {
    333             throw new IllegalStateException("The clientID has already been set");
    334         }
    335         if (this.isConnectionInfoSentToBroker) {
    336             throw new IllegalStateException("Setting clientID on a used Connection is not allowed");
    337         }
    338         checkClosed();
    339         this.clientID = newClientID;
    340         this.userSpecifiedClientID = true;
    341     }
    342
    343     /***
    344      * Gets the metadata for this connection.
    345      *
    346      * @return the connection metadata
    347      * @throws JMSException if the JMS provider fails to get the connection metadata for this connection.
    348      * @see javax.jms.ConnectionMetaData
    349      */
    350     public ConnectionMetaData getMetaData() throws JMSException {
    351         checkClosed();
    352         return this.connectionMetaData;
    353     }
    354
    355     /***
    356      * Gets the <CODE>ExceptionListener</CODE> object for this connection. Not every <CODE>Connection</CODE> has an
    357      * <CODE>ExceptionListener</CODE> associated with it.
    358      *
    359      * @return the <CODE>ExceptionListener</CODE> for this connection, or null. if no <CODE>ExceptionListener</CODE>
    360      *         is associated with this connection.
    361      * @throws JMSException if the JMS provider fails to get the <CODE>ExceptionListener</CODE> for this connection.
    362      * @see javax.jms.Connection#setExceptionListener(ExceptionListener)
    363      */
    364     public ExceptionListener getExceptionListener() throws JMSException {
    365         checkClosed();
    366         return this.exceptionListener;
    367     }
    368
    369     /***
    370      * Sets an exception listener for this connection.
    371      * <P>
    372      * If a JMS provider detects a serious problem with a connection, it informs the connection's <CODE>
    373      * ExceptionListener</CODE>, if one has been registered. It does this by calling the listener's <CODE>onException
    374      * </CODE> method, passing it a <CODE>JMSException</CODE> object describing the problem.
    375      * <P>
    376      * An exception listener allows a client to be notified of a problem asynchronously. Some connections only consume
    377      * messages, so they would have no other way to learn their connection has failed.
    378      * <P>
    379      * A connection serializes execution of its <CODE>ExceptionListener</CODE>.
    380      * <P>
    381      * A JMS provider should attempt to resolve connection problems itself before it notifies the client of them.
    382      *
    383      * @param listener the exception listener
    384      * @throws JMSException if the JMS provider fails to set the exception listener for this connection.
    385      */
    386     public void setExceptionListener(ExceptionListener listener) throws JMSException {
    387         checkClosed();
    388         this.exceptionListener = listener;
    389         this.transportChannel.setExceptionListener(listener);
    390     }
    391
    392     /***
    393      * Starts (or restarts) a connection's delivery of incoming messages. A call to <CODE>start</CODE> on a connection
    394      * that has already been started is ignored.
    395      *
    396      * @throws JMSException if the JMS provider fails to start message delivery due to some internal error.
    397      * @see javax.jms.Connection#stop()
    398      */
    399     public void start() throws JMSException {
    400         checkClosed();
    401         if (started.commit(false, true)) {
    402             sendConnectionInfoToBroker();
    403             this.transportChannel.start();
    404             for (Iterator i = sessions.iterator(); i.hasNext();) {
    405                 ActiveMQSession s = (ActiveMQSession) i.next();
    406                 s.start();
    407             }
    408         }
    409     }
    410
    411     /***
    412      * @return true if this Connection is started
    413      */
    414     protected boolean isStarted() {
    415         return started.get();
    416     }
    417
    418     /***
    419      * Temporarily stops a connection's delivery of incoming messages. Delivery can be restarted using the connection's
    420      * <CODE>start</CODE> method. When the connection is stopped, delivery to all the connection's message consumers
    421      * is inhibited: synchronous receives block, and messages are not delivered to message listeners.
    422      * <P>
    423      * This call blocks until receives and/or message listeners in progress have completed.
    424      * <P>
    425      * Stopping a connection has no effect on its ability to send messages. A call to <CODE>stop</CODE> on a
    426      * connection that has already been stopped is ignored.
    427      * <P>
    428      * A call to <CODE>stop</CODE> must not return until delivery of messages has paused. This means that a client can
    429      * rely on the fact that none of its message listeners will be called and that all threads of control waiting for
    430      * <CODE>receive</CODE> calls to return will not return with a message until the connection is restarted. The
    431      * receive timers for a stopped connection continue to advance, so receives may time out while the connection is
    432      * stopped.
    433      * <P>
    434      * If message listeners are running when <CODE>stop</CODE> is invoked, the <CODE>stop</CODE> call must wait
    435      * until all of them have returned before it may return. While these message listeners are completing, they must
    436      * have the full services of the connection available to them.
    437      *
    438      * @throws JMSException if the JMS provider fails to stop message delivery due to some internal error.
    439      * @see javax.jms.Connection#start()
    440      */
    441     public void stop() throws JMSException {
    442         checkClosed();
    443         if (started.commit(true, false)) {
    444             for (Iterator i = sessions.iterator(); i.hasNext();) {
    445                 ActiveMQSession s = (ActiveMQSession) i.next();
    446                 s.stop();
    447             }
    448             sendConnectionInfoToBroker(2000, closed.get());
    449             transportChannel.stop();
    450         }
    451     }
    452
    453     /***
    454      * Closes the connection.
    455      * <P>
    456      * Since a provider typically allocates significant resources outside the JVM on behalf of a connection, clients
    457      * should close these resources when they are not needed. Relying on garbage collection to eventually reclaim these
    458      * resources may not be timely enough.
    459      * <P>
    460      * There is no need to close the sessions, producers, and consumers of a closed connection.
    461      * <P>
    462      * Closing a connection causes all temporary destinations to be deleted.
    463      * <P>
    464      * When this method is invoked, it should not return until message processing has been shut down in an orderly
    465      * fashion. This means that all message listeners that may have been running have returned, and that all pending
    466      * receives have returned. A close terminates all pending message receives on the connection's sessions' consumers.
    467      * The receives may return with a message or with null, depending on whether there was a message available at the
    468      * time of the close. If one or more of the connection's sessions' message listeners is processing a message at the
    469      * time when connection <CODE>close</CODE> is invoked, all the facilities of the connection and its sessions must
    470      * remain available to those listeners until they return control to the JMS provider.
    471      * <P>
    472      * Closing a connection causes any of its sessions' transactions in progress to be rolled back. In the case where a
    473      * session's work is coordinated by an external transaction manager, a session's <CODE>commit</CODE> and <CODE>
    474      * rollback</CODE> methods are not used and the result of a closed session's work is determined later by the
    475      * transaction manager. Closing a connection does NOT force an acknowledgment of client-acknowledged sessions.
    476      * <P>
    477      * Invoking the <CODE>acknowledge</CODE> method of a received message from a closed connection's session must
    478      * throw an <CODE>IllegalStateException</CODE>. Closing a closed connection must NOT throw an exception.
    479      *
    480      * @throws JMSException if the JMS provider fails to close the connection due to some internal error. For example, a
    481      *                      failure to release resources or to close a socket connection can cause this exception to be thrown.
    482      */
    483     public synchronized void close() throws JMSException {
    484         if (!closed.get()) {
    485             boundedQueueManager.removeCapacityEventListener(this);
    486             try {
    487                 for (Iterator i = this.sessions.iterator(); i.hasNext();) {
    488                     ActiveMQSession s = (ActiveMQSession) i.next();
    489                     s.close();
    490                 }
    491                 for (Iterator i = this.connectionConsumers.iterator(); i.hasNext();) {
    492                     ActiveMQConnectionConsumer c = (ActiveMQConnectionConsumer) i.next();
    493                     c.close();
    494                 }
    495                 sendConnectionInfoToBroker(2000, true);
    496                 this.connectionConsumers.clear();
    497                 this.messageDispatchers.clear();
    498                 this.transportChannel.stop();
    499             }
    500             finally {
    501                 this.sessions.clear();
    502                 started.set(false);
    503                 factory.onConnectionClose(this);
    504             }
    505             closed.set(true);
    506         }
    507     }
    508
    509     /***
    510      * simply throws an exception if the Connection is already closed
    511      *
    512      * @throws IllegalStateException
    513      */
    514     protected void checkClosed() throws IllegalStateException {
    515         if (this.closed.get()) {
    516             throw new IllegalStateException("The Connection is closed");
    517         }
    518     }
    519
    520     /***
    521      * Creates a connection consumer for this connection (optional operation). This is an expert facility not used by
    522      * regular JMS clients.
    523      *
    524      * @param destination     the destination to access
    525      * @param messageSelector only messages with properties matching the message selector expression are delivered. A
    526      *                        value of null or an empty string indicates that there is no message selector for the message consumer.
    527      * @param sessionPool     the server session pool to associate with this connection consumer
    528      * @param maxMessages     the maximum number of messages that can be assigned to a server session at one time
    529      * @return the connection consumer
    530      * @throws JMSException if the <CODE>Connection</CODE> object fails to create a connection consumer due to some
    531      *                      internal error or invalid arguments for <CODE>sessionPool</CODE> and <CODE>messageSelector</CODE>.
    532      * @throws javax.jms.InvalidDestinationException
    533      *                      if an invalid destination is specified.
    534      * @throws javax.jms.InvalidSelectorException
    535      *                      if the message selector is invalid.
    536      * @see javax.jms.ConnectionConsumer
    537      * @since 1.1
    538      */
    539     public ConnectionConsumer createConnectionConsumer(Destination destination, String messageSelector,
    540                                                        ServerSessionPool sessionPool, int maxMessages) throws JMSException {
    541         checkClosed();
    542         ConsumerInfo info = new ConsumerInfo();
    543         info.setId(this.packetIdGenerator.generateId());
    544         info.setConsumerId(consumerIdGenerator.generateId());
    545         info.setDestination(ActiveMQMessageTransformation.transformDestination(destination));
    546         info.setSelector(messageSelector);
    547         return new ActiveMQConnectionConsumer(this, sessionPool, info, maxMessages);
    548     }
    549
    550     /***
    551      * Create a durable connection consumer for this connection (optional operation). This is an expert facility not
    552      * used by regular JMS clients.
    553      *
    554      * @param topic            topic to access
    555      * @param subscriptionName durable subscription name
    556      * @param messageSelector  only messages with properties matching the message selector expression are delivered. A
    557      *                         value of null or an empty string indicates that there is no message selector for the message consumer.
    558      * @param sessionPool      the server session pool to associate with this durable connection consumer
    559      * @param maxMessages      the maximum number of messages that can be assigned to a server session at one time
    560      * @return the durable connection consumer
    561      * @throws JMSException if the <CODE>Connection</CODE> object fails to create a connection consumer due to some
    562      *                      internal error or invalid arguments for <CODE>sessionPool</CODE> and <CODE>messageSelector</CODE>.
    563      * @throws javax.jms.InvalidDestinationException
    564      *                      if an invalid destination is specified.
    565      * @throws javax.jms.InvalidSelectorException
    566      *                      if the message selector is invalid.
    567      * @see javax.jms.ConnectionConsumer
    568      * @since 1.1
    569      */
    570     public ConnectionConsumer createDurableConnectionConsumer(Topic topic, String subscriptionName,
    571                                                               String messageSelector, ServerSessionPool sessionPool, int maxMessages) throws JMSException {
    572         checkClosed();
    573         ConsumerInfo info = new ConsumerInfo();
    574         info.setId(this.packetIdGenerator.generateId());
    575         info.setConsumerId(this.consumerIdGenerator.generateId());
    576         info.setDestination(ActiveMQMessageTransformation.transformDestination(topic));
    577         info.setSelector(messageSelector);
    578         info.setConsumerName(subscriptionName);
    579         return new ActiveMQConnectionConsumer(this, sessionPool, info, maxMessages);
    580     }
    581
    582     /***
    583      * @return Returns the prefetchPolicy.
    584      */
    585     public ActiveMQPrefetchPolicy getPrefetchPolicy() {
    586         return prefetchPolicy;
    587     }
    588
    589     /***
    590      * @param prefetchPolicy The prefetchPolicy to set.
    591      */
    592     public void setPrefetchPolicy(ActiveMQPrefetchPolicy prefetchPolicy) {
    593         this.prefetchPolicy = prefetchPolicy;
    594     }
    595
    596     /***
    597      * Implementation of the PacketListener interface - consume a packet
    598      *
    599      * @param packet - the Packet to consume
    600      * @see org.codehaus.activemq.message.PacketListener#consume(org.codehaus.activemq.message.Packet)
    601      */
    602     public void consume(Packet packet) {
    603         if (!closed.get() && packet != null) {
    604             if (packet.isJMSMessage()) {
    605                 ActiveMQMessage message = (ActiveMQMessage) packet;
    606                 message.setReadOnly(true);
    607                 message.setProducerID(clientID);
    608
    609                 try {
    610                     int count = 0;
    611                     for (Iterator i = this.messageDispatchers.iterator(); i.hasNext();) {
    612                         ActiveMQMessageDispatcher dispatcher = (ActiveMQMessageDispatcher) i.next();
    613                         if (dispatcher.isTarget(message)) {
    614                             if (count > 0) {
    615                                 //separate message for each Session etc.
    616                                 message = message.deepCopy();
    617                             }
    618                             dispatcher.dispatch(message);
    619                             count++;
    620                         }
    621                     }
    622                 }
    623                 catch (JMSException jmsEx) {
    624                     handleAsyncException(jmsEx);
    625                 }
    626             }
    627             else if (packet.getPacketType() == Packet.CAPACITY_INFO) {
    628                 CapacityInfo info = (CapacityInfo) packet;
    629                 flowControlSleepTime = info.getFlowControlTimeout();
    630                 //System.out.println("SET FLOW TIMEOUT = " + flowControlSleepTime + " FOR " + info);
    631             }
    632         }
    633     }
    634
    635     /***
    636      * @see javax.jms.ExceptionListener#onException(javax.jms.JMSException)
    637      */
    638     public void onException(JMSException jmsEx) {
    639         //Got an exception propagated up from the transport channel
    640         handleAsyncException(jmsEx);
    641         isTransportOK = false;
    642         try {
    643             close();
    644         }
    645         catch (JMSException ex) {
    646             log.warn("Got an exception closing the connection", ex);
    647         }
    648     }
    649
    650     /***
    651      * Creates a <CODE>TopicSession</CODE> object.
    652      *
    653      * @param transacted      indicates whether the session is transacted
    654      * @param acknowledgeMode indicates whether the consumer or the client will acknowledge any messages it receives;
    655      *                        ignored if the session is transacted. Legal values are <code>Session.AUTO_ACKNOWLEDGE</code>,
    656      *                        <code>Session.CLIENT_ACKNOWLEDGE</code>, and <code>Session.DUPS_OK_ACKNOWLEDGE</code>.
    657      * @return a newly created topic session
    658      * @throws JMSException if the <CODE>TopicConnection</CODE> object fails to create a session due to some internal
    659      *                      error or lack of support for the specific transaction and acknowledgement mode.
    660      * @see Session#AUTO_ACKNOWLEDGE
    661      * @see Session#CLIENT_ACKNOWLEDGE
    662      * @see Session#DUPS_OK_ACKNOWLEDGE
    663      */
    664     public TopicSession createTopicSession(boolean transacted, int acknowledgeMode) throws JMSException {
    665         checkClosed();
    666         return new ActiveMQSession(this, (transacted ? Session.SESSION_TRANSACTED : acknowledgeMode));
    667     }
    668
    669     /***
    670      * Creates a connection consumer for this connection (optional operation). This is an expert facility not used by
    671      * regular JMS clients.
    672      *
    673      * @param topic           the topic to access
    674      * @param messageSelector only messages with properties matching the message selector expression are delivered. A
    675      *                        value of null or an empty string indicates that there is no message selector for the message consumer.
    676      * @param sessionPool     the server session pool to associate with this connection consumer
    677      * @param maxMessages     the maximum number of messages that can be assigned to a server session at one time
    678      * @return the connection consumer
    679      * @throws JMSException                if the <CODE>TopicConnection</CODE> object fails to create a connection consumer due to
    680      *                                     some internal error or invalid arguments for <CODE>sessionPool</CODE> and <CODE>messageSelector</CODE>.
    681      * @throws InvalidDestinationException if an invalid topic is specified.
    682      * @throws InvalidSelectorException    if the message selector is invalid.
    683      * @see javax.jms.ConnectionConsumer
    684      */
    685     public ConnectionConsumer createConnectionConsumer(Topic topic, String messageSelector,
    686                                                        ServerSessionPool sessionPool, int maxMessages) throws JMSException {
    687         checkClosed();
    688         ConsumerInfo info = new ConsumerInfo();
    689         info.setId(this.packetIdGenerator.generateId());
    690         info.setConsumerId(this.consumerIdGenerator.generateId());
    691         info.setDestination(ActiveMQMessageTransformation.transformDestination(topic));
    692         info.setSelector(messageSelector);
    693         return new ActiveMQConnectionConsumer(this, sessionPool, info, maxMessages);
    694     }
    695
    696     /***
    697      * Creates a <CODE>QueueSession</CODE> object.
    698      *
    699      * @param transacted      indicates whether the session is transacted
    700      * @param acknowledgeMode indicates whether the consumer or the client will acknowledge any messages it receives;
    701      *                        ignored if the session is transacted. Legal values are <code>Session.AUTO_ACKNOWLEDGE</code>,
    702      *                        <code>Session.CLIENT_ACKNOWLEDGE</code>, and <code>Session.DUPS_OK_ACKNOWLEDGE</code>.
    703      * @return a newly created queue session
    704      * @throws JMSException if the <CODE>QueueConnection</CODE> object fails to create a session due to some internal
    705      *                      error or lack of support for the specific transaction and acknowledgement mode.
    706      * @see Session#AUTO_ACKNOWLEDGE
    707      * @see Session#CLIENT_ACKNOWLEDGE
    708      * @see Session#DUPS_OK_ACKNOWLEDGE
    709      */
    710     public QueueSession createQueueSession(boolean transacted, int acknowledgeMode) throws JMSException {
    711         checkClosed();
    712         return new ActiveMQSession(this, (transacted ? Session.SESSION_TRANSACTED : acknowledgeMode));
    713     }
    714
    715     /***
    716      * Creates a connection consumer for this connection (optional operation). This is an expert facility not used by
    717      * regular JMS clients.
    718      *
    719      * @param queue           the queue to access
    720      * @param messageSelector only messages with properties matching the message selector expression are delivered. A
    721      *                        value of null or an empty string indicates that there is no message selector for the message consumer.
    722      * @param sessionPool     the server session pool to associate with this connection consumer
    723      * @param maxMessages     the maximum number of messages that can be assigned to a server session at one time
    724      * @return the connection consumer
    725      * @throws JMSException                if the <CODE>QueueConnection</CODE> object fails to create a connection consumer due to
    726      *                                     some internal error or invalid arguments for <CODE>sessionPool</CODE> and <CODE>messageSelector</CODE>.
    727      * @throws InvalidDestinationException if an invalid queue is specified.
    728      * @throws InvalidSelectorException    if the message selector is invalid.
    729      * @see javax.jms.ConnectionConsumer
    730      */
    731     public ConnectionConsumer createConnectionConsumer(Queue queue, String messageSelector,
    732                                                        ServerSessionPool sessionPool, int maxMessages) throws JMSException {
    733         checkClosed();
    734         ConsumerInfo info = new ConsumerInfo();
    735         info.setId(this.packetIdGenerator.generateId());
    736         info.setConsumerId(this.consumerIdGenerator.generateId());
    737         info.setDestination(ActiveMQMessageTransformation.transformDestination(queue));
    738         info.setSelector(messageSelector);
    739         return new ActiveMQConnectionConsumer(this, sessionPool, info, maxMessages);
    740     }
    741
    742     /***
    743      * Ensures that the clientID was manually specified and not auto-generated. If the clientID was not specified this
    744      * method will throw an exception. This method is used to ensure that the clientID + durableSubscriber name are used
    745      * correctly.
    746      */
    747     public void checkClientIDWasManuallySpecified() throws JMSException {
    748         if (!userSpecifiedClientID) {
    749             throw new JMSException("You cannot create a durable subscriber without specifying a unique clientID on a Connection");
    750         }
    751     }
    752
    753     /***
    754      * handle disconnect/reconnect events
    755      *
    756      * @param event
    757      */
    758     public void statusChanged(TransportStatusEvent event) {
    759         log.info("channel status changed: " + event);
    760         if (event.getChannelStatus() == TransportStatusEvent.RECONNECTED) {
    761             doReconnect();
    762         }
    763     }
    764
    765     // Implementation methods
    766     //-------------------------------------------------------------------------
    767     void asyncSendPacket(Packet packet) throws JMSException {
    768         if (isTransportOK && !closed.get()) {
    769             packet.setReceiptRequired(false);
    770             if (packet.isJMSMessage() && flowControlSleepTime > 0) {
    771                 try {
    772                     Thread.sleep(flowControlSleepTime);
    773                 }
    774                 catch (InterruptedException e) {
    775                 }
    776             }
    777             this.transportChannel.asyncSend(packet);
    778         }
    779     }
    780
    781     void syncSendPacket(Packet packet) throws JMSException {
    782         syncSendPacket(packet, 0);
    783     }
    784
    785     void syncSendPacket(Packet packet, int timeout) throws JMSException {
    786         if (isTransportOK && !closed.get()) {
    787             Receipt receipt;
    788             packet.setReceiptRequired(true);
    789             receipt = this.transportChannel.send(packet, timeout);
    790             if (receipt != null) {
    791                 if (receipt.isFailed()) {
    792                     Throwable e = receipt.getException();
    793                     if (e != null) {
    794                         throw (JMSException) new JMSException(e.getMessage()).initCause(e);
    795                     }
    796                     throw new JMSException("syncSendPacket failed with unknown exception");
    797                 }
    798             }
    799         }
    800         else {
    801             throw new JMSException("syncSendTimedOut");
    802         }
    803     }
    804
    805     Receipt syncSendRequest(Packet packet) throws JMSException {
    806         if (isTransportOK && !closed.get()) {
    807             Receipt receipt;
    808             packet.setReceiptRequired(true);
    809             receipt = this.transportChannel.send(packet);
    810             if (receipt.isFailed()) {
    811                 Throwable e = receipt.getException();
    812                 if (e != null) {
    813                     throw (JMSException) new JMSException(e.getMessage()).initCause(e);
    814                 }
    815                 throw new JMSException("syncSendPacket failed with unknown exception");
    816             }
    817             return receipt;
    818         }
    819         else {
    820             throw new JMSException("Connection closed.");
    821         }
    822     }
    823
    824     /***
    825      * Used internally for adding Sessions to the Connection
    826      *
    827      * @param session
    828      * @throws JMSException
    829      */
    830     protected void addSession(ActiveMQSession session) throws JMSException {
    831         this.sessions.add(session);
    832         addMessageDispatcher(session);
    833         SessionInfo info = createSessionInfo(session);
    834         info.setStarted(true);
    835         asyncSendPacket(info);
    836     }
    837
    838     /***
    839      * Used interanlly for removing Sessions from a Connection
    840      *
    841      * @param session
    842      * @throws JMSException
    843      */
    844     protected void removeSession(ActiveMQSession session) throws JMSException {
    845         this.sessions.remove(session);
    846         removeMessageDispatcher(session);
    847         SessionInfo info = createSessionInfo(session);
    848         info.setStarted(false);
    849         asyncSendPacket(info);
    850     }
    851
    852     private SessionInfo createSessionInfo(ActiveMQSession session) {
    853         SessionInfo info = new SessionInfo();
    854         info.setId(packetIdGenerator.generateId());
    855         info.setClientId(clientID);
    856         info.setSessionId(session.getSessionId());
    857         info.setStartTime(session.getStartTime());
    858         return info;
    859     }
    860
    861     /***
    862      * Add a ConnectionConsumer
    863      *
    864      * @param connectionConsumer
    865      * @throws JMSException
    866      */
    867     protected void addConnectionConsumer(ActiveMQConnectionConsumer connectionConsumer) throws JMSException {
    868         this.connectionConsumers.add(connectionConsumer);
    869         addMessageDispatcher(connectionConsumer);
    870     }
    871
    872     /***
    873      * Remove a ConnectionConsumer
    874      *
    875      * @param connectionConsumer
    876      */
    877     protected void removeConnectionConsumer(ActiveMQConnectionConsumer connectionConsumer) {
    878         this.connectionConsumers.add(connectionConsumer);
    879         removeMessageDispatcher(connectionConsumer);
    880     }
    881
    882     /***
    883      * Add a Message dispatcher to receive messages from the Broker
    884      *
    885      * @param messageDispatch
    886      * @throws JMSException if an internal error
    887      */
    888     protected void addMessageDispatcher(ActiveMQMessageDispatcher messageDispatch) throws JMSException {
    889         this.messageDispatchers.add(messageDispatch);
    890     }
    891
    892     /***
    893      * Remove a Message dispatcher
    894      *
    895      * @param messageDispatcher
    896      */
    897     protected void removeMessageDispatcher(ActiveMQMessageDispatcher messageDispatcher) {
    898         this.messageDispatchers.remove(messageDispatcher);
    899     }
    900
    901     /***
    902      * Used for handling async exceptions
    903      *
    904      * @param jmsEx
    905      */
    906     protected void handleAsyncException(JMSException jmsEx) {
    907         if (this.exceptionListener != null) {
    908             this.exceptionListener.onException(jmsEx);
    909         }
    910         else {
    911             log.warn("async exception with no exception listener", jmsEx);
    912         }
    913     }
    914
    915     private void sendConnectionInfoToBroker() throws JMSException {
    916         sendConnectionInfoToBroker(0, closed.get());
    917     }
    918
    919     /***
    920      * Send the ConnectionInfo to the Broker
    921      *
    922      * @param timeout
    923      * @param isClosed
    924      * @throws JMSException
    925      */
    926     private void sendConnectionInfoToBroker(int timeout, boolean isClosed) throws JMSException {
    927         if (!isConnectionInfoSentToBroker) {
    928             transportChannel.setClientID(clientID);
    929         }
    930         this.isConnectionInfoSentToBroker = true;
    931         ensureClientIDInitialised();
    932         ConnectionInfo info = new ConnectionInfo();
    933         info.setClientId(this.clientID);
    934         info.setHostName(IdGenerator.getHostName());
    935         info.setUserName(userName);
    936         info.setPassword(password);
    937         info.setId(packetIdGenerator.generateId());
    938         info.setStartTime(startTime);
    939         info.setStarted(started.get());
    940         info.setClosed(isClosed);
    941         syncSendPacket(info, timeout);
    942     }
    943
    944     /***
    945      * Set the maximum amount of memory this Connection should use for buffered inbound messages
    946      *
    947      * @param newMemoryLimit the new memory limit in bytes
    948      */
    949     public void setConnectionMemoryLimit(int newMemoryLimit) {
    950         boundedQueueManager.setValueLimit(newMemoryLimit);
    951     }
    952
    953     /***
    954      * Get the current value for the maximum amount of memory this Connection should use for buffered inbound messages
    955      *
    956      * @return the current limit in bytes
    957      */
    958     public int getConnectionMemoryLimit() {
    959         return (int) boundedQueueManager.getValueLimit();
    960     }
    961
    962     /***
    963      * CapacityMonitorEventListener implementation called when the capacity of a CapacityService changes
    964      *
    965      * @param event
    966      */
    967     public void capacityChanged(CapacityMonitorEvent event) {
    968         //send the event to broker ...
    969         CapacityInfo info = new CapacityInfo();
    970         info.setId(packetIdGenerator.generateId());
    971         info.setResourceName(event.getMonitorName());
    972         info.setCapacity(event.getCapacity());
    973         //System.out.println("Cap changed: " + event);
    974         try {
    975             asyncSendPacket(info);
    976         }
    977         catch (JMSException e) {
    978             JMSException jmsEx = new JMSException("failed to send change in capacity");
    979             jmsEx.setLinkedException(e);
    980             handleAsyncException(jmsEx);
    981         }
    982     }
    983
    984     /***
    985      * @return a number unique for this connection
    986      */
    987     protected int getNextConsumerNumber() {
    988         return this.consumerNumberGenerator.increment();
    989     }
    990
    991     protected String generateSessionId() {
    992         return this.sessionIdGenerator.generateId();
    993     }
    994
    995     protected void ensureClientIDInitialised() {
    996         if (this.clientID == null) {
    997             this.clientID = this.clientIdGenerator.generateId();
    998         }
    999         this.clientIDSet = true;
    1000     }
    1001
    1002     protected MemoryBoundedQueue getMemoryBoundedQueue(String name) {
    1003         return boundedQueueManager.getMemoryBoundedQueue(name);
    1004     }
    1005
    1006     protected void doReconnect() {
    1007         try {
    1008             //send the Connection info again
    1009             sendConnectionInfoToBroker();
    1010             for (Iterator iter = sessions.iterator(); iter.hasNext();) {
    1011                 ActiveMQSession session = (ActiveMQSession) iter.next();
    1012                 SessionInfo sessionInfo = createSessionInfo(session);
    1013                 sessionInfo.setStarted(true);
    1014                 asyncSendPacket(sessionInfo);
    1015                 //send consumers
    1016                 for (Iterator consumersIterator = session.consumers.iterator(); consumersIterator.hasNext();) {
    1017                     ActiveMQMessageConsumer consumer = (ActiveMQMessageConsumer) consumersIterator.next();
    1018                     ConsumerInfo consumerInfo = session.createConsumerInfo(consumer);
    1019                     consumerInfo.setStarted(true);
    1020                     syncSendPacket(consumerInfo);
    1021                 }
    1022                 //send producers
    1023                 for (Iterator producersIterator = session.producers.iterator(); producersIterator.hasNext();) {
    1024                     ActiveMQMessageProducer producer = (ActiveMQMessageProducer) producersIterator.next();
    1025                     ProducerInfo producerInfo = session.createProducerInfo(producer);
    1026                     producerInfo.setStarted(true);
    1027                     syncSendPacket(producerInfo);
    1028                 }
    1029             }
    1030         }
    1031         catch (JMSException jmsEx) {
    1032             log.error("Failed to do reconnection");
    1033             handleAsyncException(jmsEx);
    1034             isTransportOK = false;
    1035         }
    1036     }
    1037 }