Java > Open Source Codes > org > jahia > services > cache > JMSHub


1 /*
2  * ____.
3  * __/\ ______| |__/\. _______
4  * __ .____| | \ | +----+ \
5  * _______| /--| | | - \ _ | : - \_________
6  * \\______: :---| : : | : | \________>
7  * |__\---\_____________:______: :____|____:_____\
8  * /_____|
9  *
10  * . . . i n j a h i a w e t r u s t . . .
11  *
12  *
13  *
14  * ----- BEGIN LICENSE BLOCK -----
15  * Version: JCSL 1.0
16  *
17  * The contents of this file are subject to the Jahia Community Source License
18  * 1.0 or later (the "License"); you may not use this file except in
19  * compliance with the License. You may obtain a copy of the License at
20  * http://www.jahia.org/license
21  *
22  * Software distributed under the License is distributed on an "AS IS" basis,
23  * WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
24  * for the rights, obligations and limitations governing use of the contents
25  * of the file. The Original and Upgraded Code is the Jahia CMS and Portal
26  * Server. The developer of the Original and Upgraded Code is JAHIA Ltd. JAHIA
27  * Ltd. owns the copyrights in the portions it created. All Rights Reserved.
28  *
29  * The Developer of the Shared Modifications is Jahia Solution Sarl.
30  * Portions created by the Initial Developer are Copyright (C) 2002 by the
31  * Initial Developer. All Rights Reserved.
32  *
33  * Contributor(s):
34  * 04-SEP-2003, Jahia Solutions Sarl, Fulco Houkes : Initial version
35  *
36  * ----- END LICENSE BLOCK -----
37  */
38
39
40 package org.jahia.services.cache;
41
42 import java.util.Hashtable ;
43 import javax.jms.ExceptionListener ;
44 import javax.jms.JMSException ;
45 import javax.jms.MapMessage ;
46 import javax.jms.Session ;
47 import javax.jms.Topic ;
48 import javax.jms.TopicConnection ;
49 import javax.jms.TopicConnectionFactory ;
50 import javax.jms.TopicSession ;
51 import javax.naming.Context ;
52 import javax.naming.InitialContext ;
53 import javax.naming.NamingException ;
54
55 import org.jahia.exceptions.JahiaInitializationException;
56 import org.jahia.settings.SettingsBean;
57
58
59 /** This class handles all the JMS Connections, the Message Publisher and Consumer, as
60  * well as reconnection to the JMS server, in case the connection failed.
61  *
62  * @author Fulco Houkes, Copyright (c) 2003 by Jahia Ltd.
63  * @version 1.0
64  * @since Jahia 4.0
65  *
66  * @see org.jahia.services.cache.JMSHubConsumer JMSHubConsumer
67  * @see org.jahia.services.cache.JMSHubPublisher JMSHubPublisher
68  * @see org.jahia.services.cache.JMSHubLookup JMSHubLookup
69  */
70 public class JMSHub implements ExceptionListener {
71
72     final protected static String PACKED_MSG_KEY = "packedmsgKey";
73     final protected static String PACKED_FLAG = "isPacked";
74
75     /** logging. */
76     final private static org.apache.log4j.Logger logger =
77             org.apache.log4j.Logger.getLogger (JMSHub.class);
78
79     private CacheFactory cacheFactory;
80     private TopicConnection topicConnection;
81     private TopicSession topicSession;
82     private Topic topic;
83     private JMSHubPublisher publisher;
84     private JMSHubConsumer consumer;
85     private SettingsBean settings;
86
87     /** true when there is an acitve JMS connection. */
88     private boolean isConnected;
89
90     /** <code>true</code> when the JMS connection failed and the reconnection
91      * thread is active.
92      */
93     private boolean isReconnecting;
94
95     /** <code>true</code> when the class has been initialized through the
96      * <code>init()</code> method.
97      */
98     private boolean isInitialized = false;
99
100     /** Hold the JMS reconnection class when any connection to the JMS server dropped,
101      * otherwise this attribute is <code>null</code>. */
102     private JMSHubLookup jmsServerLookup;
103
104     /** Holds the reference to the JMS server lookup thread. This attribute is
105      * <code>null</code> when no reconnection is in progress.
106      */
107     private Thread jmsServerLookupThread;
108
109     /** Default constructor, creates a new <code>JMSHub</code> instance.*/
110     protected JMSHub () {
111         publisher = new JMSHubPublisher();
112         logger.debug ("JMSHub singleton successfully instanciated!");
113     }
114
115     /** This method is called by the JMS client when any global exception occured.
116      *
117      * @param ex the intercepted exception
118      */
119     public void onException (JMSException ex) {
120         logger.warn (ex);
121
122         // there is a lot of chance the connection to the JMS Server has been lost.
123 jmsConnectionFailure ();
124     }
125
126     /** Try to disconnect from the JMS Server and initiate a new connection.
127      *
128      * @throws JahiaInitializationException
129      * when any initialization failure occured
130      */
131     public void reconnect ()
132             throws JahiaInitializationException {
133         if (!isInitialized) {
134             logger.debug ("JMSHub not initialized! Skip reconnection process");
135             return;
136         }
137
138         if (isReconnecting) {
139             logger.debug ("There is already a reconnection pending!");
140             return;
141         }
142
143         disconnect ();
144         connect ();
145     }
146
147     /** Try to connect to the JMS server using the settings previously set through
148      * the {@link org.jahia.services.cache.JMSHub#init init} method.
149      *
150      * @throws JahiaInitializationException
151      * when any initialization error occured
152      */
153     public synchronized void connect ()
154             throws JahiaInitializationException {
155         if (!isInitialized) {
156             logger.debug ("JMS not initialized! Skip connection process.");
157             return;
158         }
159
160         // no need to reconnect if already there is a pending connection retry.
161 if (isReconnecting)
162             return;
163
164         try {
165             internalConnect ();
166
167         } catch (JMSConnectionException e) {
168             logger.info ("Could not get a connection to the JMS Server on host [" +
169                     settings.lookupString (SettingsBean.JMS_CONTEXT_PROVIDER_URL) +
170                     "]. Try to connect later");
171
172             // no connection could be made to the JMS Server, initiate the
173 // lookup thread
174 startJMSServerLookupThread ();
175         }
176     }
177
178
179     /** Try to connect to the JMS server using the settings previously set through
180      * the {@link org.jahia.services.cache.JMSHub#init init} method.
181      *
182      * @throws JahiaInitializationException
183      * when any initialization error occured
184      */
185     protected void internalConnect ()
186             throws JahiaInitializationException, JMSConnectionException {
187         logger.info ("Connecting to the JMS Server...");
188         if (settings == null) {
189             throw new JahiaInitializationException (
190                     "Need the Jahia settings to connect to the JMS server.");
191         }
192
193         // connect to the JNDI server and get a reference to root context
194 Context context = getContext (settings);
195         logger.debug("Context is " + context.toString());
196
197         // get the Connection Factory and the topic
198 TopicConnectionFactory factory = getConnectionFactory (context);
199         topic = getTopic (context);
200
201         // get the connection and the session
202 topicConnection = getTopicConnection (factory);
203         topicSession = getSession ();
204
205         // finally set the cache as an exception listener, in order to receive the
206 // JMS exceptions.
207 try {
208             topicConnection.setExceptionListener (this);
209
210             logger.debug("Session is " + topicSession.toString() + ", client ID=" + topicConnection.getClientID());
211         } catch (JMSException ex) {
212             logger.warn ("Could not set the JMSHub as a ExceptionListener. Exceptions will not be catched!", ex);
213         }
214
215         // Connect the Message Consumer first, before the Message Publisher
216 connectConsumer ();
217         publisher.connect (this);
218
219         isConnected = true;
220         isReconnecting = false;
221         logger.info ("Successfully connected to the JMS Server.");
222     }
223
224     /** Disconnect from the JMS Server and frees all the attached resources.
225      */
226     public synchronized void disconnect () {
227         if (!isInitialized) {
228             logger.debug ("JMSHub not initialized! Skip disconnection process.");
229             return;
230         }
231
232         logger.info ("Disconnecting from the JMS Server...");
233
234         if (isReconnecting) {
235             stopJMSServerLookupThread ();
236         }
237
238         try {
239             isConnected = false;
240
241             // shut down the Message Publisher first and then the Message Consumer
242 publisher.disconnect ();
243             disconnectConsumer ();
244
245             // close the session
246 if (topicSession != null)
247                 topicSession.close ();
248
249             // close the connection
250 if (topicConnection != null) {
251                 topicConnection.close ();
252             }
253
254             logger.info ("Successfully disconnected from JMS Server");
255
256         } catch (JMSException ex) {
257             logger.warn ("Could not close the connection with the JMS Server.", ex);
258
259         } finally {
260             // garbage the resources. This is needed to ensure freeing the resources
261 // in case an exception occured.
262 topicSession = null;
263             topicConnection = null;
264             topic = null;
265         }
266     }
267
268
269     /** Send a <i>flush</i> message.
270      *
271      * @param cache the cache to which the message has to be send
272      *
273      * @return <code>true</code> on success, otherwise <code>false</code>.
274      */
275     public boolean sendFlushMessage (Cache cache) {
276         return sendMessage (cache, JMSCacheMessage.FLUSH_EVENT, null, null);
277     }
278
279
280     /** Send a <i>remove</i> message.
281      *
282      * @param cache the cache to which the message has to be send
283      * @param entryKey the cache entry key
284      *
285      * @return <code>true</code> on success, otherwise <code>false</code>.
286      */
287     public boolean sendRemoveMessage (Cache cache, Object entryKey) {
288         return sendMessage (cache, JMSCacheMessage.REMOVE_EVENT, entryKey, null);
289     }
290
291
292     /** Send a <i>put</i> message.
293      *
294      * @param cache the cache to which the message has to be send
295      * @param entryKey the cache entry key
296      * @param entryValue the cache entry value
297      *
298      * @return <code>true</code> on success, otherwise <code>false</code>.
299      */
300     public boolean sendPutMessage (Cache cache, Object entryKey, Object entryValue) {
301         return sendMessage (cache, JMSCacheMessage.PUT_EVENT, entryKey, entryValue);
302     }
303
304
305     /** Retrieves a displayable string of the specified <code>message</code>.
306      * Used for debbuging purpose only.
307      *
308      * @param message the message to change into a string
309      *
310      * @return the message string equivalent
311      */
312     public static String messageToString (MapMessage message) {
313
314         try {
315             String cacheName = message.getString (JMSCacheMessage.CACHE_NAME_KEY);
316             String eventName = message.getString (JMSCacheMessage.EVENT_NAME_KEY);
317             Object entryKey = message.getObject (JMSCacheMessage.ENTRY_KEY);
318             String clientID = message.getString (JMSCacheMessage.CLIENT_KEY);
319
320             StringBuffer buffer = new StringBuffer ("message (cache:");
321             buffer.append (cacheName);
322             buffer.append (", event:");
323             buffer.append (eventName);
324             buffer.append (", entryKey:");
325             buffer.append ((entryKey == null ? "null" : entryKey.toString ()));
326             buffer.append (", clientID:");
327             buffer.append (clientID);
328             buffer.append (")");
329             return buffer.toString ();
330
331         } catch (JMSException e) {
332             return "";
333         }
334     }
335
336
337     /** Retrieves the current JMS Topic Session.
338      *
339      * @return the JMS Topic Session
340      */
341     public TopicSession getTopicSession () {
342         return topicSession;
343     }
344
345
346     /** Retrieves the JMS Topic used for the cache synchronization.
347      *
348      * @return the JMS topic in use
349      */
350     public Topic getTopic () {
351         return topic;
352     }
353
354
355     /** Retrieves the JMS Topic Connection in use.
356      *
357      * @return the JMS Topic Connection in use.
358      */
359     public TopicConnection getTopicConnection () {
360         return topicConnection;
361     }
362
363
364     /** <p>This method has to be called when a JMS disconnection has occured.</p>
365      * <p>Tries to shut down the existing connections and resources, and initiate
366      * the reconnection thread to get a new connection to the JMS Server.</p>
367      */
368     protected void jmsConnectionFailure () {
369         disconnect ();
370         startJMSServerLookupThread ();
371     }
372
373     /** Initialize the cache JMS settings, but does not try to make any connection
374      * to the JMS Server. This has to be done through the
375      * {@link org.jahia.services.cache.JMSHub#connect connect()} method.
376      *
377      * @param settings the initialization settings
378      * @see org.jahia.services.cache.JMSHub#connect
379      * @see org.jahia.services.cache.JMSHub#disconnect
380      *
381      * @throws JahiaInitializationException
382      * when any initialization failure occured
383      */
384     public void init (SettingsBean settings, CacheFactory cacheFactory)
385             throws JahiaInitializationException {
386         this.cacheFactory = cacheFactory;
387         // check the settings are not null, otherwise throw an exception
388 if (settings == null) {
389             logger.warn ("Cannot use null settings to initialize the cache!!");
390             throw new JahiaInitializationException ("null settings!");
391         }
392         this.settings = settings;
393
394         // by default the synchronization is not enabled. It has explicitely to be
395 // enabled by the CacheFactory.
396 isConnected = false;
397         isReconnecting = false;
398         isInitialized = true;
399
400         // log the instanciation
401 logger.debug ("JMS Hub successfully initialized!");
402     }
403
404
405     /** Create a new Message Consumer instance and connect it to active
406      * JMS Server connection.
407      *
408      * @throws JahiaInitializationException
409      * when any initialization failure occured
410      */
411     private void connectConsumer ()
412             throws JahiaInitializationException {
413         if (!isInitialized) {
414             logger.debug ("JMS not initialized! Skip Message Consumer connection.");
415             return;
416         }
417
418         // The Message Consumer is not a thread, as it is already associated to the
419 // JMS client thread. Therefore, we can safely create a new instance of the Consumer.
420 consumer = new JMSHubConsumer ();
421         consumer.init (this);
422     }
423
424
425     /** Disconnect the Message Consumer and garbage its instance.
426      */
427     private void disconnectConsumer () {
428         // kill the consumer
429 consumer = null;
430     }
431
432
433     /** Send the <code>event</code> associated to the specified <code>cache</code>
434      * and <code>entryKey</code>.
435      *
436      * @param cache the cache to which the message is destinated
437      * @param messageType the message type
438      * @param entryKey the cache entry key related to this message
439      * @param entryValue the cache entry value for this message
440      *
441      * @return <code>true</code> on success, otherwise <code>false</code>.
442      */
443     private boolean sendMessage (Cache cache, int messageType, Object entryKey, Object entryValue) {
444         if (!isInitialized) {
445             logger.debug ("JMS not initialized! Skip sending message.");
446             return false;
447         }
448
449         if (cache == null) {
450             logger.debug ("cannot handle null cache");
451             return false;
452         }
453
454         if ((messageType != JMSCacheMessage.FLUSH_EVENT) && (entryKey == null)) {
455             logger.debug ("cannot handle null entryKey");
456             return false;
457         }
458
459         if (isConnected) {
460             if (publisher != null) {
461
462                 JMSCacheMessage message = null;
463                 switch (messageType) {
464                     // PUT message
465 case JMSCacheMessage.PUT_EVENT:
466                         message = JMSCacheMessage.createPutMessage (cache, entryKey, entryValue);
467                         break;
468
469                         // REMOVE message
470 case JMSCacheMessage.REMOVE_EVENT:
471                         message = JMSCacheMessage.createRemoveMessage (cache, entryKey);
472                         break;
473
474                         // FLUSH message
475 case JMSCacheMessage.FLUSH_EVENT:
476                         message = JMSCacheMessage.createFlushMessage (cache);
477                         break;
478                 }
479                 publisher.publishMessage (message);
480                 return true;
481
482             } else {
483                 logger.error ("JMS Publisher thread is null even if synchronization is enabled. This is really bad news!!!");
484                 return false;
485             }
486
487         } else {
488             //logger.debug ("ignoring message, as synchronization has not yet been enabled!!");
489 }
490
491         // synchronization is not enabled, so not publishing possible anyway.
492 return true;
493     }
494
495     public void sendMessagesNow() {
496         publisher.sendNow();
497     }
498
499
500     /** Initiates a new JMS Topic Session.
501      *
502      * @return a new JMS Topic Session
503      *
504      * @throws JahiaInitializationException
505      * when a initialization error occured
506      */
507     private TopicSession getSession ()
508             throws JahiaInitializationException {
509         try {
510             // get the topic session
511 TopicSession topicSession = topicConnection.createTopicSession (
512                     false, Session.AUTO_ACKNOWLEDGE);
513
514             if (topicSession == null) {
515                 throw new JahiaInitializationException (
516                         "Could not get the TopicSession instance!");
517             }
518
519             return topicSession;
520
521         } catch (JMSException ex) {
522             logger.fatal (ex);
523             throw new JahiaInitializationException (
524                     "JMS Exception while getting the Topic Session", ex);
525         }
526     }
527
528
529     /** Initiates a new JMS Topic Connection.
530      *
531      * @param factory the JMS Topic Connection Factory
532      * @return a new instance of the JMS Topic Connection
533      *
534      * @throws JahiaInitializationException
535      * when a initialization error occured
536      */
537     private TopicConnection getTopicConnection (TopicConnectionFactory factory)
538             throws JahiaInitializationException {
539         try {
540             // get the topic connection with the default user (default permission)
541 TopicConnection topicConnection = factory.createTopicConnection ();
542             if (topicConnection == null) {
543                 throw new JahiaInitializationException (
544                         "Could not get the TopicConnection instance!");
545             }
546             topicConnection.start ();
547             return topicConnection;
548
549         } catch (JMSException ex) {
550             logger.fatal (ex);
551             throw new JahiaInitializationException (
552                     "JMS Exception while initializing JMS Hub", ex);
553         }
554     }
555
556
557     /** Lookup the Topic from the JNDI Context.
558      *
559      * @param context the JNDI context to use
560      * @return the Topic to use for the cache synchronization
561      *
562      * @throws JMSConnectionException
563      * when a JMS connection failure occured.
564      */
565     private Topic getTopic (Context context)
566             throws JMSConnectionException
567     {
568         String topicName = settings.lookupString (SettingsBean.JMS_TOPIC_NAME);
569         try {
570             // Get the topic
571 return (Topic )context.lookup (topicName);
572
573         } catch (NamingException ex) {
574
575             logger.warn ("Could not lookup the specified [" + topicName +
576                     "] topic. Check if the has been previously configured in the JMS Server!", ex);
577
578             throw new JMSConnectionException (
579                     "Could not get the topic " + topicName + "] instance.");
580         }
581     }
582
583
584     /** Initiate a new JMS Connection Factory.
585      *
586      * @param context the JNDI Context to use
587      *
588      * @return the new JMS Connection Factory
589      *
590      * @throws JMSConnectionException
591      * when a JMS connection failure occured.
592      */
593     private TopicConnectionFactory getConnectionFactory (Context context)
594             throws JMSConnectionException
595     {
596         String factoryName =
597                 settings.lookupString (SettingsBean.JMS_TOPIC_CONNECTION_FACTORY_NAME);
598         try {
599             // get the factory topic connection
600 return (TopicConnectionFactory )context.lookup (factoryName);
601
602         } catch (NamingException ex) {
603             logger.warn ("Could not lookup the connection factor name [" + factoryName +
604                     "]. Check if the connection factory has been correctly configured in the jahia.properties file.", ex);
605
606             throw new JMSConnectionException (
607                     "Could not get the factory [" + factoryName + "] instance.");
608         }
609     }
610
611
612     /** Lookups up the JNDI Initial Context according to the <i>Initial Context Factory</i>
613      * and <i>URL Provider</i> defined in the Jahia configuration file.
614      *
615      * @param settings the Jahia configuration settings
616      *
617      * @return the JNDI Initial Context
618      *
619      * @throws JMSConnectionException
620      * when a JMS connection failure occured.
621      */
622     private Context getContext (SettingsBean settings)
623             throws JMSConnectionException
624     {
625         String initalContextName = settings.lookupString(
626                 SettingsBean.JMS_INITIAL_CONTEXT);
627         String providerUrl = settings.lookupString (SettingsBean.JMS_CONTEXT_PROVIDER_URL);
628
629         try {
630             Hashtable properties = new Hashtable ();
631
632             properties.put (Context.INITIAL_CONTEXT_FACTORY,initalContextName);
633
634             properties.put (Context.PROVIDER_URL,providerUrl);
635
636             logger.debug ("Try to connect to initial context (factory: " +
637                     initalContextName + ", provider: " + providerUrl + ")");
638
639             Context context = new InitialContext (properties);
640             logger.debug ("Could successfully retrieve the initial context");
641             return context;
642
643         } catch (NamingException ex) {
644
645 // StringBuffer buffer = new StringBuffer ();
646 // buffer.append ("\n\nVous savez, moi je ne crois pas qu'il y ait de bonne ou de mauvaise\n");
647 // buffer.append ("situation. Moi, si je dois resumer ma vie aujourd'hui avec vous, je\n");
648 // buffer.append ("dirais que c'est d'abord des exceptions levees. Des threads qui m'ont tendu la\n");
649 // buffer.append ("main, peut-etre a un moment ou je ne pouvais pas, ou j'etais bloque\n");
650 // buffer.append ("chez moi. Et c'est curieux de se dire que les catch, les synchronized\n");
651 // buffer.append ("forgent une destinee... Parce que quand on a le gout de l'uml,\n");
652 // buffer.append ("quand on a le gout des graphes uml bien fait, le beau geste, parfois\n");
653 // buffer.append ("on ne trouve pas l'interface en face je dirais, le GUI qui\n");
654 // buffer.append ("vous aide a avancer. Alors ce n'est pas mon cas, comme je disais\n");
655 // buffer.append ("la, puisque moi au contraire, j'ai pu: et je dis merci a la informatique, je\n");
656 // buffer.append ("lui dis merci, je chante la vie, je danse la vie... Je ne suis qu'amour!\n");
657 // buffer.append ("Et finalement, quand beaucoup de gens me disent \"Mais comment\n");
658 // buffer.append ("fais-tu pour avoir cette humanite?\", je leur reponds tres\n");
659 // buffer.append ("simplement, je leur dis que c'est ce gout de Tux qui m'a pousse\n");
660 // buffer.append ("aujourd'hui a entreprendre une construction logicielle, mais demain\n");
661 // buffer.append ("qui sait? Peut-etre seulement a me mettre au service de la\n");
662 // buffer.append ("communaute open source, a faire le don, le don de soi... ou\n");
663 // buffer.append ("simplement à arreter d'ajouter des conneries dans mon code, et\n");
664 // buffer.append ("d'avancer dans mon codage! :o)\n\n");
665 // logger.fatal (buffer.toString(), ex);
666
667             throw new JMSConnectionException ("Could not get the InitialContext [" +
668                     initalContextName + "] on provider [" + providerUrl + "].", ex);
669         }
670
671     }
672
673
674     /**
675      * Instanciates and starts the JMS Reconnection Thread.
676      */
677     private void startJMSServerLookupThread () {
678         if (!isReconnecting) {
679             isReconnecting = true;
680             jmsServerLookup = new JMSHubLookup (this,
681                     settings.lookupLong(SettingsBean.JMS_SERVER_LOOKUP_SLEEP_TIME));
682             jmsServerLookupThread = new Thread (jmsServerLookup);
683             jmsServerLookupThread.setName ("JMS Reconnection");
684             jmsServerLookupThread.start ();
685         }
686     }
687
688
689     /**
690      * Stops the JMS Reconnection Thread.
691      */
692     private synchronized void stopJMSServerLookupThread () {
693
694         if ((jmsServerLookup != null) && (jmsServerLookupThread != null)) {
695
696             logger.info ("Stopping JMS Server lookup thread");
697             jmsServerLookup.stop ();
698             synchronized (jmsServerLookupThread) {
699                 jmsServerLookupThread.notify();
700             }
701
702             try {
703                 jmsServerLookupThread.join ();
704             } catch (InterruptedException ex) {
705                 // it's what we were waiting for!
706 }
707
708             jmsServerLookupThread = null;
709             jmsServerLookup = null;
710         }
711         isReconnecting = false;
712     }
713
714     public CacheFactory getCacheFactory() {
715         return cacheFactory;
716     }
717
718     public SettingsBean getSettings () {
719         return settings;
720     }
721
722 }
723

A to Z: JavaDoc & Examples

---

Daily Java News & Articles

---

Open Source Projects

---

Open Source Codes

---

Free Computer Books

---

Remove Frame

---

Popular Tags