1 /***
2 * Redistribution and use of this software and associated documentation
3 * ("Software"), with or without modification, are permitted provided
4 * that the following conditions are met:
5 *
6 * 1. Redistributions of source code must retain copyright
7 * statements and notices. Redistributions must also contain a
8 * copy of this document.
9 *
10 * 2. Redistributions in binary form must reproduce the
11 * above copyright notice, this list of conditions and the
12 * following disclaimer in the documentation and/or other
13 * materials provided with the distribution.
14 *
15 * 3. The name "Exolab" must not be used to endorse or promote
16 * products derived from this Software without prior written
17 * permission of Exoffice Technologies. For written permission,
18 * please contact info@exolab.org.
19 *
20 * 4. Products derived from this Software may not be called "Exolab"
21 * nor may "Exolab" appear in their names without prior written
22 * permission of Exoffice Technologies. Exolab is a registered
23 * trademark of Exoffice Technologies.
24 *
25 * 5. Due credit should be given to the Exolab Project
26 * (http://www.exolab.org/).
27 *
28 * THIS SOFTWARE IS PROVIDED BY EXOFFICE TECHNOLOGIES AND CONTRIBUTORS
29 * ``AS IS'' AND ANY EXPRESSED OR IMPLIED WARRANTIES, INCLUDING, BUT
30 * NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND
31 * FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
32 * EXOFFICE TECHNOLOGIES OR ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
33 * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
34 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
35 * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
36 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
37 * STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
38 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
39 * OF THE POSSIBILITY OF SUCH DAMAGE.
40 *
41 * Copyright 2002-2003 (C) Exoffice Technologies Inc. All Rights Reserved.
42 *
43 * $Id: SentMessageCache.java,v 1.5 2003/08/07 13:33:08 tanderson Exp $
44 *
45 * Date Author Changes
46 * 30/12/2002 tanderson Extracted from JmsServerSession
47 */
48 package org.exolab.jms.server;
49
50 import java.sql.Connection;
51 import java.util.Collections;
52 import java.util.Iterator;
53 import java.util.LinkedList;
54 import java.util.List;
55
56 import javax.jms.Session;
57
58 import org.apache.commons.logging.Log;
59 import org.apache.commons.logging.LogFactory;
60
61 import org.exolab.jms.client.JmsDestination;
62 import org.exolab.jms.client.JmsQueue;
63 import org.exolab.jms.client.JmsTopic;
64 import org.exolab.jms.message.MessageHandle;
65 import org.exolab.jms.message.MessageId;
66 import org.exolab.jms.messagemgr.ConsumerManager;
67 import org.exolab.jms.messagemgr.DestinationManager;
68 import org.exolab.jms.messagemgr.MessageHandleFactory;
69 import org.exolab.jms.messagemgr.PersistentMessageHandle;
70 import org.exolab.jms.messagemgr.QueueDestinationCache;
71 import org.exolab.jms.messagemgr.TopicConsumerEndpoint;
72 import org.exolab.jms.persistence.DatabaseService;
73 import org.exolab.jms.persistence.PersistenceException;
74
75
76 /***
77 * Helper class to cache all sent messages and unacked messages for a
78 * session. It also does some other processing like marking the message
79 * as sent to minimize the number of transactions.
80 * <p>
81 * Messages will only be added to the cache, if the session is transacted
82 * or the ack mode for the session is set to CLIENT_ACKNOWLEDGE
83 *
84 * @version $Revision: 1.5 $ $Date: 2003/08/07 13:33:08 $
85 * @author <a href="mailto:jima@exoffice.com">Jim Alateras</a>
86 * @author <a href="mailto:tma@netspace.net.au">Tim Anderson</a>
87 * @see JmsServerSession
88 */
89 class SentMessageCache {
90
91 /***
92 * The session which owns this instance
93 */
94 private JmsServerSession _session;
95
96 /***
97 * Holds a list of unacked messages in the order they were sent
98 */
99 private List _unackedMessages = Collections.synchronizedList(
100 new LinkedList());
101
102 /***
103 * The logger
104 */
105 private static final Log _log = LogFactory.getLog(SentMessageCache.class);
106
107
108 /***
109 * Construct a new <code>SentMessageCache</code>
110 *
111 * @param session the session which manages this
112 */
113 public SentMessageCache(JmsServerSession session) {
114 _session = session;
115 }
116
117 /***
118 * Mark the message as delivered and do one of the following.
119 * <p>
120 * If the session is transacted or the session ack mode is set to
121 * CLIENT_ACKNOWLEDGE then add this message to the list of unacked
122 * messages.
123 * <p>
124 * If the session ack mode is anything else then destroy the handle.
125 * <p>
126 * If the handle is a reference to a persistent message then conduct
127 * this work in the context of a transaction.
128 *
129 * @param handle the message handle that should be acked
130 */
131 public void process(MessageHandle handle) {
132 if (handle instanceof PersistentMessageHandle) {
133 PersistentMessageHandle phandle = (PersistentMessageHandle) handle;
134 Connection connection = null;
135 try {
136 connection = DatabaseService.getConnection();
137
138 phandle.setDelivered(true);
139 if (_session.isTransacted() ||
140 _session.getAckMode() == Session.CLIENT_ACKNOWLEDGE) {
141 _unackedMessages.add(handle);
142 MessageHandleFactory.updatePersistentHandle(
143 connection, phandle);
144 } else {
145
146 MessageHandleFactory.destroyPersistentHandle(
147 connection, phandle);
148 }
149 connection.commit();
150 } catch (PersistenceException exception) {
151 if (connection != null) {
152 try {
153 connection.rollback();
154 } catch (Exception ignore) {
155 }
156 }
157 _log.error("Error in SentMessageCache.process", exception);
158 } catch (Exception exception) {
159 _log.error("Error in SentMessageCache.process", exception);
160 } finally {
161 if (connection != null) {
162 try {
163 connection.close();
164 } catch (Exception ignore) {
165 }
166 }
167 }
168 } else {
169 handle.setDelivered();
170 if (_session.isTransacted() ||
171 _session.getAckMode() == Session.CLIENT_ACKNOWLEDGE) {
172 _unackedMessages.add(handle);
173 } else {
174 handle.destroy();
175 }
176 }
177 }
178
179 /***
180 * Acknowledge the specified messages in the cache and all previously
181 * sent messages.
182 *
183 * @param msgid the message id of the message to ack
184 * @param clientid the client id that sent the ack.
185 */
186 public void acknowledgeMessage(MessageId msgid, long clientid) {
187
188
189 boolean exists = false;
190 Iterator iterator = _unackedMessages.iterator();
191 while (iterator.hasNext()) {
192 MessageHandle handle = (MessageHandle) iterator.next();
193 if (handle.getClientId() == clientid &&
194 handle.getMessageId().equals(msgid)) {
195 exists = true;
196 break;
197 }
198 }
199
200 if (exists) {
201 boolean intransaction = false;
202 Connection connection = null;
203
204 try {
205
206
207
208 while (!_unackedMessages.isEmpty()) {
209 MessageHandle handle =
210 (MessageHandle) _unackedMessages.remove(0);
211 if (handle instanceof PersistentMessageHandle) {
212 if (!intransaction) {
213 connection = DatabaseService.getConnection();
214
215
216 intransaction = true;
217 }
218 MessageHandleFactory.destroyPersistentHandle(
219 connection, (PersistentMessageHandle) handle);
220 } else {
221 handle.destroy();
222 }
223
224
225
226 if (handle.getClientId() == clientid &&
227 handle.getMessageId().equals(msgid)) {
228 if (intransaction) {
229 connection.commit();
230 intransaction = false;
231 }
232 break;
233 }
234 }
235 } catch (PersistenceException exception) {
236 if (connection != null) {
237 try {
238 connection.rollback();
239 } catch (Exception ignore) {
240 }
241 }
242 _log.error("Error in SentMessageCache.acknowledgeMessage",
243 exception);
244 } catch (Exception exception) {
245 _log.error("Error in SentMessageCache.acknowledgeMessage",
246 exception);
247 } finally {
248 if (connection != null) {
249 try {
250 connection.close();
251 } catch (Exception ignore) {
252 }
253 }
254 }
255 }
256 }
257
258 /***
259 * Acknowledge all the messages in the cache
260 */
261 public void acknowledgeAllMessages() {
262 boolean intransaction = false;
263 Connection connection = null;
264
265 try {
266
267
268
269 while (!_unackedMessages.isEmpty()) {
270 MessageHandle handle =
271 (MessageHandle) _unackedMessages.remove(0);
272 if (handle instanceof PersistentMessageHandle) {
273 if (!intransaction) {
274 connection = DatabaseService.getConnection();
275 intransaction = true;
276 }
277 MessageHandleFactory.destroyPersistentHandle(
278 connection, (PersistentMessageHandle) handle);
279 } else {
280 handle.destroy();
281 }
282 }
283
284 if (intransaction) {
285 connection.commit();
286 intransaction = false;
287 }
288 } catch (PersistenceException exception) {
289 if (connection != null) {
290 try {
291 connection.rollback();
292 } catch (Exception ignore) {
293 }
294 }
295 _log.error("Error in SentMessageCache.acknowledgeMessage",
296 exception);
297 } catch (Exception exception) {
298 _log.error("Error in SentMessageCache.acknowledgeMessage",
299 exception);
300 } finally {
301 if (connection != null) {
302 try {
303 connection.close();
304 } catch (Exception ignore) {
305 }
306 }
307 }
308 }
309
310 /***
311 * Clear the cache by removing each entry and doing one of the
312 * following.
313 * <p>
314 * If the entry is for a queue destination then return it to
315 * the appropriate destination cache. If the cache does not
316 * exist, because it may have been garbage collected then we
317 * need to recreated.
318 * <p>
319 * If the entry is for a topic destination and check if the
320 * corresponding endpoint is still active. If it is then return it
321 * to the endpoint, otherwise do nothing.
322 */
323 public void clear() {
324
325 Object[] unacked = _unackedMessages.toArray();
326 _unackedMessages.clear();
327
328 int count = unacked.length;
329 for (int index = 0; index < count; index++) {
330 MessageHandle handle = (MessageHandle) unacked[index];
331 JmsDestination dest = handle.getDestination();
332
333 if (dest instanceof JmsTopic) {
334 ConsumerManager conmgr = ConsumerManager.instance();
335 long clientId = handle.getClientId();
336 TopicConsumerEndpoint endpoint = (TopicConsumerEndpoint)
337 _session.getConsumerEndpoint(clientId);
338
339
340 if (endpoint != null) {
341 endpoint.returnMessage(handle);
342 }
343 } else if (dest instanceof JmsQueue) {
344 DestinationManager destmgr = DestinationManager.instance();
345 QueueDestinationCache cache = (QueueDestinationCache)
346 destmgr.getDestinationCache(dest);
347
348
349
350 if (cache == null) {
351 cache = (QueueDestinationCache)
352 destmgr.createDestinationCache(dest);
353
354
355 if (cache != null) {
356 cache.returnMessage(handle);
357 } else {
358 _log.error("Failed to return message " + handle
359 + " to destination cache " + dest);
360 }
361 } else {
362
363 cache.returnMessage(handle);
364 }
365 }
366 }
367 }
368
369 /***
370 * Check whether the specified handle is in the list of unacked
371 * messages
372 *
373 * @param handle the handle to check
374 * @return <code>true</code> if it is in the list of unacked messages
375 */
376 public boolean handleInCache(MessageHandle handle) {
377 return _unackedMessages.contains(handle);
378 }
379
380 /***
381 * Remove the specified handle from the cache
382 *
383 * @param handle the handle to remove
384 */
385 public void removeHandle(MessageHandle handle) {
386 _unackedMessages.remove(handle);
387 }
388
389 }