Merged the refactor to a common AMQMethodListener class on trunk, plus the race condition fix of Robert Godfrey. This opens the way for Request and Response managers to use a common event dispatch for both client and server.

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/qpid.0-9@496326 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 26 insertions, 69 deletions

diff --git a/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMethodListener.java b/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMethodListener.java
deleted file mode 100644
index 6596da1f8f..0000000000
--- a/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMethodListener.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- * 
- *   http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-package org.apache.qpid.server.protocol;
-
-import org.apache.qpid.AMQException;
-import org.apache.qpid.protocol.AMQMethodEvent;
-import org.apache.qpid.server.queue.QueueRegistry;
-import org.apache.qpid.server.exchange.ExchangeRegistry;
-import org.apache.qpid.framing.AMQMethodBody;
-
-/**
- * Interface that allows classes to register for interest in protocol method frames.
- * 
- */
-public interface AMQMethodListener
-{
-    /**
-     * Invoked when a method frame has been received
-     * @param evt the event that contains the method and channel
-     * @param protocolSession the protocol session associated with the event
-     * @return true if the handler has processed the method frame, false otherwise. Note
-     * that this does not prohibit the method event being delivered to subsequent listeners
-     * but can be used to determine if nobody has dealt with an incoming method frame.
-     * @throws AMQException if an error has occurred. This exception will be delivered
-     * to all registered listeners using the error() method (see below) allowing them to
-     * perform cleanup if necessary.
-     */
-    <B extends AMQMethodBody> boolean methodReceived(AMQMethodEvent<B> evt,
-                           AMQProtocolSession protocolSession,
-                           QueueRegistry queueRegistry,
-                           ExchangeRegistry exchangeRegistry) throws AMQException;
-
-    /**
-     * Callback when an error has occurred. Allows listeners to clean up.
-     * @param e
-     */
-    void error(AMQException e);
-}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java b/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java
index 558d5f4aa6..1505b2cfc5 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java
@@ -42,6 +42,7 @@ import org.apache.qpid.framing.HeartbeatBody;
 import org.apache.qpid.codec.AMQCodecFactory;
 import org.apache.qpid.codec.AMQDecoder;
 import org.apache.qpid.protocol.AMQMethodEvent;
+import org.apache.qpid.protocol.AMQMethodListener;
 
 import org.apache.qpid.server.AMQChannel;
 import org.apache.qpid.server.RequiredDeliveryException;
