Queue


1   /*
2    * JORAM: Java(TM) Open Reliable Asynchronous Messaging
3    * Copyright (C) 2001 - 2007 ScalAgent Distributed Technologies
4    * Copyright (C) 2004 Bull SA
5    * Copyright (C) 1996 - 2000 Dyade
6    *
7    * This library is free software; you can redistribute it and/or
8    * modify it under the terms of the GNU Lesser General Public
9    * License as published by the Free Software Foundation; either
10   * version 2.1 of the License, or any later version.
11   * 
12   * This library 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 library; if not, write to the Free Software
19   * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307
20   * USA.
21   *
22   * Initial developer(s): Frederic Maistre (INRIA)
23   * Contributor(s): ScalAgent Distributed Technologies
24   */
25  package org.objectweb.joram.client.jms;
26  
27  import java.util.Vector  ;
28  import java.util.Hashtable  ;
29  import java.util.Properties  ;
30  import java.net.ConnectException  ;
31  
32  import javax.naming.NamingException  ;
33  
34  import javax.jms.JMSException  ;
35  
36  import org.objectweb.util.monolog.api.BasicLevel;
37  
38  import org.objectweb.joram.client.jms.admin.AdminException;
39  import org.objectweb.joram.client.jms.admin.AdminModule;
40  
41  import org.objectweb.joram.shared.admin.*;
42  
43  import org.objectweb.util.monolog.api.BasicLevel;
44  import org.objectweb.joram.shared.JoramTracing;
45  
46  import fr.dyade.aaa.util.management.MXWrapper;
47  
48  /**
49   *  Implements the <code>javax.jms.Queue</code> interface and provides
50   * Joram specific administration and monitoring methods. This is a proxy
51   * object a client uses to specify the destination of messages it is
52   * sending and the source of messages it receives.
53   *  
54   */
55  public class Queue extends Destination implements javax.jms.Queue  , QueueMBean {
56    private final static String   QUEUE_TYPE = "queue";
57  
58    public static boolean isQueue(String   type) {
59      return Destination.isAssignableTo(type, QUEUE_TYPE);
60    }
61  
62    // Used by jndi2 SoapObjectHelper
63    public Queue() {
64      super(QUEUE_TYPE);
65    }
66  
67    public Queue(String   name) {
68      super(name, QUEUE_TYPE);
69    }
70  
71    protected Queue(String   name, String   type) {
72      super(name, type);
73    }
74  
75    /**
76     * Gets the The Joram's internal unique identifier of this queue.
77     * API method.
78     *
79     * @exception JMSException  Actually never thrown.
80     * @return    The Joram's internal unique identifier.
81     */
82    public String   getQueueName() throws JMSException   {
83      return getName();
84    }
85  
86    /**
87     *  Admin method creating and deploying (or retrieving) a queue on a
88     * given server. First a destination with the specified name is searched
89     * on the given server, if it does not exist it is created. In any case,
90     * its provider-specific address is returned.
91     * <p>
92     *  The request fails if the target server does not belong to the platform,
93     * or if the destination deployement fails server side.
94     *
95     * @param serverId   The identifier of the server where deploying the queue.
96     * @param name       The name of the queue.
97     * @param className  The MOM's queue class name.
98     * @param prop       The queue properties.
99     *
100    * @exception ConnectException  If the admin connection is closed or broken.
101    * @exception AdminException  If the request fails.
102    */
103   public static Queue create(int serverId,
104                              String   name,
105                              String   className,
106                              Properties   prop)
107     throws ConnectException  , AdminException {
108     Queue queue = new Queue();
109     doCreate(serverId, name, className, prop, queue, QUEUE_TYPE);
110 
111     StringBuffer   buff = new StringBuffer  ();
112     buff.append("type=").append(QUEUE_TYPE);
113     buff.append(",name=").append(name);
114     try {
115       MXWrapper.registerMBean(queue, "joramClient", buff.toString());
116     } catch (Exception   e) {
117       if (JoramTracing.dbgClient.isLoggable(BasicLevel.DEBUG))
118         JoramTracing.dbgClient.log(BasicLevel.DEBUG, "registerMBean", e);
119     }
120     return queue;
121   }
122 
123   /**
124    * Admin method creating and deploying a queue on a given server.
125    * <p>
126    * The request fails if the target server does not belong to the platform,
127    * or if the destination deployement fails server side.
128    *
129    * @param serverId   The identifier of the server where deploying the queue.
130    * @param className  The queue class name.
131    * @param prop       The queue properties.
132    *
133    * @exception ConnectException  If the admin connection is closed or broken.
134    * @exception AdminException  If the request fails.
135    */
136   public static Queue create(int serverId,
137                              String   className,
138                              Properties   prop)
139                 throws ConnectException  , AdminException {
140     return create(serverId, null, className, prop);
141   }
142 
143   /**
144    * Admin method creating and deploying a queue on a given server.
145    * It creates a Joram's standart queue.
146    * <p>
147    * The request fails if the target server does not belong to the platform,
148    * or if the destination deployement fails server side.
149    *
150    * @param serverId   The identifier of the server where deploying the queue.
151    * @param prop       The queue properties.
152    *
153    * @exception ConnectException  If the admin connection is closed or broken.
154    * @exception AdminException  If the request fails.
155    */
156   public static Queue create(int serverId, Properties   prop)
157                 throws ConnectException  , AdminException {
158     return create(serverId, "org.objectweb.joram.mom.dest.Queue", prop);
159   }
160 
161   /**
162    * Admin method creating and deploying (or retrieving) a queue on a given
163    * server with a given name. First a destination with the specified name is
164    * searched on the given server, if it does not exist it is created. In any
165    * case, its provider-specific address is returned.
166    * <p>
167    * The request fails if the target server does not belong to the platform,
168    * or if the destination deployement fails server side.
169    *
170    * @param serverId  The identifier of the server where deploying the queue.
171    * @param name      The queue name. 
172    *
173    * @exception ConnectException  If the admin connection is closed or broken.
174    * @exception AdminException  If the request fails.
175    */
176   public static Queue create(int serverId, String   name)
177                 throws ConnectException  , AdminException {
178     return create(serverId, name, "org.objectweb.joram.mom.dest.Queue", null);
179   }
180 
181   /**
182    * Admin method creating and deploying (or retrieving) a queue on the
183    * local server. First a destination with the specified name is searched
184    * on the given server, if it does not exist it is created. In any case,
185    * its provider-specific address is returned.
186    * <p>
187    * The request fails if the destination deployement fails server side.
188    *
189    * @param name      The queue name. 
190    *
191    * @exception ConnectException  If the admin connection is closed or broken.
192    * @exception AdminException  If the request fails.
193    */
194   public static Queue create(String   name)
195                 throws ConnectException  , AdminException {
196     return create(AdminModule.getLocalServerId(),
197                   name,
198                   "org.objectweb.joram.mom.dest.Queue",
199                   null);
200   }
201  
202   /**
203    * Admin method creating and deploying a queue on a given server.
204    * <p>
205    * The request fails if the target server does not belong to the platform,
206    * or if the destination deployement fails server side.
207    *
208    * @param serverId   The identifier of the server where deploying the queue.
209    *
210    * @exception ConnectException  If the admin connection is closed or broken.
211    * @exception AdminException  If the request fails.
212    */
213   public static Queue create(int serverId)
214                 throws ConnectException  , AdminException {
215     return create(serverId, null, "org.objectweb.joram.mom.dest.Queue", null);
216   }
217 
218   /**
219    * Admin method creating and deploying a queue on the local server. 
220    * <p>
221    * The request fails if the destination deployement fails server side.
222    *
223    * @exception ConnectException  If the admin connection is closed or broken.
224    * @exception AdminException  If the request fails.
225    */
226   public static Queue create() throws ConnectException  , AdminException {
227     return create(AdminModule.getLocalServerId());
228   }
229 
230   /**
231    * Admin method setting or unsetting the threshold for this queue.
232    * <p>
233    * The request fails if the queue is deleted server side.
234    *
235    * @param threshold  The threshold value to be set (-1 for unsetting
236    *                   previous value).
237    *
238    * @exception ConnectException  If the admin connection is closed or broken.
239    * @exception AdminException  If the request fails.
240    */
241   public void setThreshold(int threshold)
242               throws ConnectException  , AdminException
243   {
244     if (threshold == -1)
245       AdminModule.doRequest(new UnsetQueueThreshold(agentId));
246     else
247       AdminModule.doRequest(new SetQueueThreshold(agentId, threshold));
248   } 
249 
250   /** 
251    * Monitoring method returning the threshold of this queue, -1 if not set.
252    * <p>
253    * The request fails if the queue is deleted server side.
254    *
255    * @exception ConnectException  If the admin connection is closed or broken.
256    * @exception AdminException  If the request fails.
257    */
258   public int getThreshold() throws ConnectException  , AdminException
259   {
260     Monitor_GetDMQSettings request = new Monitor_GetDMQSettings(agentId);
261     Monitor_GetDMQSettingsRep reply;
262     reply = (Monitor_GetDMQSettingsRep) AdminModule.doRequest(request);
263     
264     if (reply.getThreshold() == null)
265       return -1;
266     else
267       return reply.getThreshold().intValue();
268   }
269 
270   /**
271    * Admin method setting nbMaxMsg for this queue.
272    * <p>
273    * The request fails if the queue is deleted server side.
274    *
275    * @param nbMaxMsg  nb Max of Message (-1 no limit).
276    *
277    * @exception ConnectException  If the admin connection is closed or broken.
278    * @exception AdminException  If the request fails.
279    */
280   public void setNbMaxMsg(int nbMaxMsg)
281     throws ConnectException  , AdminException {
282     AdminModule.doRequest(new SetNbMaxMsg(agentId, nbMaxMsg));
283   } 
284 
285   /** 
286    * Monitoring method returning the nbMaxMsg of this queue, -1 if no limit.
287    * <p>
288    * The request fails if the queue is deleted server side.
289    *
290    * @exception ConnectException  If the admin connection is closed or broken.
291    * @exception AdminException  If the request fails.
292    */
293   public int getNbMaxMsg() 
294     throws ConnectException  , AdminException {
295     Monitor_GetNbMaxMsg request = new Monitor_GetNbMaxMsg(agentId);
296     Monitor_GetNbMaxMsgRep reply;
297     reply = (Monitor_GetNbMaxMsgRep) AdminModule.doRequest(request);
298     return reply.getNbMaxMsg();
299   }
300    
301   /**
302    * Monitoring method returning the number of pending messages on this queue.
303    * <p>
304    * The request fails if the queue is deleted server side.
305    *
306    * @exception ConnectException  If the admin connection is closed or broken.
307    * @exception AdminException  If the request fails.
308    */
309   public int getPendingMessages() throws ConnectException  , AdminException
310   {
311     Monitor_GetPendingMessages request =
312       new Monitor_GetPendingMessages(agentId);
313     Monitor_GetNumberRep reply;
314     reply = (Monitor_GetNumberRep) AdminModule.doRequest(request);
315 
316     return reply.getNumber();
317   }
318 
319   /**
320    * Monitoring method returning the number of pending requests on this queue.
321    * <p>
322    * The request fails if the queue is deleted server side.
323    *
324    * @exception ConnectException  If the admin connection is closed or broken.
325    * @exception AdminException  If the request fails.
326    */
327   public int getPendingRequests() throws ConnectException  , AdminException {
328     Monitor_GetPendingRequests request =
329       new Monitor_GetPendingRequests(agentId);
330     Monitor_GetNumberRep reply =
331       (Monitor_GetNumberRep) AdminModule.doRequest(request);
332 
333     return reply.getNumber();
334   }
335 
336   public String  [] getMessageIds() throws AdminException, ConnectException   {
337     GetQueueMessageIdsRep reply = 
338       (GetQueueMessageIdsRep)AdminModule.doRequest(
339         new GetQueueMessageIds(agentId));
340     return reply.getMessageIds();
341   }
342   
343   public javax.jms.Message   readMessage(String   msgId) throws AdminException, ConnectException  , JMSException   {
344     GetQueueMessageRep reply = 
345       (GetQueueMessageRep)AdminModule.doRequest(
346         new GetQueueMessage(agentId, msgId));
347     return Message.wrapMomMessage(null, reply.getMessage());
348   }
349 
350   public String   getMessageDigest(String   msgId)
351     throws AdminException, ConnectException  , JMSException   {
352     GetQueueMessageRep reply = 
353       (GetQueueMessageRep)AdminModule.doRequest(
354         new GetQueueMessage(agentId, msgId));
355     Message msg =  Message.wrapMomMessage(null, reply.getMessage());
356     StringBuffer   strbuf = new StringBuffer  ();
357     strbuf.append("Message: ").append(msg.getJMSMessageID());
358     strbuf.append("\n\tTo: ").append(msg.getJMSDestination());
359     strbuf.append("\n\tCorrelationId: ").append(msg.getJMSCorrelationID());
360     strbuf.append("\n\tDeliveryMode: ").append(msg.getJMSDeliveryMode());
361     strbuf.append("\n\tExpiration: ").append(msg.getJMSExpiration());
362     strbuf.append("\n\tPriority: ").append(msg.getJMSPriority());
363     strbuf.append("\n\tRedelivered: ").append(msg.getJMSRedelivered());
364     strbuf.append("\n\tReplyTo: ").append(msg.getJMSReplyTo());
365     strbuf.append("\n\tTimestamp: ").append(msg.getJMSTimestamp());
366     strbuf.append("\n\tType: ").append(msg.getJMSType());
367     return strbuf.toString();
368   }
369 
370   public Properties   getMessageHeader(String   msgId)
371     throws AdminException, ConnectException  , JMSException   {
372     GetQueueMessageRep reply = 
373       (GetQueueMessageRep)AdminModule.doRequest(
374         new GetQueueMessage(agentId, msgId));
375     Message msg =  Message.wrapMomMessage(null, reply.getMessage());
376 
377     Properties   prop = new Properties  ();
378     prop.setProperty("JMSMessageID", msg.getJMSMessageID());
379     prop.setProperty("JMSDestination", msg.getJMSDestination().toString());
380     if (msg.getJMSCorrelationID() != null)
381       prop.setProperty("JMSCorrelationID", msg.getJMSCorrelationID());
382     prop.setProperty("JMSDeliveryMode",
383                      new Integer  (msg.getJMSDeliveryMode()).toString());
384     prop.setProperty("JMSExpiration",
385                      new Long  (msg.getJMSExpiration()).toString());
386     prop.setProperty("JMSPriority",
387                      new Integer  (msg.getJMSPriority()).toString());
388     prop.setProperty("JMSRedelivered",
389                      new Boolean  (msg.getJMSRedelivered()).toString());
390     if (msg.getJMSReplyTo() != null)
391       prop.setProperty("JMSReplyTo", msg.getJMSReplyTo().toString());
392     prop.setProperty("JMSTimestamp",
393                      new Long  (msg.getJMSTimestamp()).toString());
394     if (msg.getJMSType() != null)
395       prop.setProperty("JMSType", msg.getJMSType());
396 
397     // Adds optional header properties
398     msg.getOptionalHeader(prop);
399 
400     return prop;
401   }
402 
403   public Properties   getMessageProperties(String   msgId)
404     throws AdminException, ConnectException  , JMSException   {
405     GetQueueMessageRep reply = 
406       (GetQueueMessageRep)AdminModule.doRequest(
407         new GetQueueMessage(agentId, msgId));
408     Message msg =  Message.wrapMomMessage(null, reply.getMessage());
409 
410     Properties   prop = new Properties  ();
411     msg.getProperties(prop);
412 
413     return prop;
414   }
415 
416   public void deleteMessage(
417     String   msgId)
418     throws AdminException, ConnectException   {
419     AdminModule.doRequest(new DeleteQueueMessage(agentId, msgId));
420   }
421 
422   public void clear() 
423     throws AdminException, ConnectException   {
424     AdminModule.doRequest(new ClearQueue(agentId));
425   }
426 
427   /**
428    * Adds a queue into the cluster this queue belongs to.
429    * If this queue doesn't belong to a cluster then a cluster is
430    * created by clustering this queue with the added queue.
431    * <p>
432    * The request fails if one or both of the queues are deleted, or
433    * can't belong to a cluster.
434    *
435    * @param addedQueue queue added to the cluster
436    *
437    * @exception ConnectException  If the admin connection is closed or broken.
438    * @exception AdminException  If the request fails.
439    */
440   public void addClusteredQueue(Queue addedQueue)
441     throws ConnectException  , AdminException {
442     AdminModule.doRequest(
443       new AddQueueCluster(agentId, addedQueue.getName()));
444   }
445 
446   /**
447    * Removes a queue from the cluster this queue belongs to.
448    * <p>
449    * The request fails if the queue does not exist or is not part of any 
450    * cluster.
451    *
452    * @param removedQueue queue removed from the cluster
453    * 
454    * @exception ConnectException  If the admin connection is closed or broken.
455    * @exception AdminException  If the request fails.
456    */
457   public void removeClusteredQueue(Queue removedQueue)
458     throws ConnectException  , AdminException {
459     AdminModule.doRequest(
460       new RemoveQueueCluster(agentId, removedQueue.getName()));
461   }
462 
463   /**
464    * Returns the reference of the queues that belong to the cluster.
465    *
466    * @exception ConnectException  If the admin connection is closed or broken.
467    * @exception AdminException  If the request fails.
468    */
469   public String  [] getQueueClusterElements()
470     throws ConnectException  , AdminException {
471     AdminReply reply = AdminModule.doRequest(
472       new ListClusterQueue(agentId));
473     Vector   list = (Vector  )reply.getReplyObject();
474     String  [] res = new String  [list.size()];
475     list.copyInto(res);
476     return res;
477   }
478 }
479
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags