001 /*
002 GRANITE DATA SERVICES
003 Copyright (C) 2011 GRANITE DATA SERVICES S.A.S.
004
005 This file is part of Granite Data Services.
006
007 Granite Data Services is free software; you can redistribute it and/or modify
008 it under the terms of the GNU Library General Public License as published by
009 the Free Software Foundation; either version 2 of the License, or (at your
010 option) any later version.
011
012 Granite Data Services is distributed in the hope that it will be useful, but
013 WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
014 FITNESS FOR A PARTICULAR PURPOSE. See the GNU Library General Public License
015 for more details.
016
017 You should have received a copy of the GNU Library General Public License
018 along with this library; if not, see <http://www.gnu.org/licenses/>.
019 */
020
021 package org.granite.gravity.adapters;
022
023 import java.io.ByteArrayInputStream;
024 import java.io.ByteArrayOutputStream;
025 import java.io.IOException;
026 import java.io.Serializable;
027 import java.util.Date;
028 import java.util.Enumeration;
029 import java.util.HashMap;
030 import java.util.Map;
031 import java.util.Properties;
032 import java.util.Timer;
033 import java.util.TimerTask;
034
035 import javax.jms.ConnectionFactory;
036 import javax.jms.Destination;
037 import javax.jms.ExceptionListener;
038 import javax.jms.JMSException;
039 import javax.jms.MessageListener;
040 import javax.jms.ObjectMessage;
041 import javax.jms.Session;
042 import javax.jms.TextMessage;
043 import javax.naming.Context;
044 import javax.naming.InitialContext;
045 import javax.naming.NamingException;
046
047 import org.granite.clustering.DistributedDataFactory;
048 import org.granite.clustering.TransientReference;
049 import org.granite.context.GraniteContext;
050 import org.granite.gravity.Channel;
051 import org.granite.gravity.Gravity;
052 import org.granite.gravity.MessageReceivingException;
053 import org.granite.logging.Logger;
054 import org.granite.messaging.amf.io.AMF3Deserializer;
055 import org.granite.messaging.amf.io.AMF3Serializer;
056 import org.granite.messaging.service.ServiceException;
057 import org.granite.messaging.webapp.ServletGraniteContext;
058 import org.granite.util.XMap;
059
060 import flex.messaging.messages.AcknowledgeMessage;
061 import flex.messaging.messages.AsyncMessage;
062 import flex.messaging.messages.CommandMessage;
063 import flex.messaging.messages.ErrorMessage;
064
065 /**
066 * @author William DRAI
067 */
068 public class JMSServiceAdapter extends ServiceAdapter {
069
070 private static final Logger log = Logger.getLogger(JMSServiceAdapter.class);
071
072 public static final long DEFAULT_FAILOVER_RETRY_INTERVAL = 1000L;
073 public static final long DEFAULT_RECONNECT_RETRY_INTERVAL = 20000L;
074 public static final int DEFAULT_FAILOVER_RETRY_COUNT = 4;
075
076 protected ConnectionFactory jmsConnectionFactory = null;
077 protected javax.jms.Destination jmsDestination = null;
078 protected Map<String, JMSClient> jmsClients = new HashMap<String, JMSClient>();
079 protected String destinationName = null;
080 protected boolean textMessages = false;
081 protected boolean transactedSessions = false;
082 protected int acknowledgeMode = Session.AUTO_ACKNOWLEDGE;
083 protected int messagePriority = javax.jms.Message.DEFAULT_PRIORITY;
084 protected int deliveryMode = javax.jms.Message.DEFAULT_DELIVERY_MODE;
085 protected boolean noLocal = false;
086 protected boolean sessionSelector = false;
087
088 protected long failoverRetryInterval = DEFAULT_FAILOVER_RETRY_INTERVAL;
089 protected int failoverRetryCount = DEFAULT_FAILOVER_RETRY_COUNT;
090 protected long reconnectRetryInterval = DEFAULT_RECONNECT_RETRY_INTERVAL;
091
092 @Override
093 public void configure(XMap adapterProperties, XMap destinationProperties) throws ServiceException {
094 super.configure(adapterProperties, destinationProperties);
095
096 log.info("Using JMS configuration: %s", destinationProperties.getOne("jms"));
097
098 destinationName = destinationProperties.get("jms/destination-name");
099
100 if (Boolean.TRUE.toString().equals(destinationProperties.get("jms/transacted-sessions")))
101 transactedSessions = true;
102
103 String ackMode = destinationProperties.get("jms/acknowledge-mode");
104 if ("AUTO_ACKNOWLEDGE".equals(ackMode))
105 acknowledgeMode = Session.AUTO_ACKNOWLEDGE;
106 else if ("CLIENT_ACKNOWLEDGE".equals(ackMode))
107 acknowledgeMode = Session.CLIENT_ACKNOWLEDGE;
108 else if ("DUPS_OK_ACKNOWLEDGE".equals(ackMode))
109 acknowledgeMode = Session.DUPS_OK_ACKNOWLEDGE;
110 else if (ackMode != null)
111 log.warn("Unsupported acknowledge mode: %s (using default AUTO_ACKNOWLEDGE)", ackMode);
112
113 if ("javax.jms.TextMessage".equals(destinationProperties.get("jms/message-type")))
114 textMessages = true;
115
116 if (Boolean.TRUE.toString().equals(destinationProperties.get("jms/no-local")))
117 noLocal = true;
118
119 if (Boolean.TRUE.toString().equals(destinationProperties.get("session-selector")))
120 sessionSelector = true;
121
122 failoverRetryInterval = destinationProperties.get("jms/failover-retry-interval", Long.TYPE, DEFAULT_FAILOVER_RETRY_INTERVAL);
123 if (failoverRetryInterval <= 0) {
124 log.warn("Illegal failover retry interval: %d (using default %d)", failoverRetryInterval, DEFAULT_FAILOVER_RETRY_INTERVAL);
125 failoverRetryInterval = DEFAULT_FAILOVER_RETRY_INTERVAL;
126 }
127
128 failoverRetryCount = destinationProperties.get("jms/failover-retry-count", Integer.TYPE, DEFAULT_FAILOVER_RETRY_COUNT);
129 if (failoverRetryCount <= 0) {
130 log.warn("Illegal failover retry count: %s (using default %d)", failoverRetryCount, DEFAULT_FAILOVER_RETRY_COUNT);
131 failoverRetryCount = DEFAULT_FAILOVER_RETRY_COUNT;
132 }
133
134 reconnectRetryInterval = destinationProperties.get("jms/reconnect-retry-interval", Long.TYPE, DEFAULT_RECONNECT_RETRY_INTERVAL);
135 if (reconnectRetryInterval <= 0) {
136 log.warn("Illegal reconnect retry interval: %d (using default %d)", reconnectRetryInterval, DEFAULT_RECONNECT_RETRY_INTERVAL);
137 reconnectRetryInterval = DEFAULT_RECONNECT_RETRY_INTERVAL;
138 }
139
140 Properties environment = new Properties();
141 for (XMap property : destinationProperties.getAll("jms/initial-context-environment/property")) {
142 String name = property.get("name");
143 String value = property.get("value");
144
145 if ("Context.PROVIDER_URL".equals(name))
146 environment.put(Context.PROVIDER_URL, value);
147 else if ("Context.INITIAL_CONTEXT_FACTORY".equals(name))
148 environment.put(Context.INITIAL_CONTEXT_FACTORY, value);
149 else if ("Context.URL_PKG_PREFIXES".equals(name))
150 environment.put(Context.URL_PKG_PREFIXES, value);
151 else if ("Context.SECURITY_PRINCIPAL".equals(name))
152 environment.put(Context.SECURITY_PRINCIPAL, value);
153 else if ("Context.SECURITY_CREDENTIALS".equals(name))
154 environment.put(Context.SECURITY_CREDENTIALS, value);
155 else
156 log.warn("Unknown InitialContext property: %s (ignored)", name);
157 }
158
159 InitialContext initialContext = null;
160 try {
161 initialContext = new InitialContext(environment.size() > 0 ? environment : null);
162 }
163 catch (NamingException e) {
164 log.error(e, "Could not initialize JNDI context");
165 throw new ServiceException("Error configuring JMS Adapter", e);
166 }
167
168 String cfJndiName = destinationProperties.get("jms/connection-factory");
169 try {
170 jmsConnectionFactory = (ConnectionFactory)initialContext.lookup(cfJndiName);
171 }
172 catch (NamingException e) {
173 log.error(e, "Could not find JMS ConnectionFactory named %s in JNDI", cfJndiName);
174 throw new ServiceException("Error configuring JMS Adapter", e);
175 }
176
177 String dsJndiName = destinationProperties.get("jms/destination-jndi-name");
178 try {
179 jmsDestination = (Destination)initialContext.lookup(dsJndiName);
180 }
181 catch (NamingException e) {
182 log.error(e, "Could not find JMS destination named %s in JNDI", dsJndiName);
183 throw new ServiceException("Error configuring JMS Adapter", e);
184 }
185 }
186
187 protected javax.jms.Destination getProducerDestination(String topic) {
188 return jmsDestination;
189 }
190
191 protected javax.jms.Destination getConsumerDestination(String topic) {
192 return jmsDestination;
193 }
194
195 @Override
196 public void start() throws ServiceException {
197 super.start();
198 }
199
200 @Override
201 public void stop() throws ServiceException {
202 super.stop();
203
204 for (JMSClient jmsClient : jmsClients.values()) {
205 try {
206 jmsClient.close();
207 }
208 catch (Exception e) {
209 log.warn(e, "Could not close JMSClient: %s", jmsClient);
210 }
211 }
212 jmsClients.clear();
213 }
214
215
216 private synchronized JMSClient connectJMSClient(Channel client, String destination) throws Exception {
217 JMSClient jmsClient = jmsClients.get(client.getId());
218 if (jmsClient == null) {
219 jmsClient = new JMSClientImpl(client);
220 jmsClient.connect();
221 jmsClients.put(client.getId(), jmsClient);
222 if (sessionSelector && GraniteContext.getCurrentInstance() instanceof ServletGraniteContext)
223 ((ServletGraniteContext)GraniteContext.getCurrentInstance()).getSessionMap().put(JMSClient.JMSCLIENT_KEY_PREFIX + destination, jmsClient);
224 log.debug("JMS client connected for channel " + client.getId());
225 }
226 return jmsClient;
227 }
228
229 private synchronized void closeJMSClientIfNecessary(Channel channel, String destination) throws Exception {
230 JMSClient jmsClient = jmsClients.get(channel.getId());
231 if (jmsClient != null && !jmsClient.hasActiveConsumer()) {
232 jmsClient.close();
233 jmsClients.remove(channel.getId());
234 if (sessionSelector && GraniteContext.getCurrentInstance() instanceof ServletGraniteContext)
235 ((ServletGraniteContext)GraniteContext.getCurrentInstance()).getSessionMap().remove(JMSClient.JMSCLIENT_KEY_PREFIX + destination);
236 log.debug("JMS client closed for channel " + channel.getId());
237 }
238 }
239
240 @Override
241 public Object invoke(Channel fromClient, AsyncMessage message) {
242 String topicId = (String)message.getHeader(AsyncMessage.SUBTOPIC_HEADER);
243
244 if (getSecurityPolicy().canPublish(fromClient, topicId, message)) {
245 try {
246 JMSClient jmsClient = connectJMSClient(fromClient, message.getDestination());
247 jmsClient.send(message);
248
249 AsyncMessage reply = new AcknowledgeMessage(message);
250 reply.setMessageId(message.getMessageId());
251
252 return reply;
253 }
254 catch (Exception e) {
255 log.error(e, "Error sending message");
256 ErrorMessage error = new ErrorMessage(message, null);
257 error.setFaultString("JMS Adapter error " + e.getMessage());
258
259 return error;
260 }
261 }
262
263 log.debug("Channel %s tried to publish a message to topic %s", fromClient, topicId);
264 ErrorMessage error = new ErrorMessage(message, null);
265 error.setFaultString("Server.Publish.Denied");
266 return error;
267 }
268
269 @Override
270 public Object manage(Channel fromChannel, CommandMessage message) {
271 String topicId = (String)message.getHeader(AsyncMessage.SUBTOPIC_HEADER);
272
273 if (message.getOperation() == CommandMessage.SUBSCRIBE_OPERATION) {
274 if (getSecurityPolicy().canSubscribe(fromChannel, topicId, message)) {
275 try {
276 JMSClient jmsClient = connectJMSClient(fromChannel, message.getDestination());
277 jmsClient.subscribe(message);
278
279 AsyncMessage reply = new AcknowledgeMessage(message);
280 return reply;
281 }
282 catch (Exception e) {
283 throw new RuntimeException("JMSAdapter subscribe error on topic: " + message, e);
284 }
285 }
286
287 log.debug("Channel %s tried to subscribe to topic %s", fromChannel, topicId);
288 ErrorMessage error = new ErrorMessage(message, null);
289 error.setFaultString("Server.Subscribe.Denied");
290 return error;
291 }
292 else if (message.getOperation() == CommandMessage.UNSUBSCRIBE_OPERATION) {
293 try {
294 JMSClient jmsClient = connectJMSClient(fromChannel, message.getDestination());
295 jmsClient.unsubscribe(message);
296 closeJMSClientIfNecessary(fromChannel, message.getDestination());
297
298 AsyncMessage reply = new AcknowledgeMessage(message);
299 return reply;
300 }
301 catch (Exception e) {
302 throw new RuntimeException("JMSAdapter unsubscribe error on topic: " + message, e);
303 }
304 }
305
306 return null;
307 }
308
309
310 @TransientReference
311 private class JMSClientImpl implements JMSClient {
312
313 private Channel channel = null;
314 private String topic = null;
315 private javax.jms.Connection jmsConnection = null;
316 private javax.jms.Session jmsProducerSession = null;
317 private javax.jms.MessageProducer jmsProducer = null;
318 private Map<String, JMSConsumer> consumers = new HashMap<String, JMSConsumer>();
319 private boolean useGlassFishNoExceptionListenerWorkaround = false;
320 private boolean useGlassFishNoCommitWorkaround = false;
321
322 private ExceptionListener connectionExceptionListener = new ConnectionExceptionListener();
323
324 private class ConnectionExceptionListener implements ExceptionListener {
325
326 public void onException(JMSException ex) {
327 // Connection failure, force reconnection of the producer on next send
328 jmsProducer = null;
329 for (JMSConsumer consumer : consumers.values())
330 consumer.reset();
331 consumers.clear();
332 jmsConnection = null;
333 jmsProducerSession = null;
334 }
335 }
336
337
338 public JMSClientImpl(Channel channel) {
339 this.channel = channel;
340 }
341
342 public boolean hasActiveConsumer() {
343 return consumers != null && !consumers.isEmpty();
344 }
345
346
347 public void connect() throws ServiceException {
348 if (jmsConnection != null)
349 return;
350
351 try {
352 jmsConnection = jmsConnectionFactory.createConnection();
353 if (!useGlassFishNoExceptionListenerWorkaround) {
354 try {
355 jmsConnection.setExceptionListener(connectionExceptionListener);
356 }
357 catch (JMSException e) {
358 if (e.getMessage().startsWith("MQJMSRA_DC2001: Unsupported:setExceptionListener()"))
359 useGlassFishNoExceptionListenerWorkaround = true;
360 else
361 throw e;
362 }
363 }
364 jmsConnection.start();
365 log.debug("JMS client connected for channel " + channel.getId());
366 }
367 catch (JMSException e) {
368 throw new ServiceException("JMS Initialize error", e);
369 }
370 }
371
372 public void close() throws ServiceException {
373 try {
374 if (jmsProducer != null)
375 jmsProducer.close();
376 }
377 catch (JMSException e) {
378 log.error(e, "Could not close JMS Producer for channel " + channel.getId());
379 }
380 finally {
381 try {
382 if (jmsProducerSession != null)
383 jmsProducerSession.close();
384 }
385 catch (JMSException e) {
386 log.error(e, "Could not close JMS Producer Session for channel " + channel.getId());
387 }
388 }
389 for (JMSConsumer consumer : consumers.values()) {
390 try {
391 consumer.close();
392 }
393 catch (JMSException e) {
394 log.error(e, "Could not close JMS Consumer " + consumer.subscriptionId + " for channel " + channel.getId());
395 }
396 }
397 try {
398 jmsConnection.stop();
399 }
400 catch (JMSException e) {
401 log.debug(e, "Could not stop JMS Connection for channel " + channel.getId());
402 }
403 finally {
404 try {
405 jmsConnection.close();
406 }
407 catch (JMSException e) {
408 throw new ServiceException("JMS Stop error", e);
409 }
410 finally {
411 consumers.clear();
412 }
413 }
414 }
415
416 private void createProducer(String topic) throws Exception {
417 try {
418 // When failing over, JMS can be in a temporary illegal state. Give it some time to recover.
419 int retryCount = failoverRetryCount;
420 do {
421 try {
422 jmsProducer = jmsProducerSession.createProducer(getProducerDestination(topic != null ? topic : this.topic));
423 if (retryCount < failoverRetryCount) // We come from a failover, try to recover session
424 jmsProducerSession.recover();
425 break;
426 }
427 catch (Exception e) {
428 if (retryCount <= 0)
429 throw e;
430
431 if (log.isDebugEnabled())
432 log.debug(e, "Could not create JMS Producer (retrying %d time)", retryCount);
433 else
434 log.info("Could not create JMS Producer (retrying %d time)", retryCount);
435
436 try {
437 Thread.sleep(failoverRetryInterval);
438 }
439 catch (Exception f) {
440 throw new ServiceException("Could not sleep when retrying to create JMS Producer", f.getMessage(), e);
441 }
442 }
443 }
444 while (retryCount-- > 0);
445
446 jmsProducer.setPriority(messagePriority);
447 jmsProducer.setDeliveryMode(deliveryMode);
448 log.debug("Created JMS Producer for channel %s", channel.getId());
449 }
450 catch (JMSException e) {
451 jmsProducerSession.close();
452 jmsProducerSession = null;
453 throw e;
454 }
455 }
456
457 public void send(AsyncMessage message) throws Exception {
458 Object msg = null;
459 if (Boolean.TRUE.equals(message.getHeader(Gravity.BYTEARRAY_BODY_HEADER))) {
460 byte[] byteArray = (byte[])message.getBody();
461 ByteArrayInputStream bais = new ByteArrayInputStream(byteArray);
462 AMF3Deserializer deser = new AMF3Deserializer(bais);
463 msg = deser.readObject();
464 deser.close(); // makes jdk7 happy (Resource leak: 'deser' is never closed)...
465 }
466 else
467 msg = message.getBody();
468
469 internalSend(message.getHeaders(), msg, message.getMessageId(), message.getCorrelationId(), message.getTimestamp(), message.getTimeToLive());
470 }
471
472 public void send(Map<String, ?> params, Object msg, long timeToLive) throws Exception {
473 internalSend(params, msg, null, null, new Date().getTime(), timeToLive);
474 }
475
476 public void internalSend(Map<String, ?> headers, Object msg, String messageId, String correlationId, long timestamp, long timeToLive) throws Exception {
477 String topic = (String)headers.get(AsyncMessage.SUBTOPIC_HEADER);
478
479 if (jmsProducerSession == null) {
480 jmsProducerSession = jmsConnection.createSession(transactedSessions, acknowledgeMode);
481 log.debug("Created JMS Producer Session for channel %s (transacted: %s, ack: %s)", channel.getId(), transactedSessions, acknowledgeMode);
482 }
483
484 if (jmsProducer == null)
485 createProducer(topic);
486
487 javax.jms.Message jmsMessage = null;
488 if (textMessages)
489 jmsMessage = jmsProducerSession.createTextMessage(msg.toString());
490 else
491 jmsMessage = jmsProducerSession.createObjectMessage((Serializable)msg);
492
493 jmsMessage.setJMSMessageID(normalizeJMSMessageID(messageId));
494 jmsMessage.setJMSCorrelationID(normalizeJMSMessageID(correlationId));
495 jmsMessage.setJMSTimestamp(timestamp);
496 jmsMessage.setJMSExpiration(timeToLive);
497
498 for (Map.Entry<String, ?> me : headers.entrySet()) {
499 if ("JMSType".equals(me.getKey())) {
500 if (me.getValue() instanceof String)
501 jmsMessage.setJMSType((String)me.getValue());
502 }
503 else if ("JMSPriority".equals(me.getKey())) {
504 if (me.getValue() instanceof Integer)
505 jmsMessage.setJMSPriority(((Integer)me.getValue()).intValue());
506 }
507 else if (me.getValue() instanceof String)
508 jmsMessage.setStringProperty(me.getKey(), (String)me.getValue());
509 else if (me.getValue() instanceof Boolean)
510 jmsMessage.setBooleanProperty(me.getKey(), ((Boolean)me.getValue()).booleanValue());
511 else if (me.getValue() instanceof Integer)
512 jmsMessage.setIntProperty(me.getKey(), ((Integer)me.getValue()).intValue());
513 else if (me.getValue() instanceof Long)
514 jmsMessage.setLongProperty(me.getKey(), ((Long)me.getValue()).longValue());
515 else if (me.getValue() instanceof Double)
516 jmsMessage.setDoubleProperty(me.getKey(), ((Double)me.getValue()).doubleValue());
517 else
518 jmsMessage.setObjectProperty(me.getKey(), me.getValue());
519 }
520
521 jmsProducer.send(jmsMessage);
522
523 if (transactedSessions && !useGlassFishNoCommitWorkaround) {
524 // If we are in a container-managed transaction (data dispatch from an EJB interceptor for ex.), we should not commit the session
525 // but the behaviour is different between JBoss and GlassFish
526 try {
527 jmsProducerSession.commit();
528 }
529 catch (JMSException e) {
530 if (e.getMessage() != null && e.getMessage().startsWith("MQJMSRA_DS4001"))
531 useGlassFishNoCommitWorkaround = true;
532 else
533 log.error(e, "Could not commit JMS Session for channel %s", channel.getId());
534 }
535 }
536 }
537
538 private String normalizeJMSMessageID(String messageId) {
539 if (messageId != null && !messageId.startsWith("ID:"))
540 messageId = "ID:" + messageId;
541 return messageId;
542 }
543
544 public void subscribe(CommandMessage message) throws Exception {
545 String subscriptionId = (String)message.getHeader(AsyncMessage.DESTINATION_CLIENT_ID_HEADER);
546 String selector = (String)message.getHeader(CommandMessage.SELECTOR_HEADER);
547 this.topic = (String)message.getHeader(AsyncMessage.SUBTOPIC_HEADER);
548
549 internalSubscribe(subscriptionId, selector, message.getDestination(), this.topic);
550 }
551
552 public void subscribe(String selector, String destination, String topic) throws Exception {
553 DistributedDataFactory distributedDataFactory = GraniteContext.getCurrentInstance().getGraniteConfig().getDistributedDataFactory();
554 String subscriptionId = distributedDataFactory.getInstance().getDestinationSubscriptionId(destination);
555 if (subscriptionId != null)
556 internalSubscribe(subscriptionId, selector, destination, topic);
557 }
558
559 private void internalSubscribe(String subscriptionId, String selector, String destination, String topic) throws Exception {
560 synchronized (consumers) {
561 JMSConsumer consumer = consumers.get(subscriptionId);
562 if (consumer == null) {
563 consumer = new JMSConsumer(subscriptionId, selector, noLocal);
564 consumer.connect(selector);
565 consumers.put(subscriptionId, consumer);
566 }
567 else
568 consumer.setSelector(selector);
569 channel.addSubscription(destination, topic, subscriptionId, false);
570 }
571 }
572
573 public void unsubscribe(CommandMessage message) throws Exception {
574 String subscriptionId = (String)message.getHeader(AsyncMessage.DESTINATION_CLIENT_ID_HEADER);
575
576 synchronized (consumers) {
577 JMSConsumer consumer = consumers.get(subscriptionId);
578 try {
579 if (consumer != null)
580 consumer.close();
581 }
582 finally {
583 consumers.remove(subscriptionId);
584 channel.removeSubscription(subscriptionId);
585 }
586 }
587 }
588
589
590 private class JMSConsumer implements MessageListener {
591
592 private String subscriptionId = null;
593 private javax.jms.Session jmsConsumerSession = null;
594 private javax.jms.MessageConsumer jmsConsumer = null;
595 private boolean noLocal = false;
596 private String selector = null;
597 private boolean useJBossTCCLDeserializationWorkaround = false;
598 private boolean useGlassFishNoCommitWorkaround = false;
599 private boolean reconnected = false;
600 private Timer reconnectTimer = null;
601
602 public JMSConsumer(String subscriptionId, String selector, boolean noLocal) throws Exception {
603 this.subscriptionId = subscriptionId;
604 this.noLocal = noLocal;
605 this.selector = selector;
606 }
607
608 public void connect(String selector) throws Exception {
609 if (jmsConsumerSession != null)
610 return;
611
612 this.selector = selector;
613
614 // Reconnect to the JMS provider in case no producer has already done it
615 JMSClientImpl.this.connect();
616
617 jmsConsumerSession = jmsConnection.createSession(transactedSessions, acknowledgeMode);
618 if (reconnected)
619 jmsConsumerSession.recover();
620 log.debug("Created JMS Consumer Session for channel %s (transacted: %s, ack: %s)", channel.getId(), transactedSessions, acknowledgeMode);
621
622 if (reconnectTimer != null)
623 reconnectTimer.cancel();
624
625 try {
626 // When failing over, JMS can be in a temporary illegal state. Give it some time to recover.
627 int retryCount = failoverRetryCount;
628 do {
629 try {
630 jmsConsumer = jmsConsumerSession.createConsumer(getConsumerDestination(topic), selector, noLocal);
631 if (retryCount < failoverRetryCount) // We come from a failover, try to recover session
632 reconnected = true;
633 break;
634 }
635 catch (Exception e) {
636 if (retryCount <= 0)
637 throw e;
638
639 if (log.isDebugEnabled())
640 log.debug(e, "Could not create JMS Consumer (retrying %d time)", retryCount);
641 else
642 log.info("Could not create JMS Consumer (retrying %d time)", retryCount);
643
644 try {
645 Thread.sleep(failoverRetryInterval);
646 }
647 catch (Exception f) {
648 throw new ServiceException("Could not sleep when retrying to create JMS Consumer", f.getMessage(), e);
649 }
650 }
651 }
652 while (retryCount-- > 0);
653
654 jmsConsumer.setMessageListener(this);
655 log.debug("Created JMS Consumer for channel %s", channel.getId());
656 }
657 catch (Exception e) {
658 close();
659 throw e;
660 }
661 }
662
663 public void setSelector(String selector) throws Exception {
664 if (jmsConsumer != null) {
665 jmsConsumer.close();
666 jmsConsumer = null;
667 }
668
669 connect(selector);
670 log.debug("Changed selector to %s for JMS Consumer of channel %s", selector, channel.getId());
671 }
672
673 public void reset() {
674 jmsConsumer = null;
675 jmsConsumerSession = null;
676
677 final TimerTask reconnectTask = new TimerTask() {
678 @Override
679 public void run() {
680 try {
681 connect(selector);
682 reconnectTimer.cancel();
683 reconnectTimer = null;
684 }
685 catch (Exception e) {
686 // Wait for next task run
687 }
688 }
689 };
690 if (reconnectTimer != null)
691 reconnectTimer.cancel();
692
693 reconnectTimer = new Timer();
694 reconnectTimer.schedule(reconnectTask, failoverRetryInterval, reconnectRetryInterval);
695 }
696
697 public void close() throws JMSException {
698 if (reconnectTimer != null)
699 reconnectTimer.cancel();
700
701 try {
702 if (jmsConsumer != null) {
703 jmsConsumer.close();
704 jmsConsumer = null;
705 }
706 }
707 finally {
708 if (jmsConsumerSession != null) {
709 jmsConsumerSession.close();
710 jmsConsumerSession = null;
711 }
712 }
713 }
714
715 public void onMessage(javax.jms.Message message) {
716 if (!(message instanceof ObjectMessage) && !(message instanceof TextMessage)) {
717 log.error("JMS Adapter message type not allowed: %s", message.getClass().getName());
718
719 try {
720 if (acknowledgeMode == Session.CLIENT_ACKNOWLEDGE)
721 message.acknowledge();
722
723 if (transactedSessions)
724 jmsConsumerSession.commit();
725 }
726 catch (JMSException e) {
727 log.error(e, "Could not ack/commit JMS onMessage");
728 }
729 }
730
731 log.debug("Delivering JMS message to channel %s subscription %s", channel.getId(), subscriptionId);
732
733 AsyncMessage dmsg = new AsyncMessage();
734 try {
735 Serializable msg = null;
736
737 if (textMessages) {
738 TextMessage jmsMessage = (TextMessage)message;
739 msg = jmsMessage.getText();
740 }
741 else {
742 ObjectMessage jmsMessage = (ObjectMessage)message;
743 if (useJBossTCCLDeserializationWorkaround) {
744 // On JBoss 6, try to deserialize with application class loader if the previous attempt fails
745 ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();
746 try {
747 Thread.currentThread().setContextClassLoader(getClass().getClassLoader());
748 msg = jmsMessage.getObject();
749 }
750 finally {
751 Thread.currentThread().setContextClassLoader(contextClassLoader);
752 }
753 }
754 try {
755 msg = jmsMessage.getObject();
756 }
757 catch (JMSException e) {
758 // On JBoss 6, try to deserialize with application class loader if the previous attempt fails
759 ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();
760 try {
761 Thread.currentThread().setContextClassLoader(getClass().getClassLoader());
762 msg = jmsMessage.getObject();
763 useJBossTCCLDeserializationWorkaround = true;
764 }
765 finally {
766 Thread.currentThread().setContextClassLoader(contextClassLoader);
767 }
768 }
769 }
770
771 dmsg.setDestination(getDestination().getId());
772
773 if (Boolean.TRUE.equals(message.getBooleanProperty(Gravity.BYTEARRAY_BODY_HEADER))) {
774 getGravity().initThread(null, channel.getClientType());
775 try {
776 ByteArrayOutputStream baos = new ByteArrayOutputStream(100);
777 AMF3Serializer ser = new AMF3Serializer(baos);
778 ser.writeObject(msg);
779 ser.close();
780 baos.close();
781 dmsg.setBody(baos.toByteArray());
782 }
783 finally {
784 getGravity().releaseThread();
785 }
786 }
787 else
788 dmsg.setBody(msg);
789
790 dmsg.setMessageId(denormalizeJMSMessageID(message.getJMSMessageID()));
791 dmsg.setCorrelationId(denormalizeJMSMessageID(message.getJMSCorrelationID()));
792 dmsg.setTimestamp(message.getJMSTimestamp());
793 dmsg.setTimeToLive(message.getJMSExpiration());
794
795 Enumeration<?> ename = message.getPropertyNames();
796 while (ename.hasMoreElements()) {
797 String pname = (String)ename.nextElement();
798 dmsg.setHeader(pname, message.getObjectProperty(pname));
799 }
800
801 dmsg.setHeader("JMSType", message.getJMSType());
802 dmsg.setHeader("JMSPriority", Integer.valueOf(message.getJMSPriority()));
803 dmsg.setHeader("JMSRedelivered", Boolean.valueOf(message.getJMSRedelivered()));
804 dmsg.setHeader("JMSDeliveryMode", Integer.valueOf(message.getJMSDeliveryMode()));
805 dmsg.setHeader(AsyncMessage.DESTINATION_CLIENT_ID_HEADER, subscriptionId);
806
807 channel.receive(dmsg);
808 }
809 catch (IOException e) {
810 if (transactedSessions) {
811 try {
812 jmsConsumerSession.rollback();
813 }
814 catch (JMSException f) {
815 log.error("Could not rollback JMS session, messageId: %s", dmsg.getMessageId());
816 }
817 }
818
819 throw new RuntimeException("IO Error", e);
820 }
821 catch (JMSException e) {
822 if (transactedSessions) {
823 try {
824 jmsConsumerSession.rollback();
825 }
826 catch (JMSException f) {
827 log.error("Could not rollback JMS session, messageId: %s", dmsg.getMessageId());
828 }
829 }
830
831 throw new RuntimeException("JMS Error", e);
832 }
833 catch (MessageReceivingException e) {
834 if (transactedSessions) {
835 try {
836 jmsConsumerSession.rollback();
837 }
838 catch (JMSException f) {
839 log.error("Could not rollback JMS session, messageId: %s", dmsg.getMessageId());
840 }
841 }
842
843 throw new RuntimeException("Channel delivery Error", e);
844 }
845
846 try {
847 if (acknowledgeMode == Session.CLIENT_ACKNOWLEDGE)
848 message.acknowledge();
849
850 if (transactedSessions && !useGlassFishNoCommitWorkaround)
851 jmsConsumerSession.commit();
852 }
853 catch (JMSException e) {
854 if (e.getMessage() != null && e.getMessage().startsWith("MQJMSRA_DS4001"))
855 useGlassFishNoCommitWorkaround = true;
856 else
857 log.error(e, "Could not ack/commit JMS onMessage, messageId: %s", dmsg.getMessageId());
858
859 // Message already delivered to client, should rollback or not ?
860 }
861 }
862
863 private String denormalizeJMSMessageID(String messageId) {
864 if (messageId != null && messageId.startsWith("ID:"))
865 messageId = messageId.substring(3);
866 return messageId;
867 }
868 }
869 }
870 }