@@ -111,7 +112,16 @@ public class AMQMinaProtocolSession implements AMQProtocolSession,
                                   AMQCodecFactory codecFactory)
             throws AMQException
     {
-        this(session, queueRegistry, exchangeRegistry, codecFactory, new AMQStateManager());
+        _stateManager = new AMQStateManager(queueRegistry, exchangeRegistry, this);
+        _minaProtocolSession = session;
+        session.setAttachment(this);
+        _frameListeners.add(_stateManager);
+        _queueRegistry = queueRegistry;
+        _exchangeRegistry = exchangeRegistry;
+        _codecFactory = codecFactory;
+        _managedObject = createMBean();
+        _managedObject.register();
+//        this(session, queueRegistry, exchangeRegistry, codecFactory, new AMQStateManager());
     }
 
     public AMQMinaProtocolSession(IoSession session, QueueRegistry queueRegistry, ExchangeRegistry exchangeRegistry,
@@ -263,7 +273,7 @@ public class AMQMinaProtocolSession implements AMQProtocolSession,
 //             boolean wasAnyoneInterested = false;
 //             for (AMQMethodListener listener : _frameListeners)
 //             {
-//                 wasAnyoneInterested = listener.methodReceived(evt, this, _queueRegistry, _exchangeRegistry) ||
+//                 wasAnyoneInterested = listener.methodReceived(evt) ||
 //                                       wasAnyoneInterested;
 //             }
 //             if (!wasAnyoneInterested)
@@ -276,7 +286,7 @@ public class AMQMinaProtocolSession implements AMQProtocolSession,
 //             _logger.error("Closing channel due to: " + e.getMessage());
 //             writeFrame(e.getCloseFrame(frame.channel));
 //         }
-//         catch (AMQException e)
+//         catch (Exception e)
 //         {
 //             for (AMQMethodListener listener : _frameListeners)
 //             {
diff --git a/java/broker/src/main/java/org/apache/qpid/server/state/AMQStateManager.java b/java/broker/src/main/java/org/apache/qpid/server/state/AMQStateManager.java
index 18b8041e7c..ed266bdef6 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/state/AMQStateManager.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/state/AMQStateManager.java
@@ -25,7 +25,7 @@ import org.apache.qpid.framing.*;
 import org.apache.qpid.protocol.AMQMethodEvent;
 import org.apache.qpid.server.exchange.ExchangeRegistry;
 import org.apache.qpid.server.handler.*;
-import org.apache.qpid.server.protocol.AMQMethodListener;
+import org.apache.qpid.protocol.AMQMethodListener;
 import org.apache.qpid.server.protocol.AMQProtocolSession;
 import org.apache.qpid.server.queue.QueueRegistry;
 import org.apache.log4j.Logger;
@@ -44,6 +44,9 @@ public class AMQStateManager implements AMQMethodListener
 {
     private static final Logger _logger = Logger.getLogger(AMQStateManager.class);
 
+    private final QueueRegistry _queueRegistry;
+    private final ExchangeRegistry _exchangeRegistry;
+    private final AMQProtocolSession _protocolSession;
     /**
      * The current state
      */
@@ -58,13 +61,16 @@ public class AMQStateManager implements AMQMethodListener
 
     private CopyOnWriteArraySet<StateListener> _stateListeners = new CopyOnWriteArraySet<StateListener>();
 
-    public AMQStateManager()
+    public AMQStateManager(QueueRegistry queueRegistry, ExchangeRegistry exchangeRegistry, AMQProtocolSession protocolSession)
     {
-        this(AMQState.CONNECTION_NOT_STARTED, true);
+        this(AMQState.CONNECTION_NOT_STARTED, true, queueRegistry, exchangeRegistry, protocolSession);
     }
 
-    protected AMQStateManager(AMQState initial, boolean register)
+    protected AMQStateManager(AMQState initial, boolean register, QueueRegistry queueRegistry, ExchangeRegistry exchangeRegistry, AMQProtocolSession protocolSession)
     {
+        _queueRegistry = queueRegistry;
+        _exchangeRegistry = exchangeRegistry;
+        _protocolSession = protocolSession;
         _currentState = initial;
         if (register)
         {
@@ -158,7 +164,7 @@ public class AMQStateManager implements AMQMethodListener
         }
     }
 
-    public void error(AMQException e)
+    public void error(Exception e)
     {
         _logger.error("State manager received error notification: " + e, e);
         for (StateListener l : _stateListeners)
@@ -167,15 +173,12 @@ public class AMQStateManager implements AMQMethodListener
         }
     }
 
-    public <B extends AMQMethodBody> boolean methodReceived(AMQMethodEvent<B> evt,
-                           AMQProtocolSession protocolSession,
-                           QueueRegistry queueRegistry,
-                           ExchangeRegistry exchangeRegistry) throws AMQException
+    public <B extends AMQMethodBody> boolean methodReceived(AMQMethodEvent<B> evt) throws AMQException
     {
         StateAwareMethodListener<B> handler = findStateTransitionHandler(_currentState, evt.getMethod());
         if (handler != null)
         {
-            handler.methodReceived(this, queueRegistry, exchangeRegistry, protocolSession, evt);
+            handler.methodReceived(this, _queueRegistry, _exchangeRegistry, _protocolSession, evt);
             return true;
         }
         return false;