001 /*
002 *
003 * Licensed to the Apache Software Foundation (ASF) under one
004 * or more contributor license agreements. See the NOTICE file
005 * distributed with this work for additional information
006 * regarding copyright ownership. The ASF licenses this file
007 * to you under the Apache License, Version 2.0 (the
008 * "License"); you may not use this file except in compliance
009 * with the License. You may obtain a copy of the License at
010 *
011 * http://www.apache.org/licenses/LICENSE-2.0
012 *
013 * Unless required by applicable law or agreed to in writing,
014 * software distributed under the License is distributed on an
015 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
016 * KIND, either express or implied. See the License for the
017 * specific language governing permissions and limitations
018 * under the License.
019 *
020 */
021 package org.apache.qpid.client.util;
022
023 import java.util.concurrent.TimeUnit;
024 import java.util.concurrent.atomic.AtomicBoolean;
025 import java.util.concurrent.locks.Condition;
026 import java.util.concurrent.locks.ReentrantLock;
027
028 import org.apache.qpid.AMQException;
029 import org.apache.qpid.AMQTimeoutException;
030 import org.apache.qpid.client.failover.FailoverException;
031 import org.apache.qpid.framing.AMQMethodBody;
032 import org.apache.qpid.protocol.AMQMethodEvent;
033 import org.apache.qpid.protocol.AMQMethodListener;
034
035 /**
036 * BlockingWaiter is a 'rendezvous' which delegates handling of
037 * incoming Objects to a listener implemented as a sub-class of this and hands off the process or
038 * error to a consumer. The producer of the event does not have to wait for the consumer to take the event, so this
039 * differs from a 'rendezvous' in that sense.
040 *
041 * <p/>BlockingWaiters are used to coordinate when waiting for an an event that expect a response.
042 * They are always used in a 'one-shot' manner, that is, to recieve just one response. Usually the caller has to register
043 * them as method listeners with an event dispatcher and remember to de-register them (in a finally block) once they
044 * have been completed.
045 *
046 * <p/>The {@link #process} must return <tt>true</tt> on any incoming method that it handles. This indicates to
047 * this listeners that the object just processed ends the waiting process.
048 *
049 * <p/>Errors from the producer are rethrown to the consumer.
050 *
051 * <p/><table id="crc"><caption>CRC Card</caption>
052 * <tr><th> Responsibilities <th> Collaborations </td>
053 * <tr><td> Accept generic objects as events for processing via {@link #process}. <td>
054 * <tr><td> Delegate handling and undserstanding of the object to a concrete implementation. <td>
055 * <tr><td> Block until {@link #process} determines that waiting is no longer required <td>
056 * <tr><td> Propagate the most recent exception to the consumer.<td>
057 * </table>
058 *
059 * @todo Interuption is caught but not handled. This could be allowed to fall through. This might actually be usefull
060 * for fail-over where a thread is blocking when failure happens, it could be interrupted to abandon or retry
061 * when this happens. At the very least, restore the interrupted status flag.
062 * @todo If the retrotranslator can handle it, could use a SynchronousQueue to implement this rendezvous. Need to
063 * check that SynchronousQueue has a non-blocking put method available.
064 */
065 public abstract class BlockingWaiter<T>
066 {
067 /** This flag is used to indicate that the blocked for method has been received. */
068 private volatile boolean _ready = false;
069
070 /** This flag is used to indicate that the received error has been processed. */
071 private volatile boolean _errorAck = false;
072
073 /** Used to protect the shared event and ready flag between the producer and consumer. */
074 private final ReentrantLock _lock = new ReentrantLock();
075
076 /** Used to signal that a method has been received */
077 private final Condition _receivedCondition = _lock.newCondition();
078
079 /** Used to signal that a error has been processed */
080 private final Condition _errorConditionAck = _lock.newCondition();
081
082 /** Used to hold the most recent exception that is passed to the {@link #error(Exception)} method. */
083 private volatile Exception _error;
084
085 /** Holds the incomming Object. */
086 protected Object _doneObject = null;
087 private AtomicBoolean _waiting = new AtomicBoolean(false);
088 private boolean _closed = false;
089
090 /**
091 * Delegates processing of the incomming object to the handler.
092 *
093 * @param object The object to process.
094 *
095 * @return <tt>true</tt> if the waiting is complete, <tt>false</tt> if waiting should continue.
096 */
097 public abstract boolean process(T object);
098
099 /**
100 * An Object has been received and should be processed to see if our wait condition has been reached.
101 *
102 * @param object The object received.
103 *
104 * @return <tt>true</tt> if the waiting is complete, <tt>false</tt> if waiting should continue.
105 */
106 public boolean received(T object)
107 {
108
109 boolean ready = process(object);
110
111 if (ready)
112 {
113 // we only update the flag from inside the synchronized block
114 // so that the blockForFrame method cannot "miss" an update - it
115 // will only ever read the flag from within the synchronized block
116 _lock.lock();
117 try
118 {
119 _doneObject = object;
120 _ready = ready;
121 _receivedCondition.signal();
122 }
123 finally
124 {
125 _lock.unlock();
126 }
127 }
128
129 return ready;
130 }
131
132 /**
133 * Blocks until an object is received that is handled by process, or the specified timeout
134 * has passed.
135 *
136 * Once closed any attempt to wait will throw an exception.
137 *
138 * @param timeout The timeout in milliseconds.
139 *
140 * @return The object that resolved the blocking.
141 *
142 * @throws AMQException
143 * @throws FailoverException
144 */
145 public Object block(long timeout) throws AMQException, FailoverException
146 {
147 long nanoTimeout = TimeUnit.MILLISECONDS.toNanos(timeout);
148
149 _lock.lock();
150
151 try
152 {
153 if (_closed)
154 {
155 throw throwClosedException();
156 }
157
158 if (_error == null)
159 {
160 _waiting.set(true);
161
162 while (!_ready)
163 {
164 try
165 {
166 if (timeout == -1)
167 {
168 _receivedCondition.await();
169 }
170 else
171 {
172 nanoTimeout = _receivedCondition.awaitNanos(nanoTimeout);
173
174 if (nanoTimeout <= 0 && !_ready && _error == null)
175 {
176 _error = new AMQTimeoutException("Server did not respond in a timely fashion", null);
177 _ready = true;
178 }
179 }
180 }
181 catch (InterruptedException e)
182 {
183 System.err.println(e.getMessage());
184 // IGNORE -- //fixme this isn't ideal as being interrupted isn't equivellant to sucess
185 // if (!_ready && timeout != -1)
186 // {
187 // _error = new AMQException("Server did not respond timely");
188 // _ready = true;
189 // }
190 }
191 }
192 }
193
194 if (_error != null)
195 {
196 if (_error instanceof AMQException)
197 {
198 throw (AMQException) _error;
199 }
200 else if (_error instanceof FailoverException)
201 {
202 // This should ensure that FailoverException is not wrapped and can be caught.
203 throw (FailoverException) _error; // needed to expose FailoverException.
204 }
205 else
206 {
207 throw new AMQException("Woken up due to " + _error.getClass(), _error);
208 }
209 }
210
211 }
212 finally
213 {
214 _waiting.set(false);
215
216 //Release Error handling thread
217 if (_error != null)
218 {
219 _errorAck = true;
220 _errorConditionAck.signal();
221
222 _error = null;
223 }
224 _lock.unlock();
225 }
226
227 return _doneObject;
228 }
229
230 /**
231 * This is a callback, called when an error has occured that should interupt any waiter.
232 * It is also called from within this class to avoid code repetition but it should only be called by the MINA threads.
233 *
234 * Once closed any notification of an exception will be ignored.
235 *
236 * @param e The exception being propogated.
237 */
238 public void error(Exception e)
239 {
240 // set the error so that the thread that is blocking (against blockForFrame())
241 // can pick up the exception and rethrow to the caller
242
243 _lock.lock();
244
245 if (_closed)
246 {
247 return;
248 }
249
250 if (_error == null)
251 {
252 _error = e;
253 }
254 else
255 {
256 System.err.println("WARNING: new error arrived while old one not yet processed");
257 }
258
259 try
260 {
261 if (_waiting.get())
262 {
263
264 _ready = true;
265 _receivedCondition.signal();
266
267 while (!_errorAck)
268 {
269 try
270 {
271 _errorConditionAck.await();
272 }
273 catch (InterruptedException e1)
274 {
275 System.err.println(e.getMessage());
276 }
277 }
278 _errorAck = false;
279 }
280 }
281 finally
282 {
283 _lock.unlock();
284 }
285 }
286
287 /**
288 * Close this Waiter so that no more errors are processed.
289 * This is a preventative method to ensure that a second error thread does not get stuck in the error method after
290 * the await has returned. This has not happend but in practise but if two errors occur on the Connection at
291 * the same time then it is conceiveably possible for the second to get stuck if the first one is processed by a
292 * waiter.
293 *
294 * Once closed any attempt to wait will throw an exception.
295 * Any notification of an exception will be ignored.
296 */
297 public void close()
298 {
299 _lock.lock();
300 try
301 {
302 //if we have already closed then our job is done.
303 if (_closed)
304 {
305 return;
306 }
307
308 //Close Waiter so no more exceptions are processed
309 _closed = true;
310
311 //Wake up any await() threads
312
313 //If we are waiting then use the error() to wake them up.
314 if (_waiting.get())
315 {
316 error(throwClosedException());
317 }
318 //If they are not waiting then there is nothing to do.
319
320 // Wake up any error handling threads
321
322 if (!_errorAck)
323 {
324 _errorAck = true;
325 _errorConditionAck.signal();
326
327 _error = null;
328 }
329 }
330 finally
331 {
332 _lock.unlock();
333 }
334 }
335
336 /**
337 * Helper method to generate the a closed Exception.
338 *
339 * todo: This should be converted to something more friendly.
340 *
341 * @return AMQException to throw to waiters when the Waiter is closed.
342 */
343 private AMQException throwClosedException()
344 {
345 return new AMQException(null, "Waiter was closed.", null);
346 }
347
348 }
|