View Javadoc

1   /*** 
2    * 
3    * Copyright 2004 Protique Ltd
4    * 
5    * Licensed under the Apache License, Version 2.0 (the "License"); 
6    * you may not use this file except in compliance with the License. 
7    * You may obtain a copy of the License at 
8    * 
9    * http://www.apache.org/licenses/LICENSE-2.0
10   * 
11   * Unless required by applicable law or agreed to in writing, software
12   * distributed under the License is distributed on an "AS IS" BASIS, 
13   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
14   * See the License for the specific language governing permissions and 
15   * limitations under the License. 
16   * 
17   **/
18  package org.codehaus.activemq.service.impl;
19  
20  import EDU.oswego.cs.dl.util.concurrent.ConcurrentHashMap;
21  import org.apache.commons.logging.Log;
22  import org.apache.commons.logging.LogFactory;
23  import org.codehaus.activemq.broker.BrokerClient;
24  import org.codehaus.activemq.message.ActiveMQMessage;
25  import org.codehaus.activemq.service.MessageContainerManager;
26  import org.codehaus.activemq.service.Service;
27  import org.codehaus.activemq.service.Subscription;
28  
29  import javax.jms.JMSException;
30  import java.util.Map;
31  import java.util.Iterator;
32  
33  /***
34   * A Dispatcher that polls for updates for active Message Consumers
35   *
36   * @version $Revision: 1.9 $
37   */
38  public class DispatchWorker implements Runnable, Service {
39      private static final Log log = LogFactory.getLog(DispatchWorker.class);
40      private static final int POLL_TIMEOUT = 250;
41  
42      private Map subscriptions = new ConcurrentHashMap(1000, 0.75f);
43      private Object lock = new Object();
44      private boolean active = true;
45      private boolean started = false;
46      private MessageContainerManager containerManager;
47  
48      /***
49       * Register the MessageContainerManager for the Dispatcher
50       *
51       * @param mcm
52       */
53      public void register(MessageContainerManager mcm) {
54          this.containerManager = mcm;
55      }
56  
57      /***
58       * Called to indicate that there is work to do on a Subscription this will wake up a Dispatch Worker if it is
59       * waiting for messages to dispatch
60       */
61      public void wakeup() {
62          synchronized (lock) {
63              active = true;
64              lock.notifyAll();
65          }
66      }
67  
68      /***
69       * Add an active subscription
70       *
71       * @param client
72       * @param sub
73       */
74      public void addActiveSubscription(BrokerClient client, Subscription sub) {
75          if (log.isDebugEnabled()) {
76              log.info("Adding subscription: " + sub + " to client: " + client);
77          }
78          subscriptions.put(sub, client);
79      }
80  
81      /***
82       * remove an active subscription
83       *
84       * @param client
85       * @param sub
86       */
87      public void removeActiveSubscription(BrokerClient client, Subscription sub) {
88          if (log.isDebugEnabled()) {
89              log.info("Removing subscription: " + sub + " from client: " + client);
90          }
91          subscriptions.remove(sub);
92      }
93  
94      /***
95       * dispatch messages to active Consumers
96       *
97       * @see java.lang.Runnable#run()
98       */
99      public void run() {
100         while (started) {
101             doPoll();
102             boolean dispatched = false;
103             try {
104                 // our collection will not throw concurrent modification exception
105                 for (Iterator iter = subscriptions.keySet().iterator(); iter.hasNext();) {
106                     Subscription sub = (Subscription) iter.next();
107                     if (sub != null && sub.isReadyToDispatch()) {
108                         dispatched = dispatchMessages(sub, dispatched);
109                     }
110                 }
111             }
112             catch (JMSException jmsEx) {
113                 log.error("Could not dispatch to Subscription: " + jmsEx, jmsEx);
114             }
115             if (!dispatched) {
116                 synchronized (lock) {
117                     active = false;
118                     if (!active && started) {
119                         try {
120                             lock.wait(POLL_TIMEOUT);
121                         }
122                         catch (InterruptedException e) {
123                         }
124                     }
125                 }
126             }
127         }
128     }
129 
130 
131     /***
132      * start the DispatchWorker
133      *
134      * @see org.codehaus.activemq.service.Service#start()
135      */
136     public void start() {
137         started = true;
138     }
139 
140     /***
141      * stop the DispatchWorker
142      *
143      * @see org.codehaus.activemq.service.Service#stop()
144      */
145     public void stop() {
146         started = false;
147     }
148 
149 
150     // Implementation methods
151     //-------------------------------------------------------------------------
152 
153     protected boolean dispatchMessages(Subscription subscription, boolean dispatched) throws JMSException {
154         ActiveMQMessage[] msgs = subscription.getMessagesToDispatch();
155         if (msgs != null && msgs.length > 0) {
156             BrokerClient client = (BrokerClient) subscriptions.get(subscription);
157             if (client == null) {
158                 log.warn("Null client for subscription: " + subscription);
159             }
160             else {
161                 for (int i = 0; i < msgs.length; i++) {
162                     ActiveMQMessage msg = msgs[i].shallowCopy();
163 
164                     if (log.isDebugEnabled()) {
165                         log.debug("Dispatching message: " + msg);
166                     }
167                     int[] consumerNos = new int[1];
168                     consumerNos[0] = subscription.getConsumerNumber();
169                     msg.setConsumerNos(consumerNos);
170                     client.dispatch(msg);
171                     dispatched = true;
172                 }
173             }
174         }
175         return dispatched;
176     }
177 
178     protected void doPoll() {
179         if (containerManager != null && started) {
180             try {
181                 containerManager.poll();
182             }
183             catch (JMSException e) {
184                 log.error("Error polling from the ContainerManager: ", e);
185             }
186         }
187     }
188 }