Coverage Report

Coverage Report - org.apache.commons.messagelet.SubscriptionManager

Classes in this File

Line Coverage

Branch Coverage

Complexity

SubscriptionManager

0/92

0/46

2.923

 /*
  * Copyright (C) The Apache Software Foundation. All rights reserved.
  *
  * This software is published under the terms of the Apache Software License
  * version 1.1, a copy of which has been included with this distribution in
  * the LICENSE file.
  * 
  * $Id: SubscriptionManager.java 155459 2005-02-26 13:24:44Z dirkv $
  */
 package org.apache.commons.messagelet;
 
 import java.util.Iterator;
 
 import javax.jms.Destination;
 import javax.jms.JMSException;
 import javax.jms.MessageListener;
 import javax.servlet.ServletContext;
 import javax.servlet.ServletException;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.commons.messagelet.model.Subscription;
 import org.apache.commons.messagelet.model.SubscriptionList;
 import org.apache.commons.messenger.Messenger;
 import org.apache.commons.messenger.MessengerManager;
 import org.apache.commons.messenger.tool.StopWatchMessageListener;
 
 /** 
  * <p><code>SubscriptionManager</code> is a simple command line program that will
  * create a number of subscriptions and consume messages using just regular 
  * MDO and MessageListener classes.
  *
  * @author <a href="mailto:jstrachan@apache.org">James Strachan</a>
  * @version $Revision: 155459 $
  */
 public class SubscriptionManager {
 
     /** Logger */
     private static final Log log = LogFactory.getLog(SubscriptionManager.class);
 
     /** The JMS connections */    
     private MessengerManager manager;
 
     /** The JMS Subscriptions */
     private SubscriptionList subscriptionList;
 
     /** The context passed into MDOs */
     private ServletContext servletContext;
         
     /** Should we use a stopwatch to output performance metrics */
     private boolean useStopWatch = false;
 
     
     public SubscriptionManager() {
     }
 
 
     protected void subscribe() throws JMSException, ServletException {
         for (Iterator iter = getSubscriptionList().getSubscriptions().iterator(); iter.hasNext(); ) {
             Subscription subscription = (Subscription) iter.next();
             subscribe( subscription );
         }
     }
     
     public void subscribe( Subscription subscription ) throws JMSException, ServletException{
         String name = subscription.getConnection();
         Messenger messenger = getMessenger( name );
         if ( messenger == null ) {
             throw new JMSException( "No such Messenger called: " + name + " for subscription: " + subscription );
         }
         String subject = subscription.getSubject();
         if ( subject == null || subject.length() == 0 ) {
             throw new JMSException( "No destination defined for subscription: " + subscription );
         }
         
         Destination destination = messenger.getDestination( subject );
         if ( destination == null ) {
             throw new JMSException( "No destination could be found for name: " + subject + " for subscription: " + subscription );
         }
 
         MessageListener listener = subscription.getMessageListener();
         if ( listener == null ) {
             throw new JMSException( "No MessageListener is defined for subscription: " + subscription );
         }
         
         // if its an MDO the initialise it!
         if ( listener instanceof MessageDrivenObject ) {
             MessageDrivenObject mdo = (MessageDrivenObject) listener;
             if ( mdo instanceof MessengerMDO ) {
                 MessengerMDO messengerMDO = (MessengerMDO) mdo;
                 messengerMDO.setMessenger( messenger );
                 messengerMDO.setMessengerManager( getMessengerManager() );
             }
             mdo.init( getServletContext() );
         }
 
         listener = wrapInStopWatch( listener );
 
         String selector = subscription.getSelector();
 
         ConsumerThread thread = subscription.getConsumerThread();
         if (thread != null) {
             log.info( "Subscribing to messenger: " + name + " destination: " + subject + " selector: " + selector + " with: " + thread );
             
             thread.setMessenger(messenger);
             thread.setDestination(destination);
             thread.setSelector(selector);
             thread.setListener(listener);
             thread.start();
         }
         else {
             if ( selector != null && selector.length() > 0 ) {
                 log.info( "Subscribing to messenger: " + name + " destination: " + subject + " selector: " + selector );
                 
                 messenger.addListener( destination, selector, listener );
             }
             else {
                 log.info( "Subscribing to messenger: " + name + " destination: " + subject );
                 
                 messenger.addListener( destination, listener );
             }
             
             log.info( "Subscribed with listener: " + listener );
         }
     }
 
 
     
     public void unsubscribe() throws JMSException, ServletException {
         SubscriptionList list = getSubscriptionList();
         if ( list != null ) {
             for (Iterator iter = list.getSubscriptions().iterator(); iter.hasNext(); ) {
                 Subscription subscription = (Subscription) iter.next();
                 unsubscribe( subscription );
             }
         }
     }
     
     public void unsubscribe( Subscription subscription ) throws JMSException, ServletException {
         // lets unsubscribe first
         String name = subscription.getConnection();
         Messenger messenger = getMessenger( name );
         
         MessageListener listener = subscription.getMessageListener();
         if ( messenger != null && listener != null ) {
             Destination destination = null;        
             String subject = subscription.getSubject();
             if ( subject == null || subject.length() == 0 ) {
                 log.error( "No destination defined for subscription: " + subscription );
             }
             else {
                 try {
                     destination = messenger.getDestination( subject );
                     if ( destination == null ) {
                         log.error( "No destination could be found for name: " + subject + " for subscription: " + subscription );
                     }
                 }
                 catch (JMSException e) {
                     log.error( "Could not create destination for name: " + subject + " for subscription: " + subscription, e );
                 }
             }
             if ( destination != null ) {
                 try {
                     String selector = subscription.getSelector();
                     if ( selector != null && selector.length() > 0 ) {
                         messenger.removeListener( destination, selector, listener );
                     }
                     else {
                         messenger.removeListener( destination, listener );
                     }
                 }
                 catch (JMSException e) {
                     log.error( "Could not unsubscribe to destination:" + destination + " for subscription: " + subscription, e );
                 }
             }
         }
         
         // now lets destrory the MBO
         if ( listener instanceof MessageDrivenObject ) {
             MessageDrivenObject mdo = (MessageDrivenObject) listener;
             mdo.destroy();
         }
     }
     
     
     // Properties
     //-------------------------------------------------------------------------    
 
     public MessengerManager getMessengerManager() throws JMSException {
         return manager;
     }
     
     public void setMessengerManager(MessengerManager manager) {
         this.manager = manager;
     }
 
     /**
      * Returns the subscriptionList.
      * @return SubscriptionList
      */
     public SubscriptionList getSubscriptionList() {
         return subscriptionList;
     }
 
     /**
      * Sets the subscriptionList.
      * @param subscriptionList The subscriptionList to set
      */
     public void setSubscriptionList(SubscriptionList subscriptionList) {
         this.subscriptionList = subscriptionList;
     }
     
     /**
      * Returns the servletContext.
      * @return ServletContext
      */
     public ServletContext getServletContext() {
         return servletContext;
     }
 
     /**
      * Sets the servletContext.
      * @param servletContext The servletContext to set
      */
     public void setServletContext(ServletContext servletContext) {
         this.servletContext = servletContext;
     }
     
     // Implementation methods
     //-------------------------------------------------------------------------    
     /**
      * Allows the MessageListener to be wrapped inside a stop watch message listener if required 
      */
     protected MessageListener wrapInStopWatch( MessageListener listener ) {
         if ( useStopWatch ) {
             return new StopWatchMessageListener( listener );
         }
         return listener;
     }
 
     protected Messenger getMessenger(String name) throws JMSException {
         return getMessengerManager().getMessenger( name );
     }
 
 }

