Java > Open Source Codes > org > jboss > ejb > plugins > jms > JMSContainerInvoker


1 /*
2 * JBoss, Home of Professional Open Source
3 * Copyright 2005, JBoss Inc., and individual contributors as indicated
4 * by the @authors tag. See the copyright.txt in the distribution for a
5 * full listing of individual contributors.
6 *
7 * This is free software; you can redistribute it and/or modify it
8 * under the terms of the GNU Lesser General Public License as
9 * published by the Free Software Foundation; either version 2.1 of
10 * the License, or (at your option) any later version.
11 *
12 * This software is distributed in the hope that it will be useful,
13 * but WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15 * Lesser General Public License for more details.
16 *
17 * You should have received a copy of the GNU Lesser General Public
18 * License along with this software; if not, write to the Free
19 * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
20 * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
21 */
22 package org.jboss.ejb.plugins.jms;
23
24 import java.lang.reflect.Method ;
25 import java.security.AccessController ;
26 import java.security.Principal ;
27 import java.security.PrivilegedAction ;
28 import java.util.Collection ;
29
30 import javax.ejb.EJBMetaData ;
31 import javax.jms.Connection ;
32 import javax.jms.ConnectionConsumer ;
33 import javax.jms.Destination ;
34 import javax.jms.ExceptionListener ;
35 import javax.jms.JMSException ;
36 import javax.jms.Message ;
37 import javax.jms.MessageListener ;
38 import javax.jms.Queue ;
39 import javax.jms.QueueConnection ;
40 import javax.jms.ServerSessionPool ;
41 import javax.jms.Topic ;
42 import javax.jms.TopicConnection ;
43 import javax.management.MBeanServer ;
44 import javax.management.Notification ;
45 import javax.management.ObjectName ;
46 import javax.naming.Context ;
47 import javax.naming.InitialContext ;
48 import javax.naming.NamingException ;
49 import javax.transaction.Transaction ;
50 import javax.transaction.TransactionManager ;
51
52 import org.jboss.deployment.DeploymentException;
53 import org.jboss.ejb.Container;
54 import org.jboss.ejb.EJBProxyFactory;
55 import org.jboss.invocation.Invocation;
56 import org.jboss.invocation.InvocationType;
57 import org.jboss.jms.ConnectionFactoryHelper;
58 import org.jboss.jms.asf.ServerSessionPoolFactory;
59 import org.jboss.jms.asf.StdServerSessionPool;
60 import org.jboss.jms.jndi.JMSProviderAdapter;
61 import org.jboss.logging.Logger;
62 import org.jboss.metadata.ActivationConfigPropertyMetaData;
63 import org.jboss.metadata.InvokerProxyBindingMetaData;
64 import org.jboss.metadata.MessageDestinationMetaData;
65 import org.jboss.metadata.MessageDrivenMetaData;
66 import org.jboss.metadata.MetaData;
67 import org.jboss.system.ServiceMBeanSupport;
68 import org.w3c.dom.Element ;
69
70 /**
71  * EJBProxyFactory for JMS MessageDrivenBeans
72  *
73  * @author <a HREF="mailto:peter.antman@tim.se">Peter Antman</a> .
74  * @author <a HREF="mailto:rickard.oberg@telkel.com">Rickard Oberg</a>
75  * @author <a HREF="mailto:sebastien.alborini@m4x.org">Sebastien Alborini</a>
76  * @author <a HREF="mailto:marc.fleury@telkel.com">Marc Fleury</a>
77  * @author <a HREF="mailto:jason@planet57.com">Jason Dillon</a>
78  * @author <a HREF="mailto:adrian@jboss.com">Adrian Brock</a>
79  * @version <tt>$Revision: 57507 $</tt>
80  */
81 public class JMSContainerInvoker extends ServiceMBeanSupport
82    implements EJBProxyFactory, JMSContainerInvokerMBean
83 {
84    /** The logger */
85    private static final Logger log = Logger.getLogger(JMSContainerInvoker.class);
86    
87    /** Notification sent before connectioning */
88    private static final String CONNECTING_NOTIFICATION = "org.jboss.ejb.plugins.jms.CONNECTING";
89
90    /** Notification sent after connection */
91    private static final String CONNECTED_NOTIFICATION = "org.jboss.ejb.plugins.jms.CONNECTED";
92
93    /** Notification sent before disconnection */
94    private static final String DISCONNECTING_NOTIFICATION = "org.jboss.ejb.plugins.jms.DISCONNECTING";
95
96    /** Notification sent before disconnected */
97    private static final String DISCONNECTED_NOTIFICATION = "org.jboss.ejb.plugins.jms.DISCONNECTED";
98
99    /** Notification sent at connection failure */
100    private static final String FAILURE_NOTIFICATION = "org.jboss.ejb.plugins.jms.FAILURE";
101    
102    /** {@link MessageListener#onMessage} reference. */
103    protected static Method ON_MESSAGE;
104
105    /**
106     * Default destination type. Used when no message-driven-destination is given
107     * in ejb-jar, and a lookup of destinationJNDI from jboss.xml is not
108     * successfull. Default value: javax.jms.Topic.
109     */
110    protected final static String DEFAULT_DESTINATION_TYPE = "javax.jms.Topic";
111    
112    /**
113     * Initialize the ON_MESSAGE reference.
114     */
115    static
116    {
117       try
118       {
119          final Class type = MessageListener .class;
120          final Class arg = Message .class;
121          ON_MESSAGE = type.getMethod("onMessage", new Class []{arg});
122       }
123       catch (Exception e)
124       {
125          throw new ExceptionInInitializerError (e);
126       }
127    }
128    
129    protected boolean optimize;
130
131    /** Maximum number provider is allowed to stuff into a session. */
132    protected int maxMessagesNr = 1;
133
134    /** Minimun pool size of server sessions. */
135    protected int minPoolSize = 1;
136
137    /** Keep alive server sessions. */
138    protected long keepAlive = 30 * 1000;
139
140    /** Maximun pool size of server sessions. */
141    protected int maxPoolSize = 15;
142
143    /** Time to wait before retrying to reconnect a lost connection. */
144    protected long reconnectInterval = 10000;
145
146    /** If Dead letter queue should be used or not. */
147    protected boolean useDLQ = false;
148
149    /**
150     * JNDI name of the provider adapter.
151     *
152     * @see org.jboss.jms.jndi.JMSProviderAdapter
153     */
154    protected String providerAdapterJNDI;
155
156    /**
157     * JNDI name of the server session factory.
158     *
159     * @see org.jboss.jms.asf.ServerSessionPoolFactory
160     */
161    protected String serverSessionPoolFactoryJNDI;
162
163    /** JMS acknowledge mode, used when session is not XA. */
164    protected int acknowledgeMode;
165
166    protected boolean isContainerManagedTx;
167    protected boolean isNotSupportedTx;
168
169    /** The container. */
170    protected Container container;
171
172    /** The JMS connection. */
173    protected Connection connection;
174
175    /** The JMS connection consumer. */
176    protected ConnectionConsumer connectionConsumer;
177
178    protected TransactionManager tm;
179    protected ServerSessionPool pool;
180    protected ExceptionListenerImpl exListener;
181
182    /** Dead letter queue handler. */
183    protected DLQHandler dlqHandler;
184
185    /** DLQConfig element from MDBConfig element from jboss.xml. */
186    protected Element dlqConfig;
187
188    protected InvokerProxyBindingMetaData invokerMetaData;
189    protected String invokerBinding;
190
191    protected boolean deliveryActive = true;
192
193    protected boolean createJBossMQDestination = true;
194
195    /**
196     * Set the invoker meta data so that the ProxyFactory can initialize
197     * properly
198     */
199    public void setInvokerMetaData(InvokerProxyBindingMetaData imd)
200    {
201       invokerMetaData = imd;
202    }
203
204    /**
205     * Set the invoker jndi binding
206     */
207    public void setInvokerBinding(String binding)
208    {
209       invokerBinding = binding;
210    }
211    
212    /**
213     * Set the container for which this is an invoker to.
214     *
215     * @param container The container for which this is an invoker to.
216     */
217    public void setContainer(final Container container)
218    {
219       this.container = container;
220    }
221
222    public int getMinPoolSize()
223    {
224       return minPoolSize;
225    }
226
227    public void setMinPoolSize(int minPoolSize)
228    {
229       this.minPoolSize = minPoolSize;
230    }
231
232    public int getMaxPoolSize()
233    {
234       return maxPoolSize;
235    }
236
237    public void setMaxPoolSize(int maxPoolSize)
238    {
239       this.maxPoolSize = maxPoolSize;
240    }
241
242    public long getKeepAliveMillis()
243    {
244       return keepAlive;
245    }
246
247    public void setKeepAliveMillis(long keepAlive)
248    {
249       this.keepAlive = keepAlive;
250    }
251
252    public int getMaxMessages()
253    {
254       return maxMessagesNr;
255    }
256
257    public void setMaxMessages(int maxMessages)
258    {
259       this.maxMessagesNr = maxMessages;
260    }
261
262    public MessageDrivenMetaData getMetaData()
263    {
264       MessageDrivenMetaData config =
265          (MessageDrivenMetaData) container.getBeanMetaData();
266       return config;
267    }
268
269    public boolean getDeliveryActive()
270    {
271       return deliveryActive;
272    }
273
274    public boolean getCreateJBossMQDestination()
275    {
276       return createJBossMQDestination;
277    }
278
279    public void startDelivery()
280       throws Exception
281    {
282       if (getState() != STARTED)
283          throw new IllegalStateException ("The MDB is not started");
284       if (deliveryActive)
285          return;
286       deliveryActive = true;
287       startService();
288    }
289
290    public void stopDelivery()
291       throws Exception
292    {
293       if (getState() != STARTED)
294          throw new IllegalStateException ("The MDB is not started");
295       if (deliveryActive == false)
296          return;
297       deliveryActive = false;
298       stopService();
299    }
300
301    /**
302     * Sets the Optimized attribute of the JMSContainerInvoker object
303     *
304     * @param optimize The new Optimized value
305     */
306    public void setOptimized(final boolean optimize)
307    {
308       this.optimize = optimize;
309    }
310
311    public boolean isIdentical(Container container, Invocation mi)
312    {
313       throw new Error ("Not valid for MessageDriven beans");
314    }
315
316    public Object getEJBHome()
317    {
318       throw new Error ("Not valid for MessageDriven beans");
319    }
320
321    public EJBMetaData getEJBMetaData()
322    {
323       throw new Error ("Not valid for MessageDriven beans");
324    }
325
326    public Collection getEntityCollection(Collection ids)
327    {
328       throw new Error ("Not valid for MessageDriven beans");
329    }
330
331    public Object getEntityEJBObject(Object id)
332    {
333       throw new Error ("Not valid for MessageDriven beans");
334    }
335
336    public Object getStatefulSessionEJBObject(Object id)
337    {
338       throw new Error ("Not valid for MessageDriven beans");
339    }
340
341    public Object getStatelessSessionEJBObject()
342    {
343       throw new Error ("Not valid for MessageDriven beans");
344    }
345
346    public boolean isOptimized()
347    {
348       return optimize;
349    }
350
351    /**
352     * XmlLoadable implementation.
353     *
354     * @todo FIXME - we ought to move all config into MDBConfig, but I do not do that
355     * now due to backward compatibility.
356     *
357     * @param element Description of Parameter
358     * @throws DeploymentException Description of Exception
359     */
360    public void importXml(final Element element) throws Exception
361    {
362       try
363       {
364          if ("false".equalsIgnoreCase(MetaData.getElementContent(MetaData.getUniqueChild(element, "CreateJBossMQDestination"))))
365          {
366             createJBossMQDestination = false;
367          }
368       }
369       catch (Exception ignore)
370       {
371       }
372       
373       try
374       {
375          String maxMessages = MetaData.getElementContent
376             (MetaData.getUniqueChild(element, "MaxMessages"));
377          maxMessagesNr = Integer.parseInt(maxMessages);
378       }
379       catch (Exception ignore)
380       {
381       }
382
383       try
384       {
385          String minSize = MetaData.getElementContent
386             (MetaData.getUniqueChild(element, "MinimumSize"));
387          minPoolSize = Integer.parseInt(minSize);
388       }
389       catch (Exception ignore)
390       {
391       }
392
393       try
394       {
395          String maxSize = MetaData.getElementContent
396             (MetaData.getUniqueChild(element, "MaximumSize"));
397          maxPoolSize = Integer.parseInt(maxSize);
398       }
399       catch (Exception ignore)
400       {
401       }
402
403       try
404       {
405          String keepAliveMillis = MetaData.getElementContent
406             (MetaData.getUniqueChild(element, "KeepAliveMillis"));
407          keepAlive = Integer.parseInt(keepAliveMillis);
408       }
409       catch (Exception ignore)
410       {
411       }
412
413       Element mdbConfig = MetaData.getUniqueChild(element, "MDBConfig");
414
415       try
416       {
417          String reconnect = MetaData.getElementContent
418             (MetaData.getUniqueChild(mdbConfig, "ReconnectIntervalSec"));
419          reconnectInterval = Long.parseLong(reconnect) * 1000;
420       }
421       catch (Exception ignore)
422       {
423       }
424
425       try
426       {
427          if ("false".equalsIgnoreCase(MetaData.getElementContent(MetaData.getUniqueChild(mdbConfig, "DeliveryActive"))))
428          {
429             deliveryActive = false;
430          }
431       }
432       catch (Exception ignore)
433       {
434       }
435          
436       // Get Dead letter queue config - and save it for later use
437 Element dlqEl = MetaData.getOptionalChild(mdbConfig, "DLQConfig");
438       if (dlqEl != null)
439       {
440          dlqConfig = (Element ) dlqEl.cloneNode(true);
441          useDLQ = true;
442       }
443       else
444       {
445          useDLQ = false;
446       }
447
448       // If these are not found we will get a DeploymentException, I hope
449 providerAdapterJNDI = MetaData.getElementContent
450          (MetaData.getUniqueChild(element, "JMSProviderAdapterJNDI"));
451
452       serverSessionPoolFactoryJNDI = MetaData.getElementContent
453          (MetaData.getUniqueChild(element, "ServerSessionPoolFactoryJNDI"));
454
455       // Check java:/ prefix
456 if (!providerAdapterJNDI.startsWith("java:/"))
457       {
458          providerAdapterJNDI = "java:/" + providerAdapterJNDI;
459       }
460
461       if (!serverSessionPoolFactoryJNDI.startsWith("java:/"))
462       {
463          serverSessionPoolFactoryJNDI = "java:/" + serverSessionPoolFactoryJNDI;
464       }
465    }
466
467    /**
468     * Initialize the container invoker. Sets up a connection, a server session
469     * pool and a connection consumer for the configured destination.
470     *
471     * <p>Any JMSExceptions produced while initializing will be assumed to be caused
472     * due to JMS Provider failure.
473     *
474     * @throws Exception Failed to initalize.
475     */
476    protected void createService() throws Exception
477    {
478       importXml(invokerMetaData.getProxyFactoryConfig());
479
480       exListener = new ExceptionListenerImpl(this);
481    }
482
483    /**
484     * Initialize the container invoker. Sets up a connection, a server session
485     * pool and a connection consumer for the configured destination.
486     *
487     * @throws Exception Failed to initalize.
488     */
489    protected void innerStartDelivery() throws Exception
490    {
491       if (deliveryActive == false)
492       {
493          log.debug("Delivery is disabled");
494          return;
495       }
496       
497       sendNotification(CONNECTING_NOTIFICATION, null);
498
499       log.debug("Initializing");
500
501       // Get the JMS provider
502 JMSProviderAdapter adapter = getJMSProviderAdapter();
503       log.debug("Provider adapter: " + adapter);
504       
505       // Set up Dead Letter Queue handler
506 if (useDLQ)
507       {
508          dlqHandler = new DLQHandler(adapter, this);
509          dlqHandler.importXml(dlqConfig);
510          dlqHandler.create();
511       }
512       
513       // Store TM reference locally - should we test for CMT Required
514 tm = container.getTransactionManager();
515       
516       // Get configuration information - from EJB-xml
517 MessageDrivenMetaData config = getMetaData();
518       
519       // Selector
520 String messageSelector = config.getMessageSelector();
521       String activationConfig = getActivationConfigProperty("messageSelector");
522       if (activationConfig != null)
523          messageSelector = activationConfig;
524       
525       // Queue or Topic - optional unfortunately
526 String destinationType = config.getDestinationType();
527       activationConfig = getActivationConfigProperty("destinationType");
528       if (activationConfig != null)
529          destinationType = activationConfig;
530       
531       // Is container managed?
532 isContainerManagedTx = config.isContainerManagedTx();
533       acknowledgeMode = config.getAcknowledgeMode();
534       activationConfig = getActivationConfigProperty("acknowledgeMode");
535       if (activationConfig != null)
536       {
537          if (activationConfig.equals("DUPS_OK_ACKNOWLEDGE"))
538             acknowledgeMode = MessageDrivenMetaData.DUPS_OK_ACKNOWLEDGE_MODE;
539          else
540             acknowledgeMode = MessageDrivenMetaData.AUTO_ACKNOWLEDGE_MODE;
541       }
542       
543       byte txType = config.getMethodTransactionType("onMessage",
544          new Class []{Message .class},
545          InvocationType.LOCAL);
546       isNotSupportedTx = txType == MetaData.TX_NOT_SUPPORTED;
547       
548       // Get configuration data from jboss.xml
549 String destinationJNDI = config.getDestinationJndiName();
550       activationConfig = getActivationConfigProperty("destination");
551       if (activationConfig != null)
552          destinationJNDI = activationConfig;
553       // Try any EJB21 destination link
554 if (destinationJNDI == null)
555       {
556          String link = config.getDestinationLink();
557          if (link != null)
558          {
559             link = link.trim();
560             if (link.length() > 0)
561             {
562                MessageDestinationMetaData destinationMetaData = container.getMessageDestination(link);
563                if (destinationMetaData == null)
564                   log.warn("Unresolved message-destination-link '" + link + "' no message-destination in ejb-jar.xml");
565                else
566                {
567                   String jndiName = destinationMetaData.getJNDIName();
568                   if (jndiName == null)
569                      log.warn("The message-destination '" + link + "' has no jndi-name in jboss.xml");
570                   else
571                      destinationJNDI = jndiName;
572                }
573             }
574          }
575       }
576       
577       String user = config.getUser();
578       String password = config.getPasswd();
579       
580       // Connect to the JNDI server and get a reference to root context
581 Context context = adapter.getInitialContext();
582       log.debug("context: " + context);
583       
584       // if we can't get the root context then exit with an exception
585 if (context == null)
586       {
587          throw new RuntimeException ("Failed to get the root context");
588       }
589       
590       // Get the JNDI suffix of the destination
591 String jndiSuffix = parseJndiSuffix(destinationJNDI, config.getEjbName());
592       log.debug("jndiSuffix: " + jndiSuffix);
593       
594       // Unfortunately the destination is optional, so if we do not have one
595 // here we have to look it up if we have a destinationJNDI, else give it
596 // a default.
597 if (destinationType == null)
598       {
599          log.warn("No message-driven-destination given; using; guessing type");
600          destinationType = getDestinationType(context, destinationJNDI);
601       }
602
603       if ("javax.jms.Topic".equals(destinationType))
604       {
605          log.debug("Got destination type Topic for " + config.getEjbName());
606          
607          // create a topic connection
608 Object factory = context.lookup(adapter.getTopicFactoryRef());
609          TopicConnection tConnection = null;
610          try
611          {
612             tConnection = ConnectionFactoryHelper.createTopicConnection(factory, user, password);
613             connection = tConnection;
614          }
615          catch (ClassCastException e)
616          {
617             throw new DeploymentException("Expected a TopicConnection check your provider adaptor: "
618                + adapter.getTopicFactoryRef());
619          }
620
621          try
622          {
623             // Fix: ClientId must be set as the first method call after connection creation.
624 // Fix: ClientId is necessary for durable subscriptions.
625
626             String clientId = config.getClientId();
627             activationConfig = getActivationConfigProperty("clientID");
628             if (activationConfig != null)
629                clientId = activationConfig;
630
631             log.debug("Using client id: " + clientId);
632             if (clientId != null && clientId.length() > 0)
633                connection.setClientID(clientId);
634
635             // lookup or create the destination topic
636 Topic topic = null;
637             try
638             {
639                // First we try the specified topic
640 if (destinationJNDI != null)
641                   topic = (Topic ) context.lookup(destinationJNDI);
642                else if (createJBossMQDestination == false)
643                   throw new DeploymentException("Unable to determine destination for '" + container.getBeanMetaData().getEjbName()
644                         + "' use destination-jndi-name in jboss.xml, an activation config property or a message-destination-link");
645             }
646             catch (NamingException e)
647             {
648                if (createJBossMQDestination == false)
649                   throw new DeploymentException("Could not find the topic destination-jndi-name=" + destinationJNDI, e);
650                log.warn("Could not find the topic destination-jndi-name=" + destinationJNDI, e);
651             }
652             catch (ClassCastException e)
653             {
654                throw new DeploymentException("Expected a Topic destination-jndi-name=" + destinationJNDI, e);
655             }
656
657             // FIXME: This is not portable, only works for JBossMQ
658 if (topic == null)
659                topic = (Topic ) createDestination(Topic .class,
660                   context,
661                   "topic/" + jndiSuffix,
662                   jndiSuffix);
663             
664             // set up the server session pool
665 pool = createSessionPool(
666                topic,
667                tConnection,
668                minPoolSize,
669                maxPoolSize,
670                keepAlive,
671                true, // tx
672 acknowledgeMode,
673                new MessageListenerImpl(this));
674
675             int subscriptionDurablity = config.getSubscriptionDurability();
676             activationConfig = getActivationConfigProperty("subscriptionDurability");
677             if (activationConfig != null)
678             {
679                if (activationConfig.equals("Durable"))
680                   subscriptionDurablity = MessageDrivenMetaData.DURABLE_SUBSCRIPTION;
681                else
682                   subscriptionDurablity = MessageDrivenMetaData.NON_DURABLE_SUBSCRIPTION;
683             }
684             // To be no-durable or durable
685 if (subscriptionDurablity != MessageDrivenMetaData.DURABLE_SUBSCRIPTION)
686             {
687                // Create non durable
688 connectionConsumer =
689                   tConnection.createConnectionConsumer(topic,
690                      messageSelector,
691                      pool,
692                      maxMessagesNr);
693             }
694             else
695             {
696                // Durable subscription
697 String durableName = config.getSubscriptionId();
698                activationConfig = getActivationConfigProperty("subscriptionName");
699                if (activationConfig != null)
700                   durableName = activationConfig;
701
702                connectionConsumer =
703                   tConnection.createDurableConnectionConsumer(topic,
704                      durableName,
705                      messageSelector,
706                      pool,
707                      maxMessagesNr);
708             }
709             log.debug("Topic connectionConsumer set up");
710          }
711          catch (Throwable t)
712          {
713             try
714             {
715                tConnection.close();
716             }
717             catch (Throwable ignored)
718             {
719             }
720             DeploymentException.rethrowAsDeploymentException("Error during topic setup", t);
721          }
722       }
723       else if ("javax.jms.Queue".equals(destinationType))
724       {
725          log.debug("Got destination type Queue for " + config.getEjbName());
726          
727          // create a queue connection
728 Object qFactory = context.lookup(adapter.getQueueFactoryRef());
729          QueueConnection qConnection = null;
730          try
731          {
732             qConnection = ConnectionFactoryHelper.createQueueConnection(qFactory, user, password);
733             connection = qConnection;
734          }
735          catch (ClassCastException e)
736          {
737             throw new DeploymentException("Expected a QueueConnection check your provider adaptor: "
738                + adapter.getQueueFactoryRef());
739          }
740          
741          try
742          {
743             // Set the optional client id
744 String clientId = config.getClientId();
745             activationConfig = getActivationConfigProperty("clientID");
746             if (activationConfig != null)
747                clientId = activationConfig;
748             
749             log.debug("Using client id: " + clientId);
750             if (clientId != null && clientId.length() > 0)
751                connection.setClientID(clientId);
752             
753             // lookup or create the destination queue
754 Queue queue = null;
755             try
756             {
757                // First we try the specified queue
758 if (destinationJNDI != null)
759                   queue = (Queue ) context.lookup(destinationJNDI);
760                else if (createJBossMQDestination == false)
761                   throw new DeploymentException("Unable to determine destination for '" + container.getBeanMetaData().getEjbName()
762                         + "' use destination-jndi-name in jboss.xml, an activation config property or a message-destination-link");
763             }
764             catch (NamingException e)
765             {
766                if (createJBossMQDestination == false)
767                   throw new DeploymentException("Could not find the queue destination-jndi-name=" + destinationJNDI, e);
768                log.warn("Could not find the queue destination-jndi-name=" + destinationJNDI);
769             }
770             catch (ClassCastException e)
771             {
772                throw new DeploymentException("Expected a Queue destination-jndi-name=" + destinationJNDI);
773             }
774
775             // FIXME: This is not portable, only works for JBossMQ
776 if (queue == null)
777                queue = (Queue ) createDestination(Queue .class,
778                   context,
779                   "queue/" + jndiSuffix,
780                   jndiSuffix);
781             
782             // set up the server session pool
783 pool = createSessionPool(
784                queue,
785                qConnection,
786                minPoolSize,
787                maxPoolSize,
788                keepAlive,
789                true, // tx
790 acknowledgeMode,
791                new MessageListenerImpl(this));
792             log.debug("Server session pool: " + pool);
793             
794             // create the connection consumer
795 connectionConsumer =
796                qConnection.createConnectionConsumer(queue,
797                   messageSelector,
798                   pool,
799                   maxMessagesNr);
800             log.debug("Connection consumer: " + connectionConsumer);
801          }
802          catch (Throwable t)
803          {
804             try
805             {
806                qConnection.close();
807             }
808             catch (Throwable ignored)
809             {
810             }
811             DeploymentException.rethrowAsDeploymentException("Error during queue setup", t);
812          }
813       }
814       else
815          throw new DeploymentException("Unknown destination-type " + destinationType);
816
817       log.debug("Initialized with config " + toString());
818
819       context.close();
820       
821       if (dlqHandler != null)
822       {
823          dlqHandler.start();
824       }
825
826       if (connection != null)
827       {
828          connection.setExceptionListener(exListener);
829          connection.start();
830       }
831
832       sendNotification(CONNECTED_NOTIFICATION, null);
833    }
834
835    protected void startService() throws Exception
836    {
837       try
838       {
839          innerStartDelivery();
840       }
841       catch (final Throwable t)
842       {
843          // start a thread up to handle recovering the connection. so we can
844 // attach to the jms resources once they become available
845 exListener.handleFailure(t);
846          return;
847       }
848       finally
849       {
850          // Clear any security context established by the jms connection
851 SecurityActions.clear();
852       }
853    }
854
855    protected void stopService() throws Exception
856    {
857       // Silence the exception listener
858 if (exListener != null)
859       {
860          exListener.stop();
861       }
862
863       innerStopDelivery();
864    }
865
866    /**
867     * Stop done from inside, we should not stop the exceptionListener in inner
868     * stop.
869     */
870    protected void innerStopDelivery()
871    {
872       log.debug("innerStop");
873
874       sendNotification(DISCONNECTING_NOTIFICATION, null);
875       
876       try
877       {
878          if (connection != null)
879          {
880             connection.setExceptionListener(null);
881             log.debug("unset exception listener");
882          }
883       }
884       catch (Throwable t)
885       {
886          log.trace("Could not set ExceptionListener to null", t);
887       }
888       
889       // Stop the connection
890 try
891       {
892          if (connection != null)
893          {
894             connection.stop();
895             log.debug("connection stopped");
896          }
897       }
898       catch (Throwable t)
899       {
900          log.trace("Could not stop JMS connection", t);
901       }
902
903       try
904       {
905          if (dlqHandler != null)
906             dlqHandler.stop();
907       }
908       catch (Throwable t)
909       {
910          log.trace("Failed to stop the dlq handler", t);
911       }
912
913       // close the connection consumer
914 try
915       {
916          if (connectionConsumer != null)
917             connectionConsumer.close();
918       }
919       catch (Throwable t)
920       {
921          log.trace("Failed to close connection consumer", t);
922       }
923       connectionConsumer = null;
924       
925       // clear the server session pool (if it is clearable)
926 try
927       {
928          if (pool instanceof StdServerSessionPool)
929          {
930             StdServerSessionPool p = (StdServerSessionPool) pool;
931             p.clear();
932          }
933       }
934       catch (Throwable t)
935       {
936          log.trace("Failed to clear session pool", t);
937       }
938       
939       // close the connection
940 if (connection != null)
941       {
942          try
943          {
944             connection.close();
945          }
946          catch (Throwable t)
947          {
948             log.trace("Failed to close connection", t);
949          }
950       }
951       connection = null;
952
953       // Take down DLQ
954 try
955       {
956          if (dlqHandler != null)
957          {
958             dlqHandler.destroy();
959          }
960       }
961       catch (Throwable t)
962       {
963          log.trace("Failed to close the dlq handler", t);
964       }
965       dlqHandler = null;
966       
967       sendNotification(DISCONNECTED_NOTIFICATION, null);
968    }
969    
970    public Object invoke(Object id,
971       Method m,
972       Object [] args,
973       Transaction tx,
974       Principal identity,
975       Object credential)
976       throws Exception
977    {
978
979       Invocation invocation = new Invocation(id, m, args, tx, identity, credential);
980       invocation.setType(InvocationType.LOCAL);
981
982       // Set the right context classloader
983 ClassLoader oldCL = TCLAction.UTIL.getContextClassLoader();
984       TCLAction.UTIL.setContextClassLoader(container.getClassLoader());
985       try
986       {
987          return container.invoke(invocation);
988       }
989       finally
990       {
991          TCLAction.UTIL.setContextClassLoader(oldCL);
992       }
993    }
994
995    /**
996     * Try to get a destination type by looking up the destination JNDI, or
997     * provide a default if there is not destinationJNDI or if it is not possible
998     * to lookup.
999     *
1000    * @param ctx The naming context to lookup destinations from.
1001    * @param destinationJNDI The name to use when looking up destinations.
1002    * @return The destination type, either derived from destinationJDNI or
1003    * DEFAULT_DESTINATION_TYPE
1004    */
1005   protected String getDestinationType(Context ctx, String destinationJNDI)
1006   {
1007      String destType = null;
1008
1009      if (destinationJNDI != null)
1010      {
1011         try
1012         {
1013            Destination dest = (Destination ) ctx.lookup(destinationJNDI);
1014            if (dest instanceof javax.jms.Topic )
1015            {
1016               destType = "javax.jms.Topic";
1017            }
1018            else if (dest instanceof javax.jms.Queue )
1019            {
1020               destType = "javax.jms.Queue";
1021            }
1022         }
1023         catch (NamingException ex)
1024         {
1025            log.debug("Could not do heristic lookup of destination ", ex);
1026         }
1027
1028      }
1029      if (destType == null)
1030      {
1031         log.warn("Could not determine destination type, defaults to: " +
1032            DEFAULT_DESTINATION_TYPE);
1033
1034         destType = DEFAULT_DESTINATION_TYPE;
1035      }
1036
1037      return destType;
1038   }
1039
1040   /**
1041    * Return the JMSProviderAdapter that should be used.
1042    *
1043    * @return The JMSProviderAdapter to use.
1044    */
1045   protected JMSProviderAdapter getJMSProviderAdapter() throws NamingException
1046   {
1047      Context context = new InitialContext ();
1048      try
1049      {
1050         log.debug("Looking up provider adapter: " + providerAdapterJNDI);
1051         return (JMSProviderAdapter) context.lookup(providerAdapterJNDI);
1052      }
1053      finally
1054      {
1055         context.close();
1056      }
1057   }
1058
1059   /**
1060    * Create and or lookup a JMS destination.
1061    *
1062    * @param type Either javax.jms.Queue or javax.jms.Topic.
1063    * @param ctx The naming context to lookup destinations from.
1064    * @param jndiName The name to use when looking up destinations.
1065    * @param jndiSuffix The name to use when creating destinations.
1066    * @return The destination.
1067    * @throws IllegalArgumentException Type is not Queue or Topic.
1068    * @throws Exception Description of Exception
1069    */
1070   protected Destination createDestination(final Class type,
1071      final Context ctx,
1072      final String jndiName,
1073      final String jndiSuffix)
1074      throws Exception
1075   {
1076      try
1077      {
1078         // first try to look it up
1079 return (Destination ) ctx.lookup(jndiName);
1080      }
1081      catch (NamingException e)
1082      {
1083         // if the lookup failes, the try to create it
1084 log.warn("destination not found: " + jndiName + " reason: " + e);
1085         log.warn("creating a new temporary destination: " + jndiName);
1086
1087         //
1088 // jason: we should do away with this...
1089 //
1090 // attempt to create the destination (note, this is very
1091 // very, very unportable).
1092 //
1093
1094         MBeanServer server = org.jboss.mx.util.MBeanServerLocator.locateJBoss();
1095
1096         String methodName;
1097         if (type == Topic .class)
1098         {
1099            methodName = "createTopic";
1100         }
1101         else if (type == Queue .class)
1102         {
1103            methodName = "createQueue";
1104         }
1105         else
1106         {
1107            // type was not a Topic or Queue, bad user
1108 throw new IllegalArgumentException
1109               ("Expected javax.jms.Queue or javax.jms.Topic: " + type);
1110         }
1111
1112         // invoke the server to create the destination
1113 server.invoke(new ObjectName ("jboss.mq:service=DestinationManager"),
1114            methodName,
1115            new Object []{jndiSuffix},
1116            new String []{"java.lang.String"});
1117         
1118         // try to look it up again
1119 return (Destination ) ctx.lookup(jndiName);
1120      }
1121   }
1122
1123   protected String getActivationConfigProperty(String property)
1124   {
1125      MessageDrivenMetaData mdmd = getMetaData();
1126      ActivationConfigPropertyMetaData acpmd = mdmd.getActivationConfigProperty(property);
1127      if (acpmd != null)
1128         return acpmd.getValue();
1129      else
1130         return null;
1131   }
1132   
1133   /**
1134    * Create a server session pool for the given connection.
1135    *
1136    * @param destination the destination
1137    * @param connection The connection to use.
1138    * @param minSession The minumum number of sessions
1139    * @param maxSession The maximum number of sessions.
1140    * @param keepAlive The time to keep sessions alive
1141    * @param isTransacted True if the sessions are transacted.
1142    * @param ack The session acknowledgement mode.
1143    * @param listener The message listener.
1144    * @return A server session pool.
1145    * @throws JMSException
1146    * @throws NamingException Description of Exception
1147    */
1148   protected ServerSessionPool createSessionPool(
1149      final Destination destination,
1150      final Connection connection,
1151      final int minSession,
1152      final int maxSession,
1153      final long keepAlive,
1154      final boolean isTransacted,
1155      final int ack,
1156      final MessageListener listener)
1157      throws NamingException , JMSException
1158   {
1159      ServerSessionPool pool;
1160      Context context = new InitialContext ();
1161
1162      try
1163      {
1164         // first lookup the factory
1165 log.debug("looking up session pool factory: " +
1166            serverSessionPoolFactoryJNDI);
1167         ServerSessionPoolFactory factory = (ServerSessionPoolFactory)
1168            context.lookup(serverSessionPoolFactoryJNDI);
1169         
1170         // the create the pool
1171 pool = factory.getServerSessionPool(destination, connection, minSession, maxSession, keepAlive, isTransacted, ack, !isContainerManagedTx || isNotSupportedTx, listener);
1172      }
1173      finally
1174      {
1175         context.close();
1176      }
1177
1178      return pool;
1179   }
1180
1181   /**
1182    * Notify of an event
1183    *
1184    * @param event the event
1185    * @param userData any user data, e.g. the exception on a failure
1186    */
1187   protected void sendNotification(String event, Object userData)
1188   {
1189      Notification notif = new Notification (event, getServiceName(), getNextNotificationSequenceNumber());
1190      notif.setUserData(userData);
1191      sendNotification(notif);
1192   }
1193   
1194   /**
1195    * Parse the JNDI suffix from the given JNDI name.
1196    *
1197    * @param jndiname The JNDI name used to lookup the destination.
1198    * @param defautSuffix Description of Parameter
1199    * @return The parsed suffix or the defaultSuffix
1200    */
1201   protected String parseJndiSuffix(final String jndiname,
1202      final String defautSuffix)
1203   {
1204      // jndiSuffix is merely the name that the user has given the MDB.
1205 // since the jndi name contains the message type I have to split
1206 // at the "/" if there is no slash then I use the entire jndi name...
1207 String jndiSuffix = "";
1208
1209      if (jndiname != null)
1210      {
1211         int indexOfSlash = jndiname.indexOf("/");
1212         if (indexOfSlash != -1)
1213         {
1214            jndiSuffix = jndiname.substring(indexOfSlash + 1);
1215         }
1216         else
1217         {
1218            jndiSuffix = jndiname;
1219         }
1220      }
1221      else
1222      {
1223         // if the jndi name from jboss.xml is null then lets use the ejbName
1224 jndiSuffix = defautSuffix;
1225      }
1226
1227      return jndiSuffix;
1228   }
1229
1230   /**
1231    * An implementation of MessageListener that passes messages on to the
1232    * container invoker.
1233    */
1234   class MessageListenerImpl implements MessageListener
1235   {
1236      /** The container invoker. */
1237      JMSContainerInvoker invoker;
1238      
1239      /**
1240       * Construct a <tt>MessageListenerImpl</tt> .
1241       *
1242       * @param invoker The container invoker. Must not be null.
1243       */
1244      MessageListenerImpl(final JMSContainerInvoker invoker)
1245      {
1246         this.invoker = invoker;
1247      }
1248
1249      /**
1250       * Process a message.
1251       *
1252       * @param message The message to process.
1253       */
1254      public void onMessage(final Message message)
1255      {
1256         if (log.isTraceEnabled())
1257         {
1258            log.trace("processing message: " + message);
1259         }
1260
1261         Object id;
1262         try
1263         {
1264            id = message.getJMSMessageID();
1265         }
1266         catch (JMSException e)
1267         {
1268            // what ?
1269 id = "JMSContainerInvoker";
1270         }
1271
1272         // Invoke, shuld we catch any Exceptions??
1273 try
1274         {
1275            Transaction tx = tm.getTransaction();
1276
1277            // DLQHandling
1278 if (useDLQ && // Is Dead Letter Queue used at all
1279 message.getJMSRedelivered() && // Was message resent
1280 dlqHandler.handleRedeliveredMessage(message, tx)) //Did the DLQ handler take care of the message
1281 {
1282               // Message will be placed on Dead Letter Queue,
1283 // if redelivered to many times
1284 return;
1285            }
1286
1287            invoker.invoke(id, // Object id - where used?
1288 ON_MESSAGE, // Method to invoke
1289 new Object []{message}, // argument
1290 tx, // Transaction
1291 null, // Principal
1292 null); // Cred
1293
1294         }
1295         catch (Exception e)
1296         {
1297            log.error("Exception in JMSCI message listener", e);
1298         }
1299      }
1300   }
1301
1302   /** ExceptionListener for failover handling. */
1303   class ExceptionListenerImpl implements ExceptionListener
1304   {
1305      Object lock = new Object ();
1306      JMSContainerInvoker invoker;
1307      Thread currentThread;
1308      boolean notStopped = true;
1309
1310      /**
1311       * Create a new ExceptionListenerImpl.
1312       *
1313       * @param invoker the container invoker
1314       */
1315      ExceptionListenerImpl(final JMSContainerInvoker invoker)
1316      {
1317         this.invoker = invoker;
1318      }
1319
1320      /**
1321       * Called on jms connection failure events
1322       *
1323       * @param ex the jms connection failure exception
1324       */
1325      public void onException(JMSException ex)
1326      {
1327         handleFailure(ex);
1328      }
1329
1330      /**
1331       * Handle a failure
1332       *
1333       * @param t the failure
1334       */
1335      public void handleFailure(Throwable t)
1336      {
1337         MessageDrivenMetaData metaData = invoker.getMetaData();
1338         log.warn("JMS provider failure detected for " + metaData.getEjbName(), t);
1339         
1340         // JBAS-3750 - Help debug integration with foreign JMS providers
1341 if (t instanceof JMSException )
1342         {
1343            Exception le = ((JMSException )t).getLinkedException();
1344            if (le != null)
1345               log.debug("Linked exception: " + le + ", cause: " + le.getCause());
1346         }
1347         
1348         // Run the reconnection in the background
1349 String name = "JMSContainerInvoker("+metaData.getEjbName()+") Reconnect";
1350         synchronized (lock)
1351         {
1352            if (currentThread != null)
1353            {
1354               log.debug("Already a reconnect thread: " + currentThread + " for " + metaData.getEjbName());
1355               return;
1356            }
1357            Runnable runnable = new ExceptionListenerRunnable(t);
1358            currentThread = new Thread (runnable, name);
1359            try
1360            {
1361               currentThread.setDaemon(true);
1362               currentThread.start();
1363            }
1364            catch (RuntimeException rethrow)
1365            {
1366               currentThread = null;
1367               throw rethrow;
1368            }
1369            catch (Error rethrow)
1370            {
1371               currentThread = null;
1372               throw rethrow;
1373            }
1374         }
1375      }
1376
1377      class ExceptionListenerRunnable implements Runnable
1378      {
1379         Throwable failure;
1380
1381         /**
1382          * Create a new ExceptionListenerRunnable.
1383          *
1384          * @param failure the error
1385          */
1386         public ExceptionListenerRunnable(Throwable failure)
1387         {
1388            this.failure = failure;
1389         }
1390         
1391         /**
1392          * Try to reconnect to the jms provider until explicitly stopped.
1393          */
1394         public void run()
1395         {
1396            MessageDrivenMetaData metaData = invoker.getMetaData();
1397            try
1398            {
1399               boolean tryIt = true;
1400               while (tryIt && notStopped)
1401               {
1402                  try
1403                  {
1404                     invoker.innerStopDelivery();
1405                  }
1406                  catch (Throwable t)
1407                  {
1408                     log.error("Unhandled error stopping connection for " + metaData.getEjbName(), t);
1409                  }
1410
1411                  sendNotification(FAILURE_NOTIFICATION, failure);
1412                  
1413                  try
1414                  {
1415                     log.info("Waiting for reconnect internal " + reconnectInterval + "ms for " + metaData.getEjbName());
1416                     try
1417                     {
1418                        Thread.sleep(reconnectInterval);
1419                     }
1420                     catch (InterruptedException ie)
1421                     {
1422                        tryIt = false;
1423                        return;
1424                     }
1425                     
1426                     // Reboot container
1427 log.info("Trying to reconnect to JMS provider for " + metaData.getEjbName());
1428                     invoker.innerStartDelivery();
1429                     tryIt = false;
1430
1431                     log.info("Reconnected to JMS provider for " + metaData.getEjbName());
1432                  }
1433                  catch (Throwable t)
1434                  {
1435                     log.error("Reconnect failed: JMS provider failure detected for " + metaData.getEjbName(), t);
1436                  }
1437               }
1438            }
1439            finally
1440            {
1441               synchronized (lock)
1442               {
1443                  currentThread = null;
1444               }
1445            }
1446         }
1447      }
1448
1449      void stop()
1450      {
1451         synchronized (lock)
1452         {
1453            log.debug("Stop requested for recovery thread: " + currentThread);
1454            notStopped = false;
1455            if (currentThread != null)
1456            {
1457               currentThread.interrupt();
1458               log.debug("Recovery thread interrupted: " + currentThread);
1459            }
1460         }
1461      }
1462   }
1463
1464   /**
1465    * Return a string representation of the current config state.
1466    */
1467   public String toString()
1468   {
1469      MessageDrivenMetaData metaData = getMetaData();
1470      String destinationJNDI = metaData.getDestinationJndiName();
1471      return super.toString() +
1472         "{ maxMessagesNr=" + maxMessagesNr +
1473         ", maxPoolSize=" + maxPoolSize +
1474         ", reconnectInterval=" + reconnectInterval +
1475         ", providerAdapterJNDI=" + providerAdapterJNDI +
1476         ", serverSessionPoolFactoryJNDI=" + serverSessionPoolFactoryJNDI +
1477         ", acknowledgeMode=" + acknowledgeMode +
1478         ", isContainerManagedTx=" + isContainerManagedTx +
1479         ", isNotSupportedTx=" + isNotSupportedTx +
1480         ", useDLQ=" + useDLQ +
1481         ", dlqHandler=" + dlqHandler +
1482         ", destinationJNDI=" + destinationJNDI +
1483         " }";
1484   }
1485
1486   interface TCLAction
1487   {
1488      class UTIL
1489      {
1490         static TCLAction getTCLAction()
1491         {
1492            return System.getSecurityManager() == null ? NON_PRIVILEGED : PRIVILEGED;
1493         }
1494
1495         static ClassLoader getContextClassLoader()
1496         {
1497            return getTCLAction().getContextClassLoader();
1498         }
1499
1500         static ClassLoader getContextClassLoader(Thread thread)
1501         {
1502            return getTCLAction().getContextClassLoader(thread);
1503         }
1504
1505         static void setContextClassLoader(ClassLoader cl)
1506         {
1507            getTCLAction().setContextClassLoader(cl);
1508         }
1509
1510         static void setContextClassLoader(Thread thread, ClassLoader cl)
1511         {
1512            getTCLAction().setContextClassLoader(thread, cl);
1513         }
1514      }
1515
1516      TCLAction NON_PRIVILEGED = new TCLAction()
1517      {
1518         public ClassLoader getContextClassLoader()
1519         {
1520            return Thread.currentThread().getContextClassLoader();
1521         }
1522
1523         public ClassLoader getContextClassLoader(Thread thread)
1524         {
1525            return thread.getContextClassLoader();
1526         }
1527
1528         public void setContextClassLoader(ClassLoader cl)
1529         {
1530            Thread.currentThread().setContextClassLoader(cl);
1531         }
1532
1533         public void setContextClassLoader(Thread thread, ClassLoader cl)
1534         {
1535            thread.setContextClassLoader(cl);
1536         }
1537      };
1538
1539      TCLAction PRIVILEGED = new TCLAction()
1540      {
1541         private final PrivilegedAction getTCLPrivilegedAction = new PrivilegedAction ()
1542         {
1543            public Object run()
1544            {
1545               return Thread.currentThread().getContextClassLoader();
1546            }
1547         };
1548
1549         public ClassLoader getContextClassLoader()
1550         {
1551            return (ClassLoader ) AccessController.doPrivileged(getTCLPrivilegedAction);
1552         }
1553
1554         public ClassLoader getContextClassLoader(final Thread thread)
1555         {
1556            return (ClassLoader ) AccessController.doPrivileged(new PrivilegedAction ()
1557            {
1558               public Object run()
1559               {
1560                  return thread.getContextClassLoader();
1561               }
1562            });
1563         }
1564
1565         public void setContextClassLoader(final ClassLoader cl)
1566         {
1567            AccessController.doPrivileged(new PrivilegedAction ()
1568            {
1569               public Object run()
1570               {
1571                  Thread.currentThread().setContextClassLoader(cl);
1572                  return null;
1573               }
1574            });
1575         }
1576
1577         public void setContextClassLoader(final Thread thread, final ClassLoader cl)
1578         {
1579            AccessController.doPrivileged(new PrivilegedAction ()
1580            {
1581               public Object run()
1582               {
1583                  thread.setContextClassLoader(cl);
1584                  return null;
1585               }
1586            });
1587         }
1588      };
1589
1590      ClassLoader getContextClassLoader();
1591
1592      ClassLoader getContextClassLoader(Thread thread);
1593
1594      void setContextClassLoader(ClassLoader cl);
1595
1596      void setContextClassLoader(Thread thread, ClassLoader cl);
1597   }
1598}
1599

A to Z: JavaDoc & Examples

---

Daily Java News & Articles

---

Open Source Projects

---

Open Source Codes

---

Free Computer Books

---

Remove Frame

---

Popular Tags