foundation/org.eclipse.persistence.core/src/main/java/org/eclipse/persistence/internal/sessions/coordination/jms/JMSTopicRemoteConnection.java - eclipselink - Git at Google

 /*
  * Copyright (c) 1998, 2021 Oracle and/or its affiliates. All rights reserved.
  *
  * This program and the accompanying materials are made available under the
  * terms of the Eclipse Public License v. 2.0 which is available at
  * http://www.eclipse.org/legal/epl-2.0,
  * or the Eclipse Distribution License v. 1.0 which is available at
  * http://www.eclipse.org/org/documents/edl-v10.php.
  *
  * SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause
  */

 // Contributors:
 //     Oracle - initial API and implementation from Oracle TopLink
 //     cdelahun - Bug 214534: added JMS Cache Coordination for publishing only
 package org.eclipse.persistence.internal.sessions.coordination.jms;

 import java.io.Serializable;

 import jakarta.jms.BytesMessage;
 import jakarta.jms.JMSException;
 import jakarta.jms.Message;
 import jakarta.jms.ObjectMessage;
 import jakarta.jms.Topic;
 import jakarta.jms.TopicConnection;
 import jakarta.jms.TopicConnectionFactory;
 import jakarta.jms.TopicPublisher;
 import jakarta.jms.TopicSession;
 import jakarta.jms.TopicSubscriber;

 import org.eclipse.persistence.exceptions.RemoteCommandManagerException;
 import org.eclipse.persistence.internal.sessions.AbstractSession;
 import org.eclipse.persistence.internal.sessions.coordination.broadcast.BroadcastRemoteConnection;
 import org.eclipse.persistence.sessions.coordination.RemoteCommandManager;
 import org.eclipse.persistence.sessions.coordination.jms.JMSTopicTransportManager;
 import org.eclipse.persistence.sessions.serializers.JavaSerializer;
 import org.eclipse.persistence.sessions.serializers.Serializer;

 /**
  * <p>
  * <b>Purpose</b>: Define the implementation of the abstract RemoteConnection for JMS.
  * <p>
  * <b>Description</b>:  Executing commands implementation of RemoteConnection is done via JMS Publisher.
  *
  * Using a single TopicConnection for both publishing and subscribing would
  * allow subscriber to ignore messages sent through the same TopicConnection - and therefore
  * allow JMSTopicRemoteConnection to ignore messages that it has itself published.
  * Unfortunately J2EE doesn't allow that:
  * J2EE spec. (J2EE.6.6 in v1.4) states:
  *   "The following methods may only be used by application components executing
  *     in the application client container: javax.jms.Session method setMessageListener ...
  *   Application components in the web and EJB containers must not attempt to create more than one
  *     active (not closed) Session object per connection."
  * Because of these restrictions
  * a) two JMSTopicRemoteConnection are required - one for publishing (external) and another one for listening (local);
  * b) listening should be done using subscriber.receive() in an infinite loop in a separate thread,
  * that's why the class implements Runnable interface.
  * c) publishing connection (external) could be used concurrently to send messages, so it cannot use the same publisher/session/topicConnection
  *   Instead, it will store the TopicConnectionFactory and use it to create connections when executeCommandInternal is called (much like
  *   DatabaseAccessor when an external pool is used)
  *
  * @author Steven Vo
  * @since OracleAS TopLink 10<i>g</i> (10.0.3)
  */
 public class JMSTopicRemoteConnection extends BroadcastRemoteConnection implements Runnable {
     protected TopicConnectionFactory topicConnectionFactory;
     protected Topic topic;
     // indicates whether it's a local connection.
     protected boolean isLocal;

     // not used by External (publishing) connection unless it is set to reuse the JMSTopicPublisher
     private TopicPublisher publisher;
     protected TopicConnection topicConnection;
     protected TopicSession topicSession;

     // Used only by local (listening) connection
     protected TopicSubscriber subscriber;


     // Used only by local (listening) connection and
     // only in case shouldRemoveConnectionOnError==false.
     // Time to wait after error receiving jms message:
     // wait to avoid a busy thread throwing barrage of exceptions
     // in case JMS server goes down.
     // User can avoid the wait by handling ERROR_RECEIVING_JMS_MESSAGE exception.
     public static long WAIT_ON_ERROR_RECEIVING_JMS_MESSAGE = 10000;

     /**
      * INTERNAL:
      * Constructor creating either a local or external connection.  Local connections created this way connect to the topicSession and cache
      * the session and subscriber.  External connections cache only the topicConnection and will obtain the session/publisher when needed.
      */
     public JMSTopicRemoteConnection(RemoteCommandManager rcm, TopicConnectionFactory topicConnectionFactory, Topic topic, boolean isLocalConnectionBeingCreated, boolean reuseJMSTopicPublisher) throws JMSException {
         super(rcm);
         this.topicConnectionFactory = topicConnectionFactory;
         this.topic = topic;
         this.isLocal = isLocalConnectionBeingCreated;
         rcm.logDebug("creating_broadcast_connection", getInfo());
         try {
             if(isLocalConnectionBeingCreated) {
                 // it's a local connection
                 this.topicConnection = topicConnectionFactory.createTopicConnection();
                 this.topicSession = topicConnection.createTopicSession(false, jakarta.jms.Session.AUTO_ACKNOWLEDGE);
                 this.subscriber = topicSession.createSubscriber(topic);
                 topicConnection.start();
                 rcm.logDebug("broadcast_connection_created", getInfo());
                 rcm.getServerPlatform().launchContainerRunnable(this);
             } else if (reuseJMSTopicPublisher) {
                 // it's an external connection and is set to reuse the TopicPublisher (legacy)
                 this.topicConnection = topicConnectionFactory.createTopicConnection();
                 this.topicSession = topicConnection.createTopicSession(false, jakarta.jms.Session.AUTO_ACKNOWLEDGE);
                 this.setPublisher(topicSession.createPublisher(topic));
                 rcm.logDebug("broadcast_connection_created", getInfo());
             } //else bug214534: it's an external connection, with TopicConnections, sessions and publishers created as needed
         } catch (JMSException ex) {
             rcm.logDebug("failed_to_create_broadcast_connection", getInfo());
             close();
             throw ex;
         }
     }

     /**
      * Creates local connections that do not use a TopicConnection or TopicSession,
      * useful only for processing already received JMS messages
      * @see onMessage
      */
     public JMSTopicRemoteConnection(RemoteCommandManager rcm){
         super(rcm);
         this.isLocal = true;
     }

     /**
      * INTERNAL:
      * Indicates whether connection is local (subscriber)
      * or external (publisher).
      */
     public boolean isLocal() {
         return isLocal;
     }

     /**
      * INTERNAL:
      * Execute the remote command. The result of execution is returned.
      * This method is used only by external (publishing) connection.
      */
     @Override
     protected Object executeCommandInternal(Object command) throws Exception {
         TopicConnection jmsConnection = null;
         try {
             TopicPublisher topicPublisher = this.publisher;
             TopicSession publishingSession = this.topicSession;
             //if the publisher is set, reuse it.  Otherwise, create it (and the connection and session) from the topicConnectionFactory
             if ( topicPublisher == null ){
                 jmsConnection = topicConnectionFactory.createTopicConnection();
                 publishingSession = jmsConnection.createTopicSession(false, jakarta.jms.Session.AUTO_ACKNOWLEDGE);
                 topicPublisher = publishingSession.createPublisher(topic);
             }

             Message message;
             if (command instanceof byte[]) {
                 message = publishingSession.createBytesMessage();
                 ((BytesMessage)message).writeBytes((byte[])command);
             } else {
                 message = publishingSession.createObjectMessage();
                 ((ObjectMessage)message).setObject((Serializable)command);
             }

             Object[] debugInfo = null;
             if (rcm.shouldLogDebugMessage()) {
                 // null passed because JMSMessageId is not yet created.
                 debugInfo = logDebugBeforePublish(null);
             }

             topicPublisher.publish(message);

             // debug logging is on
             if (debugInfo != null) {
                 // now messageId has been created - let's use it.
                 logDebugAfterPublish(debugInfo, message.getJMSMessageID());
             }

             return null;
         } finally {
             //only need to close the topicConnection, not the session or publisher, and only if it was created in this method.
             if (jmsConnection != null) {
                 jmsConnection.close();
             }
         }
     }

     /**
      * INTERNAL:
      * Process received JMS message.
      * This method is used only by local (listening) connection.
      */
     public void onMessage(Message message) {
         String topic = null;
         String messageId = "";
         if (rcm.shouldLogDebugMessage()) {
             try {
                 messageId = message.getJMSMessageID();
                 logDebugOnReceiveMessage(messageId);
                 topic = logDebugJMSTopic(message);
             } catch (JMSException ex) {
                 // ignore
             }
         }

         Object object = null;
         try {
             if (message instanceof ObjectMessage) {
                 object = ((ObjectMessage)message).getObject();
             } else if (message instanceof BytesMessage) {
                 BytesMessage byteMessage = (BytesMessage)message;
                 byte[] bytes = new byte[(int)byteMessage.getBodyLength()];
                 byteMessage.readBytes(bytes);
                 Serializer serializer = this.rcm.getSerializer();
                 if (serializer == null) {
                     serializer = JavaSerializer.instance;
                 }
                 object = serializer.deserialize(bytes, (AbstractSession)this.rcm.getCommandProcessor());
             } else {
                 if (this.rcm.shouldLogWarningMessage() && (topic == null)) {
                     try {
                         topic = ((Topic)message.getJMSDestination()).getTopicName();
                     } catch (JMSException ex) {
                         // ignore
                         topic = "";
                     }
                     Object[] args = { message.getClass().getName(), topic };
                     this.rcm.logWarningWithoutLevelCheck("received_unexpected_message_type", args);
                 }
                 return;
             }
         } catch (Exception exception) {
             if (messageId.length() == 0) {
                 try {
                     messageId = message.getJMSMessageID();
                 } catch (JMSException ex) {
                     // ignore
                 }
             }
             failDeserializeMessage(messageId, exception);
             return;
         }

         processReceivedObject(object, messageId);
     }

     /**
      * INTERNAL:
      * Indicates whether all the resources used by connection are freed after close method returns.
      * Usually that's the case. However in case of local (listening) JMSTopicRemoteConnection
      * close merely indicates to the listening thread that it should free TopicConnection and exit.
      * Note that it may take a while: the listening thread waits until subscriber.receive method either
      * returns a message or throws an exception.
      */
     @Override
     protected boolean areAllResourcesFreedOnClose() {
         return !isLocal();
     }

     /**
      * INTERNAL:
      * This method is called by close method.
      * This method usually
      * (but not always see comment to areAllResourcesFreedOnClose method)
      * frees all the resources.
      */
     @Override
     protected void closeInternal() throws JMSException {
         //this method should be a no-op now that external connections open/close TopicConnection when needed.  Close on Local
         //connections will eventually cause topicConnection.close() in their listening thread, so it should not be called here
         if(areAllResourcesFreedOnClose() && topicConnection!=null) {
             // There is no need to close the sessions, producers, and consumers of a closed TopicConnection.
             topicConnection.close();
         }
     }

     /**
      * INTERNAL:
      */
     protected String logDebugJMSTopic(Message message) throws JMSException {
         String topic = ((Topic)message.getJMSDestination()).getTopicName();
         Object[] args = { topic };
         // call logDebugWithoutLevelCheck to avoid the second rcm.shouldLogDebugMessage() check
         rcm.logDebugWithoutLevelCheck("retreived_remote_message_from_JMS_topic", args);
         return topic;
     }

     /**
      * INTERNAL:
      * This method is used by local (listening) connection only.
      * The only way to exit the loop is to set isActive to false -
      * there should be no uncaught exceptions thrown from inside the loop.
      * The execution exits the loop either in case of exception in remove connection on error mode;
      * or by trasportManager.removeLocalConnection() call
      * (which calls connection.close(), which sets isActive to false).
      */
     @Override
     public void run() {
         JMSTopicTransportManager tm = (JMSTopicTransportManager)rcm.getTransportManager();
         rcm.logDebug("broadcast_connection_start_listening", getInfo());

         // indicates whether to create a new connection before exiting the thread.
         boolean shouldReconnect = false;
         // exception indicating that the received message is null.
         RuntimeException messageIsNullException = null;
         // isActive() returning false indicates that close method has been called.
         // Should never exit this loop because of an uncaught exception.
         while (isActive()) {
             try {
                 Message message = subscriber.receive();
                 // need the second isActive check here:
                 // close method could have been called while subscriber.receive() was waiting.
                 if (isActive()) {
                     if(message == null) {
                         try {
                             // user has a chance to handle exception - for instance to ignore it.
                             rcm.handleException(RemoteCommandManagerException.errorJMSMessageIsNull());
                             // exception has been handled, go to the next iteration.
                             continue;
                         } catch (RuntimeException ex) {
                             messageIsNullException = ex;
                             // throw a dummy JMSException to get into catch block
                             throw new JMSException("");
                         }
                     }
                     // process the message and log a warning without throwing exception if there was exception.
                     rcm.getServerPlatform().launchContainerRunnable(new JMSOnMessageHelper(message));
                 }
             } catch (JMSException e) {
                 // need the second isActive check here:
                 // close method could have been called while subscriber.receive() was waiting.
                 if (isActive()) {
                     RemoteCommandManagerException rcmException;
                     if(messageIsNullException != null) {
                         rcmException = RemoteCommandManagerException.errorReceivingJMSMessage(messageIsNullException);
                         messageIsNullException = null;
                     } else {
                         rcmException = RemoteCommandManagerException.errorReceivingJMSMessage(e);
                     }
                     if (tm.shouldRemoveConnectionOnError()) {
                         shouldReconnect = true;
                         Object[] args = { getServiceId(), rcmException };
                         rcm.logWarning("drop_connection_on_error", args);
                         // after connection is closed isActive will return false.
                         tm.removeLocalConnection();
                     } else {
                         try {
                             // user has a chance to handle exception:
                             // for instance to shut down the command manager.
                             rcm.handleException(rcmException);
                         } catch (RuntimeException ex) {
                             // Ignore the exception, sleep before going back to listening.
                             Object[] args = { toString(), rcmException, WAIT_ON_ERROR_RECEIVING_JMS_MESSAGE };
                             rcm.logWarning("broadcast_listening_sleep_on_error", args);
                             try {
                                 Thread.sleep(WAIT_ON_ERROR_RECEIVING_JMS_MESSAGE);
                             } catch (InterruptedException interruptedEception) {
                                 // Ignore
                             }
                         }
                     }
                 }
             }
         }

         // Out of the loop - that means close method has been called.
         rcm.logDebug("broadcast_connection_stop_listening", getInfo());
         if(isClosing()) {
             try {
                 // There is no need to close the sessions, producers, and consumers of a closed TopicConnection.
                 topicConnection.close();
             } catch (JMSException closeException) {
                 Object[] args = { displayString, closeException };
                 rcm.logWarning("broadcast_exception_thrown_when_attempting_to_close_connection", args);
             } finally {
                 rcm.logDebug("broadcast_connection_closed", getInfo());
                 state = STATE_CLOSED;
             }
         }

         if(shouldReconnect && !tm.getRemoteCommandManager().isStopped()) {
             try {
                 tm.createLocalConnection();
             } catch (RemoteCommandManagerException ex) {
                 // Ignore exception - user had a chance to handle it in createLocalConnection method:
                 // for instance to change host url and create a new local connection.
             }
         }
     }

     /**
      * INTERNAL:
      * Used for debug logging
      */
     @Override
     protected void createDisplayString() {
         super.createDisplayString();
         displayString = (isLocal() ? "Local " : "External ") + displayString;
     }

     /**
      * INTERNAL:
      * Return whether a BroadcastConnection should check a ServiceId against its
      * own ServiceId to avoid the processing of Commands with the same ServiceId.
      * This should take place (return true) for a JMSTopicRemoteConnection.
      * @return boolean
      */
     @Override
     protected boolean shouldCheckServiceId() {
         return true;
     }

     /**
      * INTERNAL:
      * set the TopicPublisher to be used when this RemoteConnection executes a command.
      * Setting the TopicPublisher avoids having it obtained on each executeCommandInternal
      * call.  Passing in a publisher requires a TopicSession to also be set.  These will
      * not be closed until the external RemoteConnection is closed, and then only if the
      * TopicConnection is also set.
      */
     public void setPublisher(TopicPublisher publisher) {
         this.publisher = publisher;
     }

     public TopicPublisher getPublisher() {
         return publisher;
     }

     /**
      * INTERNAL:
      * set the TopicSubscriber on a local RemoteConnection for reading JMS messages when
      * this runnable connection is started in a thread.  If setting this, a TopicConnection
      * is also required to be set, in order for it to be closed when the thread completes.
      * This is only to be used when using the JMSTopicRemoteConnection(rcm) constructor.
      */
     public void setSuscriber(TopicSubscriber subscriber) {
         this.subscriber = subscriber;
     }

     public TopicSubscriber getSubscriber() {
         return subscriber;
     }

     /**
      * INTERNAL:
      * set the TopicSession to be used when this RemoteConnection executes a command if
      * the publisher is also set.  Setting the TopicSession and Publisher avoids having
      * them obtained on each executeCommandInternal call.  Passing in a TopicSession
      * requires a TopicPublisher to also be set.  These will not be closed until the
      * external RemoteConnection is closed, and then only if the TopicConnection is also
      * set.
      */
     public void setTopicSession(TopicSession topicSession) {
         this.topicSession = topicSession;
     }

     public TopicSession getTopicSession() {
         return topicSession;
     }

     /**
      * INTERNAL:
      * Set the TopicConnectionFactory, which is used if the publisher is not set to
      * obtain the TopicConnection, TopicSession and TopicPublisher
      */
     public void setTopicConnectionFactory(TopicConnectionFactory topicConnectionFactory){
         this.topicConnectionFactory = topicConnectionFactory;
     }

     public TopicConnection getTopicConnectionFactory() {
         return topicConnection;
     }

     /**
      * INTERNAL:
      * Set the TopicConnection. If this is set, a Publisher and TopicSession must also
      * be set, or a new TopicConnection will be obtained on each executeCommandInternal
      * call.  This TopicConnection is only used on close, as closing the TopicConnection
      * also closes any open TopicSessions and Publishers obtained from it.
      */
     public void setTopicConnection(TopicConnection topicConnection){
         this.topicConnection = topicConnection;
     }

     public TopicConnection getTopicConnection() {
         return topicConnection;
     }

     /**
      * INTERNAL:
      * Set the Topic.  The Topic is required with the TopicConnectionFactory to obtain connections
      * if the TopicPublisher is not set.
      */
     public void setTopic(Topic topic){
         this.topic = topic;
     }

     public Topic getTopic() {
         return topic;
     }

     class JMSOnMessageHelper implements Runnable {
         Message message = null;

         public JMSOnMessageHelper(Message message) {
             this.message = message;
         }

         @Override
         public void run() {
             onMessage(message);
         }
     }
 }
	/*
	* Copyright (c) 1998, 2021 Oracle and/or its affiliates. All rights reserved.
	*
	* This program and the accompanying materials are made available under the
	* terms of the Eclipse Public License v. 2.0 which is available at
	* http://www.eclipse.org/legal/epl-2.0,
	* or the Eclipse Distribution License v. 1.0 which is available at
	* http://www.eclipse.org/org/documents/edl-v10.php.
	*
	* SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause
	*/

	// Contributors:
	// Oracle - initial API and implementation from Oracle TopLink
	// cdelahun - Bug 214534: added JMS Cache Coordination for publishing only
	package org.eclipse.persistence.internal.sessions.coordination.jms;

	import java.io.Serializable;

	import jakarta.jms.BytesMessage;
	import jakarta.jms.JMSException;
	import jakarta.jms.Message;
	import jakarta.jms.ObjectMessage;
	import jakarta.jms.Topic;
	import jakarta.jms.TopicConnection;
	import jakarta.jms.TopicConnectionFactory;
	import jakarta.jms.TopicPublisher;
	import jakarta.jms.TopicSession;
	import jakarta.jms.TopicSubscriber;

	import org.eclipse.persistence.exceptions.RemoteCommandManagerException;
	import org.eclipse.persistence.internal.sessions.AbstractSession;
	import org.eclipse.persistence.internal.sessions.coordination.broadcast.BroadcastRemoteConnection;
	import org.eclipse.persistence.sessions.coordination.RemoteCommandManager;
	import org.eclipse.persistence.sessions.coordination.jms.JMSTopicTransportManager;
	import org.eclipse.persistence.sessions.serializers.JavaSerializer;
	import org.eclipse.persistence.sessions.serializers.Serializer;

	/**
	* <p>
	* <b>Purpose</b>: Define the implementation of the abstract RemoteConnection for JMS.
	* <p>
	* <b>Description</b>: Executing commands implementation of RemoteConnection is done via JMS Publisher.
	*
	* Using a single TopicConnection for both publishing and subscribing would
	* allow subscriber to ignore messages sent through the same TopicConnection - and therefore
	* allow JMSTopicRemoteConnection to ignore messages that it has itself published.
	* Unfortunately J2EE doesn't allow that:
	* J2EE spec. (J2EE.6.6 in v1.4) states:
	* "The following methods may only be used by application components executing
	* in the application client container: javax.jms.Session method setMessageListener ...
	* Application components in the web and EJB containers must not attempt to create more than one
	* active (not closed) Session object per connection."
	* Because of these restrictions
	* a) two JMSTopicRemoteConnection are required - one for publishing (external) and another one for listening (local);
	* b) listening should be done using subscriber.receive() in an infinite loop in a separate thread,
	* that's why the class implements Runnable interface.
	* c) publishing connection (external) could be used concurrently to send messages, so it cannot use the same publisher/session/topicConnection
	* Instead, it will store the TopicConnectionFactory and use it to create connections when executeCommandInternal is called (much like
	* DatabaseAccessor when an external pool is used)
	*
	* @author Steven Vo
	* @since OracleAS TopLink 10<i>g</i> (10.0.3)
	*/
	public class JMSTopicRemoteConnection extends BroadcastRemoteConnection implements Runnable {
	protected TopicConnectionFactory topicConnectionFactory;
	protected Topic topic;
	// indicates whether it's a local connection.
	protected boolean isLocal;

	// not used by External (publishing) connection unless it is set to reuse the JMSTopicPublisher
	private TopicPublisher publisher;
	protected TopicConnection topicConnection;
	protected TopicSession topicSession;

	// Used only by local (listening) connection
	protected TopicSubscriber subscriber;


	// Used only by local (listening) connection and
	// only in case shouldRemoveConnectionOnError==false.
	// Time to wait after error receiving jms message:
	// wait to avoid a busy thread throwing barrage of exceptions
	// in case JMS server goes down.
	// User can avoid the wait by handling ERROR_RECEIVING_JMS_MESSAGE exception.
	public static long WAIT_ON_ERROR_RECEIVING_JMS_MESSAGE = 10000;

	/**
	* INTERNAL:
	* Constructor creating either a local or external connection. Local connections created this way connect to the topicSession and cache
	* the session and subscriber. External connections cache only the topicConnection and will obtain the session/publisher when needed.
	*/
	public JMSTopicRemoteConnection(RemoteCommandManager rcm, TopicConnectionFactory topicConnectionFactory, Topic topic, boolean isLocalConnectionBeingCreated, boolean reuseJMSTopicPublisher) throws JMSException {
	super(rcm);
	this.topicConnectionFactory = topicConnectionFactory;
	this.topic = topic;
	this.isLocal = isLocalConnectionBeingCreated;
	rcm.logDebug("creating_broadcast_connection", getInfo());
	try {
	if(isLocalConnectionBeingCreated) {
	// it's a local connection
	this.topicConnection = topicConnectionFactory.createTopicConnection();
	this.topicSession = topicConnection.createTopicSession(false, jakarta.jms.Session.AUTO_ACKNOWLEDGE);
	this.subscriber = topicSession.createSubscriber(topic);
	topicConnection.start();
	rcm.logDebug("broadcast_connection_created", getInfo());
	rcm.getServerPlatform().launchContainerRunnable(this);
	} else if (reuseJMSTopicPublisher) {
	// it's an external connection and is set to reuse the TopicPublisher (legacy)
	this.topicConnection = topicConnectionFactory.createTopicConnection();
	this.topicSession = topicConnection.createTopicSession(false, jakarta.jms.Session.AUTO_ACKNOWLEDGE);
	this.setPublisher(topicSession.createPublisher(topic));
	rcm.logDebug("broadcast_connection_created", getInfo());
	} //else bug214534: it's an external connection, with TopicConnections, sessions and publishers created as needed
	} catch (JMSException ex) {
	rcm.logDebug("failed_to_create_broadcast_connection", getInfo());
	close();
	throw ex;
	}
	}

	/**
	* Creates local connections that do not use a TopicConnection or TopicSession,
	* useful only for processing already received JMS messages
	* @see onMessage
	*/
	public JMSTopicRemoteConnection(RemoteCommandManager rcm){
	super(rcm);
	this.isLocal = true;
	}

	/**
	* INTERNAL:
	* Indicates whether connection is local (subscriber)
	* or external (publisher).
	*/
	public boolean isLocal() {
	return isLocal;
	}

	/**
	* INTERNAL:
	* Execute the remote command. The result of execution is returned.
	* This method is used only by external (publishing) connection.
	*/
	@Override
	protected Object executeCommandInternal(Object command) throws Exception {
	TopicConnection jmsConnection = null;
	try {
	TopicPublisher topicPublisher = this.publisher;
	TopicSession publishingSession = this.topicSession;
	//if the publisher is set, reuse it. Otherwise, create it (and the connection and session) from the topicConnectionFactory
	if ( topicPublisher == null ){
	jmsConnection = topicConnectionFactory.createTopicConnection();
	publishingSession = jmsConnection.createTopicSession(false, jakarta.jms.Session.AUTO_ACKNOWLEDGE);
	topicPublisher = publishingSession.createPublisher(topic);
	}

	Message message;
	if (command instanceof byte[]) {
	message = publishingSession.createBytesMessage();
	((BytesMessage)message).writeBytes((byte[])command);
	} else {
	message = publishingSession.createObjectMessage();
	((ObjectMessage)message).setObject((Serializable)command);
	}

	Object[] debugInfo = null;
	if (rcm.shouldLogDebugMessage()) {
	// null passed because JMSMessageId is not yet created.
	debugInfo = logDebugBeforePublish(null);
	}

	topicPublisher.publish(message);

	// debug logging is on
	if (debugInfo != null) {
	// now messageId has been created - let's use it.
	logDebugAfterPublish(debugInfo, message.getJMSMessageID());
	}

	return null;
	} finally {
	//only need to close the topicConnection, not the session or publisher, and only if it was created in this method.
	if (jmsConnection != null) {
	jmsConnection.close();
	}
	}
	}

	/**
	* INTERNAL:
	* Process received JMS message.
	* This method is used only by local (listening) connection.
	*/
	public void onMessage(Message message) {
	String topic = null;
	String messageId = "";
	if (rcm.shouldLogDebugMessage()) {
	try {
	messageId = message.getJMSMessageID();
	logDebugOnReceiveMessage(messageId);
	topic = logDebugJMSTopic(message);
	} catch (JMSException ex) {
	// ignore
	}
	}

	Object object = null;
	try {
	if (message instanceof ObjectMessage) {
	object = ((ObjectMessage)message).getObject();
	} else if (message instanceof BytesMessage) {
	BytesMessage byteMessage = (BytesMessage)message;
	byte[] bytes = new byte[(int)byteMessage.getBodyLength()];
	byteMessage.readBytes(bytes);
	Serializer serializer = this.rcm.getSerializer();
	if (serializer == null) {
	serializer = JavaSerializer.instance;
	}
	object = serializer.deserialize(bytes, (AbstractSession)this.rcm.getCommandProcessor());
	} else {
	if (this.rcm.shouldLogWarningMessage() && (topic == null)) {
	try {
	topic = ((Topic)message.getJMSDestination()).getTopicName();
	} catch (JMSException ex) {
	// ignore
	topic = "";
	}
	Object[] args = { message.getClass().getName(), topic };
	this.rcm.logWarningWithoutLevelCheck("received_unexpected_message_type", args);
	}
	return;
	}
	} catch (Exception exception) {
	if (messageId.length() == 0) {
	try {
	messageId = message.getJMSMessageID();
	} catch (JMSException ex) {
	// ignore
	}
	}
	failDeserializeMessage(messageId, exception);
	return;
	}

	processReceivedObject(object, messageId);
	}

	/**
	* INTERNAL:
	* Indicates whether all the resources used by connection are freed after close method returns.
	* Usually that's the case. However in case of local (listening) JMSTopicRemoteConnection
	* close merely indicates to the listening thread that it should free TopicConnection and exit.
	* Note that it may take a while: the listening thread waits until subscriber.receive method either
	* returns a message or throws an exception.
	*/
	@Override
	protected boolean areAllResourcesFreedOnClose() {
	return !isLocal();
	}

	/**
	* INTERNAL:
	* This method is called by close method.
	* This method usually
	* (but not always see comment to areAllResourcesFreedOnClose method)
	* frees all the resources.
	*/
	@Override
	protected void closeInternal() throws JMSException {
	//this method should be a no-op now that external connections open/close TopicConnection when needed. Close on Local
	//connections will eventually cause topicConnection.close() in their listening thread, so it should not be called here
	if(areAllResourcesFreedOnClose() && topicConnection!=null) {
	// There is no need to close the sessions, producers, and consumers of a closed TopicConnection.
	topicConnection.close();
	}
	}

	/**
	* INTERNAL:
	*/
	protected String logDebugJMSTopic(Message message) throws JMSException {
	String topic = ((Topic)message.getJMSDestination()).getTopicName();
	Object[] args = { topic };
	// call logDebugWithoutLevelCheck to avoid the second rcm.shouldLogDebugMessage() check
	rcm.logDebugWithoutLevelCheck("retreived_remote_message_from_JMS_topic", args);
	return topic;
	}

	/**
	* INTERNAL:
	* This method is used by local (listening) connection only.
	* The only way to exit the loop is to set isActive to false -
	* there should be no uncaught exceptions thrown from inside the loop.
	* The execution exits the loop either in case of exception in remove connection on error mode;
	* or by trasportManager.removeLocalConnection() call
	* (which calls connection.close(), which sets isActive to false).
	*/
	@Override
	public void run() {
	JMSTopicTransportManager tm = (JMSTopicTransportManager)rcm.getTransportManager();
	rcm.logDebug("broadcast_connection_start_listening", getInfo());

	// indicates whether to create a new connection before exiting the thread.
	boolean shouldReconnect = false;
	// exception indicating that the received message is null.
	RuntimeException messageIsNullException = null;
	// isActive() returning false indicates that close method has been called.
	// Should never exit this loop because of an uncaught exception.
	while (isActive()) {
	try {
	Message message = subscriber.receive();
	// need the second isActive check here:
	// close method could have been called while subscriber.receive() was waiting.
	if (isActive()) {
	if(message == null) {
	try {
	// user has a chance to handle exception - for instance to ignore it.
	rcm.handleException(RemoteCommandManagerException.errorJMSMessageIsNull());
	// exception has been handled, go to the next iteration.
	continue;
	} catch (RuntimeException ex) {
	messageIsNullException = ex;
	// throw a dummy JMSException to get into catch block
	throw new JMSException("");
	}
	}
	// process the message and log a warning without throwing exception if there was exception.
	rcm.getServerPlatform().launchContainerRunnable(new JMSOnMessageHelper(message));
	}
	} catch (JMSException e) {
	// need the second isActive check here:
	// close method could have been called while subscriber.receive() was waiting.
	if (isActive()) {
	RemoteCommandManagerException rcmException;
	if(messageIsNullException != null) {
	rcmException = RemoteCommandManagerException.errorReceivingJMSMessage(messageIsNullException);
	messageIsNullException = null;
	} else {
	rcmException = RemoteCommandManagerException.errorReceivingJMSMessage(e);
	}
	if (tm.shouldRemoveConnectionOnError()) {
	shouldReconnect = true;
	Object[] args = { getServiceId(), rcmException };
	rcm.logWarning("drop_connection_on_error", args);
	// after connection is closed isActive will return false.
	tm.removeLocalConnection();
	} else {
	try {
	// user has a chance to handle exception:
	// for instance to shut down the command manager.
	rcm.handleException(rcmException);
	} catch (RuntimeException ex) {
	// Ignore the exception, sleep before going back to listening.
	Object[] args = { toString(), rcmException, WAIT_ON_ERROR_RECEIVING_JMS_MESSAGE };
	rcm.logWarning("broadcast_listening_sleep_on_error", args);
	try {
	Thread.sleep(WAIT_ON_ERROR_RECEIVING_JMS_MESSAGE);
	} catch (InterruptedException interruptedEception) {
	// Ignore
	}
	}
	}
	}
	}
	}

	// Out of the loop - that means close method has been called.
	rcm.logDebug("broadcast_connection_stop_listening", getInfo());
	if(isClosing()) {
	try {
	// There is no need to close the sessions, producers, and consumers of a closed TopicConnection.
	topicConnection.close();
	} catch (JMSException closeException) {
	Object[] args = { displayString, closeException };
	rcm.logWarning("broadcast_exception_thrown_when_attempting_to_close_connection", args);
	} finally {
	rcm.logDebug("broadcast_connection_closed", getInfo());
	state = STATE_CLOSED;
	}
	}

	if(shouldReconnect && !tm.getRemoteCommandManager().isStopped()) {
	try {
	tm.createLocalConnection();
	} catch (RemoteCommandManagerException ex) {
	// Ignore exception - user had a chance to handle it in createLocalConnection method:
	// for instance to change host url and create a new local connection.
	}
	}
	}

	/**
	* INTERNAL:
	* Used for debug logging
	*/
	@Override
	protected void createDisplayString() {
	super.createDisplayString();
	displayString = (isLocal() ? "Local " : "External ") + displayString;
	}

	/**
	* INTERNAL:
	* Return whether a BroadcastConnection should check a ServiceId against its
	* own ServiceId to avoid the processing of Commands with the same ServiceId.
	* This should take place (return true) for a JMSTopicRemoteConnection.
	* @return boolean
	*/
	@Override
	protected boolean shouldCheckServiceId() {
	return true;
	}

	/**
	* INTERNAL:
	* set the TopicPublisher to be used when this RemoteConnection executes a command.
	* Setting the TopicPublisher avoids having it obtained on each executeCommandInternal
	* call. Passing in a publisher requires a TopicSession to also be set. These will
	* not be closed until the external RemoteConnection is closed, and then only if the
	* TopicConnection is also set.
	*/
	public void setPublisher(TopicPublisher publisher) {
	this.publisher = publisher;
	}

	public TopicPublisher getPublisher() {
	return publisher;
	}

	/**
	* INTERNAL:
	* set the TopicSubscriber on a local RemoteConnection for reading JMS messages when
	* this runnable connection is started in a thread. If setting this, a TopicConnection
	* is also required to be set, in order for it to be closed when the thread completes.
	* This is only to be used when using the JMSTopicRemoteConnection(rcm) constructor.
	*/
	public void setSuscriber(TopicSubscriber subscriber) {
	this.subscriber = subscriber;
	}

	public TopicSubscriber getSubscriber() {
	return subscriber;
	}

	/**
	* INTERNAL:
	* set the TopicSession to be used when this RemoteConnection executes a command if
	* the publisher is also set. Setting the TopicSession and Publisher avoids having
	* them obtained on each executeCommandInternal call. Passing in a TopicSession
	* requires a TopicPublisher to also be set. These will not be closed until the
	* external RemoteConnection is closed, and then only if the TopicConnection is also
	* set.
	*/
	public void setTopicSession(TopicSession topicSession) {
	this.topicSession = topicSession;
	}

	public TopicSession getTopicSession() {
	return topicSession;
	}

	/**
	* INTERNAL:
	* Set the TopicConnectionFactory, which is used if the publisher is not set to
	* obtain the TopicConnection, TopicSession and TopicPublisher
	*/
	public void setTopicConnectionFactory(TopicConnectionFactory topicConnectionFactory){
	this.topicConnectionFactory = topicConnectionFactory;
	}

	public TopicConnection getTopicConnectionFactory() {
	return topicConnection;
	}

	/**
	* INTERNAL:
	* Set the TopicConnection. If this is set, a Publisher and TopicSession must also
	* be set, or a new TopicConnection will be obtained on each executeCommandInternal
	* call. This TopicConnection is only used on close, as closing the TopicConnection
	* also closes any open TopicSessions and Publishers obtained from it.
	*/
	public void setTopicConnection(TopicConnection topicConnection){
	this.topicConnection = topicConnection;
	}

	public TopicConnection getTopicConnection() {
	return topicConnection;
	}

	/**
	* INTERNAL:
	* Set the Topic. The Topic is required with the TopicConnectionFactory to obtain connections
	* if the TopicPublisher is not set.
	*/
	public void setTopic(Topic topic){
	this.topic = topic;
	}

	public Topic getTopic() {
	return topic;
	}

	class JMSOnMessageHelper implements Runnable {
	Message message = null;

	public JMSOnMessageHelper(Message message) {
	this.message = message;
	}

	@Override
	public void run() {
	onMessage(message);
	}
	}
	}