MantaConnectionConsumer


1   /*
2    * Copyright 2002 by
3    * <a HREF="http://www.coridan.com">Coridan</a>
4    * <a HREF="mailto: support@coridan.com ">support@coridan.com</a>
5    *
6    * The contents of this file are subject to the Mozilla Public License Version
7    * 1.1 (the "License"); you may not use this file except in compliance with the
8    * License.  You may obtain a copy of the License at
9    * http://www.mozilla.org/MPL/
10   *
11   * Software distributed under the License is distributed on an "AS IS" basis,
12   * WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
13   * for the specific language governing rights and limitations under the
14   * License.
15   *
16   * The Original Code is "MantaRay" (TM).
17   *
18   * The Initial Developer of the Original Code is Nimo 20-MAR-2004.
19   * Portions created by the Initial Developer are Copyright (C) 2006
20   * Coridan Inc.  All Rights Reserved.
21   *
22   * Contributor(s): all the names of the contributors are added in the source
23   * code where applicable.
24   *
25   * Alternatively, the contents of this file may be used under the terms of the
26   * LGPL license (the "GNU LESSER GENERAL PUBLIC LICENSE"), in which case the
27   * provisions of LGPL are applicable instead of those above.  If you wish to
28   * allow use of your version of this file only under the terms of the LGPL
29   * License and not to allow others to use your version of this file under
30   * the MPL, indicate your decision by deleting the provisions above and
31   * replace them with the notice and other provisions required by the LGPL.
32   * If you do not delete the provisions above, a recipient may use your version
33   * of this file under either the MPL or the GNU LESSER GENERAL PUBLIC LICENSE.
34   
35   *
36   * This library is free software; you can redistribute it and/or modify it
37   * under the terms of the MPL as stated above or under the terms of the GNU
38   * Lesser General Public License as published by the Free Software Foundation;
39   * either version 2.1 of the License, or any later version.
40   *
41   * This library is distributed in the hope that it will be useful, but WITHOUT
42   * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
43   * FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
44   * License for more details.
45   */
46  
47  package org.mr.api.jms;
48  
49  import java.util.ArrayList  ;
50  
51  import javax.jms.ConnectionConsumer  ;
52  import javax.jms.Destination  ;
53  import javax.jms.JMSException  ;
54  import javax.jms.Message  ;
55  import javax.jms.MessageListener  ;
56  import javax.jms.ServerSession  ;
57  import javax.jms.ServerSessionPool  ;
58  import javax.jms.Session  ;
59  
60  import org.apache.commons.logging.Log;
61  import org.apache.commons.logging.LogFactory;
62  import org.mr.IMessageListener;
63  import org.mr.api.jms.selector.syntax.Selector;
64  import org.mr.core.protocol.MantaBusMessage;
65  
66  /**
67   * This class implemets a ConnectionConsumer for the application servers' use.
68   * Upon registering an MDB, the application server will use this class.
69   * 
70   * @author Nimo 20-MAR-2004
71   *
72   */
73  public class MantaConnectionConsumer implements ConnectionConsumer  , IMessageListener, MessageListener   {
74      
75      // the destination that this listener listens to
76      private Destination   destination;
77      
78      // the selector used by this listener
79      private String   messageSelector;
80      
81      // If true this consumer will accumulate messages and not forward them
82      protected boolean holdMessages;
83      
84      // Used to hold messages in case we choose to hold messages and not 
85      // to forward them 
86      private ArrayList   msgs;
87      
88      // The logger
89      private Log log;
90      
91      // The consumer used to listen to the messages
92      private MantaMessageConsumer consumer;
93      
94      // This is the application-server supplied pooling mechanism.
95      private ServerSessionPool   sessionPool;
96      
97      // The MantaSession used by the ServerSession
98      private MantaSession session;
99      
100     private MantaConnection connection;
101     
102     // true when closing
103     private boolean close = false;
104     
105     
106     /**
107      * the constructor - create a consumer and hook up to the application server.
108      * 
109      * @param dest - the destination to register on
110      * @param msgSel - the message selector to use
111      * @param pool - the server pool
112      * @param maxMsg - max messages for a session
113      * @throws JMSException
114      */
115     protected MantaConnectionConsumer (MantaConnection connection, Destination   dest, String   msgSel, ServerSessionPool   pool, int maxMsg) throws JMSException   {
116         
117         if (pool == null) {
118             throw new JMSException  ("MNJMS00030 : APPLICATION SERVER ERROR. ServerSessionPool EMPTY. ERROR IN CONSTRUCTING MantaConnectionConsumer.");
119         }
120 
121         // Throw InvalidSelectorException if the selector is erroneous
122         new Selector(msgSel);  
123 
124         log = LogFactory.getLog("MantaConnectionConsumer");
125         this.sessionPool = pool;
126         this.connection = connection;
127         this.destination = dest;
128         this.messageSelector = msgSel;
129         
130         try {
131             // create the session from which we will receive the messages.
132             // the consumer is not used in the traditional way.
133             // we create it in order for the session to receive messages, but we don't
134             // use it to actually receive the messages. Instear we get the messages
135             // directly from the session by using IMessageListener interface.
136             session = (MantaSession)connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
137             consumer = (MantaMessageConsumer)session.createConsumer(this.destination, this.messageSelector);
138             // we register a listener that doesn't do anything only to enable
139             // the feature that farwards messages on the session. In the session code
140             // the message will be forwared to the IMessageListener we are registering here.
141             consumer.setMessageListener(this);
142             session.setBusMessageListener(this);
143         } 
144         catch (JMSException   e) {
145             this.close();
146             throw new JMSException  ("MNJMS00031 : APPLICATION SERVER ERROR. FAILED TO RECEIVE SESSION. ERROR IN CONSTRUCTING MantaConnectionConsumer.");
147         }
148     }
149     
150     protected MantaConnectionConsumer (MantaConnection connection, Destination   dest, String   msgSel, ServerSessionPool   pool, int maxMsg,boolean hold) throws JMSException   {
151         this(connection,dest,msgSel,pool,maxMsg);
152         msgs = new ArrayList  ();
153         holdMessages=hold;
154     }
155     
156     
157     /**
158      * close this consumer and clean up.
159      */
160     public void close() throws JMSException   {
161         
162         if (close) {
163             return;
164         }
165         
166         // kill the inner thread
167         close = true;
168         
169         // close the session
170         if (session != null) {
171             session.close();
172             session = null;
173         }
174         
175         // close the consumer if needed
176         if (consumer != null) {
177             consumer.close();
178             consumer = null;
179         }
180         
181         messageSelector = null;
182         destination = null;
183     }
184     
185     // when the session receives a message, it calls this method, which
186     // forwards the message to the application server's server session.
187     public void onMessage(MantaBusMessage message) {
188         try {
189             // forward the messages to the endpoint
190             ServerSession   serverSession = sessionPool.getServerSession();
191             MantaSession session = (MantaSession)serverSession.getSession();
192             session.addConsumerMessage(message);
193             serverSession.start();
194         }
195         catch (JMSException   e) {
196             log.error("Exception while forwarding a message to the Application Server.",e);
197         }
198     }
199     
200     
201     /**
202      * Return this consumer's session pool object.
203      */
204     public ServerSessionPool   getServerSessionPool() throws JMSException   {
205         return this.sessionPool;
206     }
207     
208     
209     // implement a MessageListener that doesn't do anything.
210     // the IMessageListener will receive the message directly
211     // from the session, in the form of MantaBusMessage.
212     public void onMessage(Message   message) {
213         // do nothing
214     }
215     
216     
217     public Message   receive() {
218         return receive(0);
219     }
220     
221     public MantaMessage receive(long timeout) {
222         MantaMessage mm = null;
223         
224         if (timeout == 0)
225             timeout = Long.MAX_VALUE;
226         
227         synchronized(msgs) {
228             try {
229                 if (msgs.isEmpty())  {
230                     msgs.wait(timeout);
231                 }
232             }
233             catch (InterruptedException   ie) {
234                 //nothing...
235             }
236             if (!msgs.isEmpty()) {
237                 mm = (MantaMessage) msgs.remove(0);
238             }
239         }
240         return mm;
241     }
242 }
243
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags