Page History

...

Having the Dispatcher signal that it has stopped processing will allow us to know that we have hit the stopped state. However, this will mean that we have the opportunity to process one extra message AFTER the rollback command has been requested.

Proposed Solution

Currently there is a lot of synchronisation to ensure that we can safely start the rollback process in the AMQSession before asking the Dispatcher to clean up its resources. To ensure signal that we have stopped the dispatcher and so can guarantee we are no longer holding a message will require more synchronisation, which is both error prone and will and additional complexity to the client.

The proposed alternative is to modify the FlowControllingBlockingQueue so that we can delegate all rollback processing to the Dispatcher. This removes the need to stop the Dispatcher and if the Dispatcher is performing the rollback then it can be sure it is not currently processing an UnprocessedMessage.

While delegating the rollback of consumed messages to the Dispatcher it makes sense to give the Dispatcher more formal control over the receipt and dispatching of incoming messages. By extracting the Dispatcher from the Session class we can simplify the both classes. Locking will be reduced and the responsibility of message processing will be more cleanly delegated to the Dispatcher.

/** 
 * Dispatcher is responsible for delivering messages from the IO layer to registered consumers.
 * The dispatcher provides asynchronous dispatch via Consumer MessageListeners and
 * synchronous delivery by placing incoming messages in a Consumers recieveQueue.
 */
public interface DispatcherInterface<C extends BasicMessageConsumer>
{
        /** Start this Dispatcher if required */
        public void startDispatcherIfNecessary(boolean initiallyStopped);
        /** Close this Dispatcher and release any resources held. Cannot be re-opened */
        public void close();
        /** Register a consumer to receive messages from this Dispatcher. */
        public void registerConsumer(C consumer);
        /** Unregister a consumer from this Dispatcher @return C removed consumer */
        public C unregisterConsumer(C consumer);
        /** Rollback the received message state in this Dispatcher */
        public void rollback();
        /** Process the received message from the IO Layer */
        public void dispatchMessage(UnprocessedMessage message);
}

The clean interface with the Dispatcher from the Session means that we can more clearly delegate the rollback() control to the Dispatcher. The FlowControllingBlockingQueue will need to be augmented so that when an asynchronous request for rollback is made the Dispatcher can then pick up on this 'ServiceRequest'.

Steps

...

Further Details

After a discussion with Rafi/Rob on the recent Python changes expending effort in refactoring the client is probably not worth the effort. If the client message delivery were re-written to mirror the approach taken in the Python codebase then it would be simplier and easier to reason about. As a result I have devised a much smaller, though slightly ugly approach that will address our immediate rollback issues. The approach can be found here .

Comment Responses