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 2000-2003 (C) Exoffice Technologies Inc. All Rights Reserved.
42 */
43
44 package org.exolab.jms.messagemgr;
45
46
47 import java.sql.Connection;
48 import java.sql.SQLException;
49 import java.util.Enumeration;
50 import java.util.HashMap;
51 import java.util.Vector;
52
53 import javax.jms.JMSException;
54 import javax.jms.Message;
55 import javax.transaction.TransactionManager;
56
57 import org.apache.commons.logging.Log;
58 import org.apache.commons.logging.LogFactory;
59
60 import org.exolab.jms.lease.BaseLease;
61 import org.exolab.jms.lease.LeaseEventListenerIfc;
62 import org.exolab.jms.lease.LeaseManager;
63 import org.exolab.jms.lease.MessageLease;
64 import org.exolab.jms.message.MessageHandle;
65 import org.exolab.jms.message.MessageImpl;
66 import org.exolab.jms.persistence.DatabaseService;
67 import org.exolab.jms.persistence.PersistenceException;
68 import org.exolab.jms.persistence.SQLHelper;
69
70
71 /***
72 * This is a helper class for registering leases for messages with
73 * LeaseManager. The lease is based on the JMSExpiration property of the
74 * message. <br>
75 * When the lease expires, the listener's onLeaseExpired() method is invoked
76 * with a MessageHandle object passed as the argument. <br>
77 * If JMSExpiration is 0, the message never expires. <br>
78 *
79 * @version $Revision: 1.16 $ $Date: 2003/08/17 01:32:24 $
80 * @author <a href="mailto:tima@intalio.com">Tim Anderson</a>
81 * @see MessageHandle
82 * @see LeaseManager
83 * @see LeaseEventListenerIfc
84 */
85 public class MessageLeaseHelper implements LeaseEventListenerIfc {
86
87 /***
88 * The listener to notify when a lease expires
89 */
90 private LeaseEventListenerIfc _listener;
91
92 /***
93 * A map of MessageId -> MessageLease objects, representing
94 * the active leases
95 */
96 private HashMap _leases = new HashMap();
97
98 /***
99 * A reference to the LeaseManager
100 */
101 private LeaseManager _leaseMgr;
102
103 /***
104 * The logger
105 */
106 private static final Log _log =
107 LogFactory.getLog(MessageLeaseHelper.class);
108
109
110 /***
111 * Construct a helper for the specified destination cache.
112 *
113 * @param listener the object to notify when a lease expires
114 * @throws PersistenceException if the destination is administered
115 * and the set of non-expired messages cannot be determined
116 */
117 public MessageLeaseHelper(DestinationCache listener)
118 throws PersistenceException {
119 _listener = listener;
120 _leaseMgr = LeaseManager.instance();
121
122 Connection connection = null;
123
124 try {
125 connection = DatabaseService.getConnection();
126
127
128 init(listener, connection);
129
130
131 connection.commit();
132 } catch (PersistenceException exception) {
133 SQLHelper.rollback(connection);
134 throw exception;
135 } catch (SQLException exception) {
136 SQLHelper.rollback(connection);
137 throw new PersistenceException(exception);
138 } finally {
139 SQLHelper.close(connection);
140 }
141 }
142
143 /***
144 * Construct a helper for the specified destination cache.
145 * <p>
146 * This method is only called during cache init time and is used to
147 * retrieve persistent messages with leases.
148 *
149 * @param connection the connection to use to retrieve persistent messages
150 * with leases
151 * @param listener the object to notify when a lease expires
152 * @throws PersistenceException if the destination is administered
153 * and the set of non-expired messages cannot be determined
154 */
155 public MessageLeaseHelper(Connection connection, DestinationCache listener)
156 throws PersistenceException {
157 _listener = listener;
158 _leaseMgr = LeaseManager.instance();
159
160 init(listener, connection);
161 }
162
163 /***
164 * Add a lease for message to notify listener when message expires.
165 * The lease uses JMSExpiration property of the message. If this
166 * is unset or 0, then no lease is registered. If non-zero, a
167 * MessageHandle object is registered with LeaseManager, and
168 * the listener will be notified with this object when the lease expires.
169 *
170 * @param message the message to add a lease for
171 */
172 public void addLease(MessageImpl message) {
173 synchronized (_leases) {
174
175 if (!_leases.containsKey(message.getMessageId())) {
176 try {
177 long expiry = message.getJMSExpiration();
178 if (expiry > 0) {
179
180
181
182 MessageHandle handle =
183 MessageHandleFactory.getHandle(message);
184 long duration = expiry - System.currentTimeMillis();
185 MessageLease lease = _leaseMgr.createMessageLease(
186 handle, (duration <= 0 ? 1 : duration), this);
187 _leases.put(handle.getMessageId(), lease);
188 }
189 } catch (JMSException exception) {
190 _log.error("Failed to create lease", exception);
191 }
192 }
193 }
194 }
195
196 /***
197 * Add a lease for the handle to notify listener when message expires.
198 * The lease uses JMSExpiration property of the message. If this
199 * is unset or 0, then no lease is registered. If non-zero, a
200 * MessageHandle object is registered with LeaseManager, and
201 * the listener will be notified with this object when the lease expires.
202 *
203 * @param handle message handle to add
204 */
205 public void addLease(MessageHandle handle) {
206 synchronized (_leases) {
207
208 if (!_leases.containsKey(handle.getMessageId())) {
209 long expiry = handle.getExpiryTime();
210 if (expiry != 0) {
211 long duration = expiry - System.currentTimeMillis();
212 MessageLease lease = _leaseMgr.createMessageLease(
213 handle, (duration <= 0 ? 1 : duration), this);
214 _leases.put(handle.getMessageId(), lease);
215 }
216 }
217 }
218 }
219
220 /***
221 * Remove a lease for a message
222 *
223 * @param message the message to remove the lease for
224 */
225 public void removeLease(MessageImpl message) {
226 try {
227 MessageHandle handle = MessageHandleFactory.getHandle(message);
228 if (handle != null) {
229 synchronized (_leases) {
230 _leases.remove(handle.getMessageId());
231 }
232 }
233 } catch (JMSException exception) {
234 _log.error("Failed to remove lease", exception);
235 }
236 }
237
238 /***
239 * Clears all leases
240 */
241 public void clear() {
242 Object[] leases = null;
243 synchronized (_leases) {
244 leases = _leases.values().toArray();
245 _leases.clear();
246 }
247
248 for (int i = 0; i < leases.length; ++i) {
249 BaseLease lease = (BaseLease) leases[i];
250 _leaseMgr.removeLease(lease);
251 }
252 }
253
254 /***
255 * Invoked when a lease has expired.
256 * <br>
257 * It removes the local lease, and notifies the listener.
258 *
259 * @param handle An instance of MessageHandle
260 */
261 public void onLeaseExpired(Object handle) {
262 synchronized (_leases) {
263 _leases.remove(((MessageHandle) handle).getMessageId());
264 }
265
266 _listener.onLeaseExpired(handle);
267 }
268
269 /***
270 * This method is used for all common initialization code. It basically
271 * retrieves all non-expired messages and places them in the memory
272 *
273 * @param listener the cache listening for expired leases.
274 * @param connection the persistent connection to use
275 * @throws PersistenceException if the destination is administered
276 * and the set of non-expired messages cannot be determined
277 */
278 protected void init(DestinationCache listener, Connection connection)
279 throws PersistenceException {
280
281
282
283
284 if (DatabaseService.getAdapter().checkDestination(
285 connection, listener.getDestinationByName())) {
286 Vector handles =
287 DatabaseService.getAdapter().getNonExpiredMessages(
288 connection, listener.getDestination());
289 if (handles != null) {
290 Enumeration iter = handles.elements();
291 while (iter.hasMoreElements()) {
292 PersistentMessageHandle handle =
293 (PersistentMessageHandle) iter.nextElement();
294 addLease(handle);
295 }
296 }
297 }
298 }
299
300 }