Event listener use only one JMS Connection

git-svn-id: https://svn.argeo.org/slc/trunk@2415 4cfe0d0a-d680-48aa-b62c-e0a02a3f76cc

diff --git a/runtime/org.argeo.slc.support.activemq/src/main/java/org/argeo/slc/jms/JmsSlcEventListener.java b/runtime/org.argeo.slc.support.activemq/src/main/java/org/argeo/slc/jms/JmsSlcEventListener.java
index 0c702ab620e14fecd906f6978034c5ab95761216..5cea81d5841b88ca469052e6b5965aa22a5458b2 100644 (file)
--- a/runtime/org.argeo.slc.support.activemq/src/main/java/org/argeo/slc/jms/JmsSlcEventListener.java
+++ b/runtime/org.argeo.slc.support.activemq/src/main/java/org/argeo/slc/jms/JmsSlcEventListener.java
@@ -4,14 +4,12 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
 
 import javax.jms.Connection;
 import javax.jms.ConnectionFactory;
-import javax.jms.DeliveryMode;
-import javax.jms.Destination;
 import javax.jms.JMSException;
 import javax.jms.Message;
-import javax.jms.MessageConsumer;
 import javax.jms.Session;
 import javax.jms.Topic;
 import javax.jms.TopicSubscriber;
@@ -22,8 +20,7 @@ import org.argeo.slc.SlcException;
 import org.argeo.slc.msg.event.SlcEvent;
 import org.argeo.slc.msg.event.SlcEventListener;
 import org.argeo.slc.msg.event.SlcEventListenerDescriptor;
-import org.argeo.slc.msg.event.SlcEventListenerRegister;
-import org.springframework.jms.core.JmsTemplate;
+import org.springframework.jms.connection.ConnectionFactoryUtils;
 import org.springframework.jms.support.JmsUtils;
 import org.springframework.jms.support.converter.MessageConverter;
 
