* 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).
*/
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.
}
}