1		/*
2		* Copyright (C) The Apache Software Foundation. All rights reserved.
3		*
4		* This software is published under the terms of the Apache Software License
5		* version 1.1, a copy of which has been included with this distribution in
6		* the LICENSE file.
7		*
8		* $Id: SubscriptionManager.java 155459 2005-02-26 13:24:44Z dirkv $
9		*/
10		package org.apache.commons.messagelet;
11
12		import java.util.Iterator;
13
14		import javax.jms.Destination;
15		import javax.jms.JMSException;
16		import javax.jms.MessageListener;
17		import javax.servlet.ServletContext;
18		import javax.servlet.ServletException;
19
20		import org.apache.commons.logging.Log;
21		import org.apache.commons.logging.LogFactory;
22		import org.apache.commons.messagelet.model.Subscription;
23		import org.apache.commons.messagelet.model.SubscriptionList;
24		import org.apache.commons.messenger.Messenger;
25		import org.apache.commons.messenger.MessengerManager;
26		import org.apache.commons.messenger.tool.StopWatchMessageListener;
27
28		/**
29		* <p><code>SubscriptionManager</code> is a simple command line program that will
30		* create a number of subscriptions and consume messages using just regular
31		* MDO and MessageListener classes.
32		*
33		* @author <a href="mailto:jstrachan@apache.org">James Strachan</a>
34		* @version $Revision: 155459 $
35		*/
36		public class SubscriptionManager {
37
38		/** Logger */
39	0	private static final Log log = LogFactory.getLog(SubscriptionManager.class);
40
41		/** The JMS connections */
42		private MessengerManager manager;
43
44		/** The JMS Subscriptions */
45		private SubscriptionList subscriptionList;
46
47		/** The context passed into MDOs */
48		private ServletContext servletContext;
49
50		/** Should we use a stopwatch to output performance metrics */
51	0	private boolean useStopWatch = false;
52
53
54	0	public SubscriptionManager() {
55	0	}
56
57
58		protected void subscribe() throws JMSException, ServletException {
59	0	for (Iterator iter = getSubscriptionList().getSubscriptions().iterator(); iter.hasNext(); ) {
60	0	Subscription subscription = (Subscription) iter.next();
61	0	subscribe( subscription );
62	0	}
63	0	}
64
65		public void subscribe( Subscription subscription ) throws JMSException, ServletException{
66	0	String name = subscription.getConnection();
67	0	Messenger messenger = getMessenger( name );
68	0	if ( messenger == null ) {
69	0	throw new JMSException( "No such Messenger called: " + name + " for subscription: " + subscription );
70		}
71	0	String subject = subscription.getSubject();
72	0	if ( subject == null \|\| subject.length() == 0 ) {
73	0	throw new JMSException( "No destination defined for subscription: " + subscription );
74		}
75
76	0	Destination destination = messenger.getDestination( subject );
77	0	if ( destination == null ) {
78	0	throw new JMSException( "No destination could be found for name: " + subject + " for subscription: " + subscription );
79		}
80
81	0	MessageListener listener = subscription.getMessageListener();
82	0	if ( listener == null ) {
83	0	throw new JMSException( "No MessageListener is defined for subscription: " + subscription );
84		}
85
86		// if its an MDO the initialise it!
87	0	if ( listener instanceof MessageDrivenObject ) {
88	0	MessageDrivenObject mdo = (MessageDrivenObject) listener;
89	0	if ( mdo instanceof MessengerMDO ) {
90	0	MessengerMDO messengerMDO = (MessengerMDO) mdo;
91	0	messengerMDO.setMessenger( messenger );
92	0	messengerMDO.setMessengerManager( getMessengerManager() );
93		}
94	0	mdo.init( getServletContext() );
95		}
96
97	0	listener = wrapInStopWatch( listener );
98
99	0	String selector = subscription.getSelector();
100
101	0	ConsumerThread thread = subscription.getConsumerThread();
102	0	if (thread != null) {
103	0	log.info( "Subscribing to messenger: " + name + " destination: " + subject + " selector: " + selector + " with: " + thread );
104
105	0	thread.setMessenger(messenger);
106	0	thread.setDestination(destination);
107	0	thread.setSelector(selector);
108	0	thread.setListener(listener);
109	0	thread.start();
110		}
111		else {
112	0	if ( selector != null && selector.length() > 0 ) {
113	0	log.info( "Subscribing to messenger: " + name + " destination: " + subject + " selector: " + selector );
114
115	0	messenger.addListener( destination, selector, listener );
116		}
117		else {
118	0	log.info( "Subscribing to messenger: " + name + " destination: " + subject );
119
120	0	messenger.addListener( destination, listener );
121		}
122
123	0	log.info( "Subscribed with listener: " + listener );
124		}
125	0	}
126
127
128
129		public void unsubscribe() throws JMSException, ServletException {
130	0	SubscriptionList list = getSubscriptionList();
131	0	if ( list != null ) {
132	0	for (Iterator iter = list.getSubscriptions().iterator(); iter.hasNext(); ) {
133	0	Subscription subscription = (Subscription) iter.next();
134	0	unsubscribe( subscription );
135	0	}
136		}
137	0	}
138
139		public void unsubscribe( Subscription subscription ) throws JMSException, ServletException {
140		// lets unsubscribe first
141	0	String name = subscription.getConnection();
142	0	Messenger messenger = getMessenger( name );
143
144	0	MessageListener listener = subscription.getMessageListener();
145	0	if ( messenger != null && listener != null ) {
146	0	Destination destination = null;
147	0	String subject = subscription.getSubject();
148	0	if ( subject == null \|\| subject.length() == 0 ) {
149	0	log.error( "No destination defined for subscription: " + subscription );
150		}
151		else {
152		try {
153	0	destination = messenger.getDestination( subject );
154	0	if ( destination == null ) {
155	0	log.error( "No destination could be found for name: " + subject + " for subscription: " + subscription );
156		}
157		}
158	0	catch (JMSException e) {
159	0	log.error( "Could not create destination for name: " + subject + " for subscription: " + subscription, e );
160	0	}
161		}
162	0	if ( destination != null ) {
163		try {
164	0	String selector = subscription.getSelector();
165	0	if ( selector != null && selector.length() > 0 ) {
166	0	messenger.removeListener( destination, selector, listener );
167		}
168		else {
169	0	messenger.removeListener( destination, listener );
170		}
171		}
172	0	catch (JMSException e) {
173	0	log.error( "Could not unsubscribe to destination:" + destination + " for subscription: " + subscription, e );
174	0	}
175		}
176		}
177
178		// now lets destrory the MBO
179	0	if ( listener instanceof MessageDrivenObject ) {
180	0	MessageDrivenObject mdo = (MessageDrivenObject) listener;
181	0	mdo.destroy();
182		}
183	0	}
184
185
186		// Properties
187		//-------------------------------------------------------------------------
188
189		public MessengerManager getMessengerManager() throws JMSException {
190	0	return manager;
191		}
192
193		public void setMessengerManager(MessengerManager manager) {
194	0	this.manager = manager;
195	0	}
196
197		/**
198		* Returns the subscriptionList.
199		* @return SubscriptionList
200		*/
201		public SubscriptionList getSubscriptionList() {
202	0	return subscriptionList;
203		}
204
205		/**
206		* Sets the subscriptionList.
207		* @param subscriptionList The subscriptionList to set
208		*/
209		public void setSubscriptionList(SubscriptionList subscriptionList) {
210	0	this.subscriptionList = subscriptionList;
211	0	}
212
213		/**
214		* Returns the servletContext.
215		* @return ServletContext
216		*/
217		public ServletContext getServletContext() {
218	0	return servletContext;
219		}
220
221		/**
222		* Sets the servletContext.
223		* @param servletContext The servletContext to set
224		*/
225		public void setServletContext(ServletContext servletContext) {
226	0	this.servletContext = servletContext;
227	0	}
228
229		// Implementation methods
230		//-------------------------------------------------------------------------
231		/**
232		* Allows the MessageListener to be wrapped inside a stop watch message listener if required
233		*/
234		protected MessageListener wrapInStopWatch( MessageListener listener ) {
235	0	if ( useStopWatch ) {
236	0	return new StopWatchMessageListener( listener );
237		}
238	0	return listener;
239		}
240
241		protected Messenger getMessenger(String name) throws JMSException {
242	0	return getMessengerManager().getMessenger( name );
243		}
244
245		}