Index: src/main/java/org/apache/axis2/transport/jms/ServiceTaskManager.java =================================================================== --- src/main/java/org/apache/axis2/transport/jms/ServiceTaskManager.java (revision 1088474) +++ src/main/java/org/apache/axis2/transport/jms/ServiceTaskManager.java (working copy) @@ -26,15 +26,16 @@ import javax.jms.*; import javax.jms.IllegalStateException; +import javax.naming.Context; import javax.naming.InitialContext; -import javax.naming.Context; import javax.naming.NamingException; -import javax.transaction.UserTransaction; import javax.transaction.NotSupportedException; +import javax.transaction.Status; import javax.transaction.SystemException; -import javax.transaction.Status; +import javax.transaction.UserTransaction; import java.util.*; import java.util.concurrent.atomic.AtomicInteger; +import java.text.SimpleDateFormat; /** * Each service will have one ServiceTaskManager instance that will create, manage and also destroy @@ -144,6 +145,10 @@ /** The JMS Connection shared between multiple polling tasks - when enabled (reccomended) */ private Connection sharedConnection = null; + private static int exponent = 1; + + private static SimpleDateFormat sdf = new SimpleDateFormat("MMM dd,yyyy HH:mm"); + /** * Start or re-start the Task Manager by shutting down any existing worker tasks and * re-creating them. However, if this is STM is PAUSED, a start request is ignored. @@ -163,7 +168,7 @@ } if (cacheLevel == JMSConstants.CACHE_AUTO) { - cacheLevel = + cacheLevel = transactionality == BaseConstants.TRANSACTION_NONE ? JMSConstants.CACHE_CONSUMER : JMSConstants.CACHE_NONE; } @@ -409,7 +414,29 @@ } // Get a message by polling, or receive null - Message message = receiveMessage(); + Message message = null; + try { + message = receiveMessage(); + } catch (Exception e) { +// if (e instanceof AxisJMSConnectionRefusedException) { +// // if the is a connection refused error re-inilize the endpoint +// log.error("Listener Task receives a connection refused error", e); +// long nextSleepTime = (long) Math.exp(exponent); +// exponent = exponent + 1; +// long finalTime = nextSleepTime + exponent * 1000; +// Date newDate = new Date(finalTime); +// log.info("Next re-try will be after" + nextSleepTime + " seconds at '" + +// sdf.format(newDate)); +// try { +// Thread.sleep(nextSleepTime * 1000); +// } catch (InterruptedException e1) { +// // ignore +// } +// log.info("Re-inilizing the listner task for the service '" + +// serviceName + "' because it receives a connection refused error"); +// scheduleNewTaskIfAppropriate(); +// } + } if (log.isTraceEnabled()) { if (message != null) { @@ -439,32 +466,32 @@ } } finally { - + if (log.isTraceEnabled()) { log.trace("Listener task with Thread ID : " + Thread.currentThread().getId() + " is stopping after processing : " + messageCount + " messages :: " + " isActive : " + isActive() + " maxMessagesPerTask : " + getMaxMessagesPerTask() + " concurrentConsumers : " + getConcurrentConsumers() + - " idleExecutionCount : " + idleExecutionCount + " idleTaskExecutionLimit : " + + " idleExecutionCount : " + idleExecutionCount + " idleTaskExecutionLimit : " + getIdleTaskExecutionLimit()); } else if (log.isDebugEnabled()) { log.debug("Listener task with Thread ID : " + Thread.currentThread().getId() + " is stopping after processing : " + messageCount + " messages"); } - + // Close the consumer and session before decrementing activeTaskCount. // (If we have a shared connection, Qpid deadlocks if the shared connection // is closed on another thread while closing the session) closeConsumer(true); closeSession(true); closeConnection(); - + workerState = STATE_STOPPED; activeTaskCount--; synchronized(pollingTasks) { pollingTasks.remove(this); } - + // My time is up, so if I am going away, create another scheduleNewTaskIfAppropriate(); } @@ -677,10 +704,10 @@ connection = createConnection(); setConnected(true); } - + } else if (connection == null) { // Connection is shared, but may not have been created - + synchronized(ServiceTaskManager.this) { if (sharedConnection == null) { sharedConnection = createConnection(); @@ -688,16 +715,15 @@ } connection = sharedConnection; setConnected(true); - + } // else: Connection is shared and is already referenced by this.connection - + return connection; } /** * Get a Session that could/should be used by this task - depends on the cache level to reuse - * @param connection the connection (could be the shared connection) to use to create a Session * @return the shared Session if cache level is higher than CACHE_CONNECTION, or a new Session * created using the Connection passed, or a new/shared connection */ @@ -711,8 +737,6 @@ /** * Get a MessageConsumer that chould/should be used by this task - depends on the cache * level to reuse - * @param connection option Connection to be used - * @param session optional Session to be used * @return the shared MessageConsumer if cache level is higher than CACHE_SESSION, or a new * MessageConsumer possibly using the Connection and Session passed in */ @@ -725,7 +749,6 @@ /** * Close the given Connection, hiding exceptions if any which are logged - * @param connection the Connection to be closed */ private void closeConnection() { if (connection != null && @@ -745,7 +768,7 @@ /** * Close the given Session, hiding exceptions if any which are logged - * @param session the Session to be closed + * @param forced should the session should close forcefully */ private void closeSession(boolean forced) { if (session != null && @@ -765,7 +788,7 @@ /** * Close the given Consumer, hiding exceptions if any which are logged - * @param consumer the Consumer to be closed + * @param forced forcefully closed the consumer */ private void closeConsumer(boolean forced) { if (consumer != null && @@ -811,16 +834,19 @@ connection.start(); log.debug("JMS Connection for service : " + serviceName + " created and started"); - } catch (JMSException e) { - handleException("Error acquiring a JMS connection to : " + getConnFactoryJNDIName() + - " using JNDI properties : " + jmsProperties, e); + } catch (Exception e) { + if (e.getMessage() != null && e.getMessage().contains("Connection refused")) { + handleConnectionRefusedException(e); + } else { + handleException("Error acquiring a JMS connection to : " + getConnFactoryJNDIName() + + " using JNDI properties : " + jmsProperties, e); + } } return connection; } /** * Create a new Session for this STM - * @param connection the Connection to be used * @return a new Session created using the Connection passed in */ private Session createSession() { @@ -839,7 +865,6 @@ /** * Create a new MessageConsumer for this STM - * @param session the Session to be used * @return a new MessageConsumer created using the Session passed in */ private MessageConsumer createConsumer() { @@ -867,7 +892,7 @@ /** * Get the InitialContext for lookup using the JNDI parameters applicable to the service * @return the InitialContext to be used - * @throws NamingException + * @throws NamingException throws in case of an error */ private Context getInitialContext() throws NamingException { if (context == null) { @@ -878,7 +903,7 @@ /** * Return the JMS Destination for the JNDI name of the Destination from the InitialContext - * @param session which is used to create the destinations if not present and if possible + * @param session which is used to create the destinations if not present and if possible * @return the JMS Destination to which this STM listens for messages */ private Destination getDestination(Session session) { @@ -937,7 +962,7 @@ " using JNDI properties : " + jmsProperties, e); } } - + if (sharedUserTransaction == null) { try { context = getInitialContext(); @@ -967,7 +992,7 @@ if (destinationType == JMSConstants.GENERIC) { return null; } else { - return destinationType == JMSConstants.QUEUE; + return destinationType == JMSConstants.QUEUE; } } @@ -985,6 +1010,11 @@ throw new AxisJMSException(msg); } + private void handleConnectionRefusedException(Exception e){ + log.error(e); + throw new AxisJMSConnectionRefusedException(e); + } + // -------------- getters and setters ------------------ public String getServiceName() { return serviceName; @@ -1221,16 +1251,16 @@ public int getActiveTaskCount() { return activeTaskCount; } - + /** * Get the number of existing JMS message consumers. - * + * * @return the number of consumers */ public int getConsumerCount() { return consumerCount.get(); } - + public void setServiceTaskManagerState(int serviceTaskManagerState) { this.serviceTaskManagerState = serviceTaskManagerState; } Index: src/main/java/org/apache/axis2/transport/jms/AxisJMSConnectionRefusedException.java =================================================================== --- src/main/java/org/apache/axis2/transport/jms/AxisJMSConnectionRefusedException.java (revision 0) +++ src/main/java/org/apache/axis2/transport/jms/AxisJMSConnectionRefusedException.java (revision 0) @@ -0,0 +1,30 @@ +/* + * Copyright WSO2, Inc. (http://wso2.com) + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.axis2.transport.jms; + +public class AxisJMSConnectionRefusedException extends RuntimeException{ + AxisJMSConnectionRefusedException(){ + super(); + } + + AxisJMSConnectionRefusedException(Exception e){ + super(e); + } + + AxisJMSConnectionRefusedException(String msg, Exception e){ + super(msg, e); + } +}