runtime/org.argeo.slc.support.activemq/src/main/java/org/argeo/slc/jms/JmsSlcEventListener.java

   1 /*
   2  * Copyright (C) 2010 Mathieu Baudier <mbaudier@argeo.org>
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *         http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 package org.argeo.slc.jms;
  18
  19 import java.util.ArrayList;
  20 import java.util.List;
  21 import java.util.UUID;
  22
  23 import javax.jms.Connection;
  24 import javax.jms.ConnectionFactory;
  25 import javax.jms.JMSException;
  26 import javax.jms.Message;
  27 import javax.jms.Session;
  28 import javax.jms.Topic;
  29 import javax.jms.TopicSubscriber;
  30
  31 import org.apache.commons.logging.Log;
  32 import org.apache.commons.logging.LogFactory;
  33 import org.argeo.slc.SlcException;
  34 import org.argeo.slc.msg.event.SlcEvent;
  35 import org.argeo.slc.msg.event.SlcEventListener;
  36 import org.argeo.slc.msg.event.SlcEventListenerDescriptor;
  37 import org.springframework.jms.connection.ConnectionFactoryUtils;
  38 import org.springframework.jms.support.JmsUtils;
  39 import org.springframework.jms.support.converter.MessageConverter;
  40
  41 public class JmsSlcEventListener implements SlcEventListener {
  42         private final static Log log = LogFactory.getLog(JmsSlcEventListener.class);
  43
  44         // IoC
  45         private Topic eventsDestination;
  46         private ConnectionFactory jmsConnectionFactory;
  47         private MessageConverter messageConverter;
  48
  49         // Initialized with init() method, released with close()
  50         private Connection connection = null;
  51
  52         // One by instance
  53         private String connectionClientId = getClass() + "#"
  54                         + UUID.randomUUID().toString();
  55         private Boolean isClosed = false;
  56
  57         private List<String> subscriberIds = new ArrayList<String>();
  58
  59         // private Map<String, ListeningClient> clients = Collections
  60         // .synchronizedMap(new HashMap<String, ListeningClient>());
  61
  62         public SlcEvent listen(String subscriberId,
  63                         List<SlcEventListenerDescriptor> descriptors, Long timeout) {
  64                 if (descriptors.size() == 0) {
  65                         // No listener, just waiting
  66                         try {
  67                                 if (log.isTraceEnabled())
  68                                         log.trace("No event listener registered, sleeping...");
  69                                 Thread.sleep(timeout);
  70                         } catch (InterruptedException e) {
  71                                 // silent
  72                         }
  73                         return null;
  74                 } else {
  75                         Object obj = null;
  76                         synchronized (subscriberIds) {
  77                                 while (subscriberIds.contains(subscriberId)) {
  78                                         try {
  79                                                 subscriberIds.wait(500);
  80                                                 if (isClosed)
  81                                                         return null;
  82                                         } catch (InterruptedException e) {
  83                                                 // silent
  84                                         }
  85                                 }
  86                                 subscriberIds.add(subscriberId);
  87                                 Session session = null;
  88                                 TopicSubscriber topicSubscriber = null;
  89                                 try {
  90                                         // ListeningClient client = (ListeningClient)
  91                                         // getClient(clientId);
  92                                         session = connection.createSession(false,
  93                                                         Session.AUTO_ACKNOWLEDGE);
  94                                         topicSubscriber = session.createDurableSubscriber(
  95                                                         eventsDestination, subscriberId,
  96                                                         createSelector(descriptors), true);
  97                                         Message message = topicSubscriber.receive(timeout);
  98                                         obj = messageConverter.fromMessage(message);
  99                                 } catch (JMSException e) {
 100                                         throw new SlcException("Cannot poll events for subscriber "
 101                                                         + subscriberId, e);
 102                                 } finally {
 103                                         JmsUtils.closeMessageConsumer(topicSubscriber);
 104                                         JmsUtils.closeSession(session);
 105                                         subscriberIds.remove(subscriberId);
 106                                         subscriberIds.notifyAll();
 107                                 }
 108
 109                         }
 110
 111                         if (obj == null)
 112                                 return null;
 113                         else
 114                                 return (SlcEvent) obj;
 115                 }
 116         }
 117
 118         /** Returns null if no filter */
 119         protected String createSelector(List<SlcEventListenerDescriptor> descriptors) {
 120                 if (descriptors.size() == 0)
 121                         throw new SlcException("No listeners, cannot generate JMS selector");
 122
 123                 StringBuffer buf = new StringBuffer(256);
 124                 Boolean first = true;
 125                 for (SlcEventListenerDescriptor descriptor : descriptors) {
 126                         if (first)
 127                                 first = false;
 128                         else
 129                                 buf.append(" OR ");
 130
 131                         buf.append('(');
 132                         buf.append(SlcEvent.EVENT_TYPE).append("=").append('\'').append(
 133                                         descriptor.getEventType()).append('\'');
 134                         if (descriptor.getFilter() != null) {
 135                                 buf.append(" AND ");
 136                                 buf.append('(').append(descriptor.getFilter()).append(')');
 137                         }
 138                         buf.append(')');
 139                 }
 140                 if (log.isTraceEnabled())
 141                         log.trace("selector created : " + buf.toString());
 142                 return buf.toString();
 143         }
 144
 145         public boolean isClosed() {
 146                 return isClosed;
 147         }
 148
 149         // Ioc
 150         public void setEventsDestination(Topic eventsDestination) {
 151                 this.eventsDestination = eventsDestination;
 152         }
 153
 154         public void setJmsConnectionFactory(ConnectionFactory jmsConnectionFactory) {
 155                 this.jmsConnectionFactory = jmsConnectionFactory;
 156         }
 157
 158         public void setMessageConverter(MessageConverter messageConverter) {
 159                 this.messageConverter = messageConverter;
 160         }
 161
 162         // Life Cycle
 163         public void init() {
 164                 try {
 165                         connection = jmsConnectionFactory.createConnection();
 166                         connection.setClientID(connectionClientId);
 167                         connection.start();
 168                 } catch (JMSException e) {
 169                         throw new SlcException("Could not init connection", e);
 170                 }
 171         }
 172
 173         public void close() {
 174                 ConnectionFactoryUtils.releaseConnection(connection,
 175                                 jmsConnectionFactory, true);
 176                 isClosed = true;
 177                 synchronized (subscriberIds) {
 178                         subscriberIds.notifyAll();
 179                 }
 180         }
 181
 182         // public void close(String clientId) {
 183         // // Session session = null;
 184         // // // ListeningClient client = getClient(clientId);
 185         // // // Connection connection = client.getConnection();
 186         // // try {
 187         // // session = client.getSession();
 188         // // session.unsubscribe(clientId);
 189         // // } catch (JMSException e) {
 190         // // log.warn("Could not unsubscribe client " + clientId, e);
 191         // // } finally {
 192         // // JmsUtils.closeSession(session);
 193         // // }
 194         // //
 195         // // // synchronized (client) {
 196         // // // clients.remove(clientId);
 197         // // // client.notify();
 198         // // // }
 199         // }
 200
 201         // protected ListeningClient getClient(String clientId) {
 202         // ListeningClient client = clients.get(clientId);
 203         // if (client == null) {
 204         // // Lazy init
 205         // client = new ListeningClient(connection);
 206         // clients.put(clientId, client);
 207         // }
 208         // return client;
 209         // }
 210
 211         // protected class ListeningClient {
 212         // private final Connection connection;
 213         // private final Session session;
 214         //
 215         // public ListeningClient(Connection connection) {
 216         // super();
 217         // this.connection = connection;
 218         // try {
 219         // session = connection.createSession(false,
 220         // Session.AUTO_ACKNOWLEDGE);
 221         // } catch (JMSException e) {
 222         // throw new SlcException("Cannot create session");
 223         // }
 224         // }
 225         //
 226         // public Connection getConnection() {
 227         // return connection;
 228         // }
 229         //
 230         // public Session getSession() {
 231         // return session;
 232         // }
 233         //
 234         // }
 235 }