@@ -34,6 +31,10 @@ public class JmsSlcEventListener implements SlcEventListener {
        private ConnectionFactory jmsConnectionFactory;
        private MessageConverter messageConverter;
 
+       private Connection connection = null;
+       private String connectionClientId = getClass() + "#"
+                       + UUID.randomUUID().toString();
+
        private Map<String, ListeningClient> clients = Collections
                        .synchronizedMap(new HashMap<String, ListeningClient>());
 
@@ -56,27 +57,18 @@ public class JmsSlcEventListener implements SlcEventListener {
                        Object obj = null;
                        Session session = null;
                        TopicSubscriber topicSubscriber = null;
-                       // MessageConsumer messageConsumer = null;
                        try {
-                               // Connection connection = getClient(clientId).getConnection();
-                               // session = connection.createSession(false,
-                               // Session.AUTO_ACKNOWLEDGE);
                                session = getClient(clientId).getSession();
                                topicSubscriber = session.createDurableSubscriber(
                                                eventsDestination, clientId,
                                                createSelector(descriptors), true);
                                Message message = topicSubscriber.receive(timeout);
-                               // messageConsumer = session.createConsumer(eventsDestination,
-                               // createSelector(descriptors));
-                               // Message message = messageConsumer.receive(timeout);
                                obj = messageConverter.fromMessage(message);
                        } catch (JMSException e) {
                                throw new SlcException("Cannot poll events for client "
                                                + clientId, e);
                        } finally {
-                               // JmsUtils.closeMessageConsumer(messageConsumer);
                                JmsUtils.closeMessageConsumer(topicSubscriber);
-                               // JmsUtils.closeSession(session);
                        }
 
                        if (obj == null)
@@ -86,28 +78,6 @@ public class JmsSlcEventListener implements SlcEventListener {
                }
        }
 
-       /*
-        * public SlcEvent listen(SlcEventListenerRegister register, Long timeout) {
-        * 
-        * JmsTemplate jmsTemplate = new JmsTemplate(jmsConnectionFactory);
-        * jmsTemplate.setMessageConverter(messageConverter);
-        * jmsTemplate.setReceiveTimeout(timeout);
-        * 
-        * List<SlcEventListenerDescriptor> descriptors = register
-        * .getDescriptorsCopy();
-        * 
-        * if (descriptors.size() == 0) { // No listeners, just waiting try {
-        * Thread.sleep(timeout); } catch (InterruptedException e) { // silent }
-        * return null; } else { String selector = createSelector(descriptors);
-        * 
-        * if (log.isTraceEnabled()) log.debug("Selector: " + selector);
-        * 
-        * Object obj = jmsTemplate.receiveSelectedAndConvert( eventsDestination,
-        * selector);
-        * 
-        * if (obj == null) return null; else return (SlcEvent) obj; } }
-        */
-
        /** Returns null if no filter */
        protected String createSelector(List<SlcEventListenerDescriptor> descriptors) {
                if (descriptors.size() == 0)
@@ -145,25 +115,25 @@ public class JmsSlcEventListener implements SlcEventListener {
                this.messageConverter = messageConverter;
        }
 
-       public ListeningClient init(String clientId) {
-               Connection connection = null;
+       public void init() {
                try {
                        connection = jmsConnectionFactory.createConnection();
-                       connection.setClientID(clientId);
+                       connection.setClientID(connectionClientId);
                        connection.start();
-                       ListeningClient client = new ListeningClient(connection);
-                       return client;
                } catch (JMSException e) {
-                       throw new SlcException("Could not init listening client "
-                                       + clientId, e);
-               } finally {
+                       throw new SlcException("Could not init connection", e);
                }
        }
 
+       public void close() {
+               ConnectionFactoryUtils.releaseConnection(connection,
+                               jmsConnectionFactory, true);
+       }
+
        public void close(String clientId) {
                Session session = null;
                ListeningClient client = getClient(clientId);
-               Connection connection = client.getConnection();
+               // Connection connection = client.getConnection();
                try {
                        session = client.getSession();
                        session.unsubscribe(clientId);
@@ -176,22 +146,22 @@ public class JmsSlcEventListener implements SlcEventListener {
                // JmsUtils.closeSession(client.getSession());
                clients.remove(clientId);
 
-               try {
-                       connection.stop();
-                       connection.close();
-               } catch (JMSException e) {
-                       throw new SlcException("Could not close JMS connection for client "
-                                       + clientId, e);
-               } finally {
-                       clients.remove(clientId);
-               }
+               // try {
+               // connection.stop();
+               // connection.close();
+               // } catch (JMSException e) {
+               // throw new SlcException("Could not close JMS connection for client "
+               // + clientId, e);
+               // } finally {
+               // clients.remove(clientId);
+               // }
        }
 
        protected ListeningClient getClient(String clientId) {
                ListeningClient client = clients.get(clientId);
                if (client == null) {
                        // Lazy init
-                       client = init(clientId);
+                       client = new ListeningClient(connection);
                        clients.put(clientId, client);
                }
                return client;

diff --git a/server/org.argeo.slc.siteserver/bundles/org.argeo.slc.server.jms/META-INF/spring/jms.xml b/server/org.argeo.slc.siteserver/bundles/org.argeo.slc.server.jms/META-INF/spring/jms.xml
index 0edb121ac361734e0ba329e9e1cb8a6c8d3b5e92..ec69008247dd31bf4a662efeac362e8733ecc2b4 100644 (file)
--- a/server/org.argeo.slc.siteserver/bundles/org.argeo.slc.server.jms/META-INF/spring/jms.xml
+++ b/server/org.argeo.slc.siteserver/bundles/org.argeo.slc.server.jms/META-INF/spring/jms.xml
@@ -7,7 +7,8 @@
        <import resource="classpath:org/argeo/slc/activemq/spring.xml" />
 
        <!-- Events -->
-       <bean id="jmsEventListener" class="org.argeo.slc.jms.JmsSlcEventListener">
+       <bean id="jmsEventListener" class="org.argeo.slc.jms.JmsSlcEventListener"
+               init-method="init" destroy-method="close">
                <property name="jmsConnectionFactory">
                        <!--
                                Need its own connection factory in order to set client ids (not