svn commit: r447994 [33/46] - in /incubator/qpid/trunk/qpid: ./ cpp/ cpp/bin/ cpp/broker/ cpp/broker/inc/ cpp/broker/src/ cpp/broker/test/ cpp/client/ cpp/client/inc/ cpp/client/src/ cpp/client/test/ cpp/common/ cpp/common/concurrent/ cpp/common/concur...

[rh...@apache.org]

Added: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/ServerHandlerRegistry.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/ServerHandlerRegistry.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/ServerHandlerRegistry.java (added)
+++ incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/ServerHandlerRegistry.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,91 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.cluster;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.server.state.AMQState;
+import org.apache.qpid.server.state.AMQStateManager;
+import org.apache.qpid.server.state.IllegalStateTransitionException;
+import org.apache.qpid.server.state.StateAwareMethodListener;
+import org.apache.qpid.server.cluster.util.LogMessage;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * An extension of server.AMQStateManager that allows different handlers to be registered.
+ *
+ */
+class ServerHandlerRegistry extends AMQStateManager
+{
+    private final Logger _logger = Logger.getLogger(ServerHandlerRegistry.class);
+    private final Map<AMQState, MethodHandlerRegistry> _handlers = new HashMap<AMQState, MethodHandlerRegistry>();
+
+    ServerHandlerRegistry()
+    {
+        super(AMQState.CONNECTION_NOT_STARTED, false);
+    }
+
+    ServerHandlerRegistry(ServerHandlerRegistry s)
+    {
+        this();
+        _handlers.putAll(s._handlers);
+    }
+
+    ServerHandlerRegistry(MethodHandlerFactory factory)
+    {
+        this();
+        init(factory);
+    }
+
+    void setHandlers(AMQState state, MethodHandlerRegistry handlers)
+    {
+        _handlers.put(state, handlers);
+    }
+
+    void init(MethodHandlerFactory factory)
+    {
+        for (AMQState s : AMQState.values())
+        {
+            setHandlers(s, factory.register(s, new MethodHandlerRegistry()));
+        }
+    }
+
+    protected <B extends AMQMethodBody> StateAwareMethodListener<B> findStateTransitionHandler(AMQState state, B frame) throws IllegalStateTransitionException
+    {
+        MethodHandlerRegistry registry = _handlers.get(state);
+        StateAwareMethodListener<B> handler = (registry == null) ? null : registry.getHandler(frame);
+        if (handler == null)
+        {
+            _logger.warn(new LogMessage("No handler for {0}, {1}", state, frame));
+        }
+        return handler;
+    }
+
+    <A extends AMQMethodBody, B extends Class<A>> void addHandler(AMQState state, B type, StateAwareMethodListener<A> handler)
+    {
+        MethodHandlerRegistry registry = _handlers.get(state);
+        if (registry == null)
+        {
+            registry = new MethodHandlerRegistry();
+            _handlers.put(state, registry);
+        }
+        registry.addHandler(type, handler);
+    }
+}

Propchange: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/ServerHandlerRegistry.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/SimpleMemberHandle.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/SimpleMemberHandle.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/SimpleMemberHandle.java (added)
+++ incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/SimpleMemberHandle.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,156 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.cluster;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.List;
+
+public class SimpleMemberHandle implements MemberHandle
+{
+    private final String _host;
+    private final int _port;
+
+    public SimpleMemberHandle(String host, int port)
+    {
+        _host = host;
+        _port = port;
+    }
+
+    public SimpleMemberHandle(String details)
+    {
+        String[] parts = details.split(":");
+        _host = parts[0];
+        _port = Integer.parseInt(parts[1]);
+    }
+
+    public SimpleMemberHandle(InetSocketAddress address) throws UnknownHostException
+    {
+        this(address.getAddress(), address.getPort());
+    }
+
+    public SimpleMemberHandle(InetAddress address, int port) throws UnknownHostException
+    {
+        this(canonical(address).getHostAddress(), port);
+    }
+
+    public String getHost()
+    {
+        return _host;
+    }
+
+    public int getPort()
+    {
+        return _port;
+    }
+
+    public int hashCode()
+    {
+        return getPort();
+    }
+
+    public boolean equals(Object o)
+    {
+        return o instanceof MemberHandle && matches((MemberHandle) o);
+    }
+
+    public boolean matches(MemberHandle m)
+    {
+        return matches(m.getHost(), m.getPort());
+    }
+
+    public boolean matches(String host, int port)
+    {
+        return _host.equals(host) && _port == port;
+    }
+
+    public String getDetails()
+    {
+        return _host + ":" + _port;
+    }
+
+    public String toString()
+    {
+        return getDetails();
+    }
+
+    static List<MemberHandle> stringToMembers(String membership)
+    {
+        String[] names = membership.split("\\s");
+        List<MemberHandle> members = new ArrayList<MemberHandle>();
+        for (String name : names)
+        {
+            members.add(new SimpleMemberHandle(name));
+        }
+        return members;
+    }
+
+    static String membersToString(List<MemberHandle> members)
+    {
+        StringBuffer buffer = new StringBuffer();
+        boolean first = true;
+        for (MemberHandle m : members)
+        {
+            if (first)
+            {
+                first = false;
+            }
+            else
+            {
+                buffer.append(" ");
+            }
+            buffer.append(m.getDetails());
+        }
+
+        return buffer.toString();
+    }
+
+    private static InetAddress canonical(InetAddress address) throws UnknownHostException
+    {
+        if (address.isLoopbackAddress())
+        {
+            return InetAddress.getLocalHost();
+        }
+        else
+        {
+            return address;
+        }
+    }
+
+    public MemberHandle resolve()
+    {
+        return resolve(this);
+    }
+
+    public static MemberHandle resolve(MemberHandle handle)
+    {
+        try
+        {
+            return new SimpleMemberHandle(new InetSocketAddress(handle.getHost(), handle.getPort()));
+        }
+        catch (UnknownHostException e)
+        {
+            e.printStackTrace();
+            return handle;
+        }
+    }
+
+
+}

Propchange: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/SimpleMemberHandle.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/SimpleSendable.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/SimpleSendable.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/SimpleSendable.java (added)
+++ incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/SimpleSendable.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,53 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.cluster;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQBody;
+import org.apache.qpid.framing.AMQFrame;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class SimpleSendable implements Sendable
+{
+    private final List<AMQBody> _bodies;
+
+    public SimpleSendable(AMQBody body)
+    {
+        this(Arrays.asList(body));
+    }
+
+    public SimpleSendable(List<AMQBody> bodies)
+    {
+        _bodies = bodies;
+    }
+
+    public void send(int channel, Member member) throws AMQException
+    {
+        for (AMQBody body : _bodies)
+        {
+            member.send(new AMQFrame(channel, body));
+        }
+    }
+
+    public String toString()
+    {
+        return _bodies.toString();
+    }
+}

Propchange: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/SimpleSendable.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ChainedClusterMethodHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ChainedClusterMethodHandler.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ChainedClusterMethodHandler.java (added)
+++ incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ChainedClusterMethodHandler.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,69 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.cluster.handler;
+
+import org.apache.qpid.server.state.AMQStateManager;
+import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.exchange.ExchangeRegistry;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.protocol.AMQMethodEvent;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQMethodBody;
+
+import java.util.List;
+import java.util.ArrayList;
+
+public class ChainedClusterMethodHandler <A extends AMQMethodBody> extends ClusterMethodHandler<A>
+{
+    private final List<ClusterMethodHandler<A>> _handlers;
+
+    private ChainedClusterMethodHandler()
+    {
+        this(new ArrayList<ClusterMethodHandler<A>>());
+    }
+
+    public ChainedClusterMethodHandler(List<ClusterMethodHandler<A>> handlers)
+    {
+        _handlers = handlers;
+    }
+
+    public ChainedClusterMethodHandler(ClusterMethodHandler<A>... handlers)
+    {
+        this();
+        for(ClusterMethodHandler<A>handler: handlers)
+        {
+            _handlers.add(handler);
+        }
+    }
+
+    protected final void peer(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<A> evt) throws AMQException
+    {
+        for(ClusterMethodHandler<A> handler : _handlers)
+        {
+            handler.peer(stateMgr, queues, exchanges, session, evt);
+        }
+    }
+
+    protected final void client(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<A> evt) throws AMQException
+    {
+        for(ClusterMethodHandler<A> handler : _handlers)
+        {
+            handler.client(stateMgr, queues, exchanges, session, evt);
+        }
+    }
+}

Propchange: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ChainedClusterMethodHandler.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ChannelQueueManager.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ChannelQueueManager.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ChannelQueueManager.java (added)
+++ incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ChannelQueueManager.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,136 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.cluster.handler;
+
+import org.apache.qpid.server.state.StateAwareMethodListener;
+import org.apache.qpid.server.state.AMQStateManager;
+import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.exchange.ExchangeRegistry;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.protocol.AMQMethodEvent;
+import org.apache.qpid.server.cluster.util.LogMessage;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.framing.QueueDeclareBody;
+import org.apache.qpid.framing.QueueBindBody;
+import org.apache.qpid.framing.QueueDeleteBody;
+import org.apache.qpid.framing.BasicConsumeBody;
+import org.apache.log4j.Logger;
+
+import java.util.Map;
+import java.util.HashMap;
+
+/**
+ * Maintains the default queue names for a channel, and alters subsequent frames where necessary
+ * to use this (i.e. when no queue is explictly specified).
+ *
+ */
+class ChannelQueueManager
+{
+    private static final Logger _logger = Logger.getLogger(ChannelQueueManager.class);
+    private final Map<Integer, String> _channelQueues = new HashMap<Integer, String>();
+
+    ClusterMethodHandler<QueueDeclareBody> createQueueDeclareHandler()
+    {
+        return new QueueDeclareHandler();
+    }
+
+    ClusterMethodHandler<QueueDeleteBody> createQueueDeleteHandler()
+    {
+        return new QueueDeleteHandler();
+    }
+
+    ClusterMethodHandler<QueueBindBody> createQueueBindHandler()
+    {
+        return new QueueBindHandler();
+    }
+
+    ClusterMethodHandler<BasicConsumeBody> createBasicConsumeHandler()
+    {
+        return new BasicConsumeHandler();
+    }
+
+    private void set(int channel, String queue)
+    {
+        _channelQueues.put(channel, queue);
+        _logger.info(new LogMessage("Set default queue for {0} to {1}", channel, queue));
+    }
+
+    private String get(int channel)
+    {
+        String queue = _channelQueues.get(channel);
+        _logger.info(new LogMessage("Default queue for {0} is {1}", channel, queue));
+        return queue;
+    }
+
+    private class QueueDeclareHandler extends ClusterMethodHandler<QueueDeclareBody>
+    {
+        protected void peer(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<QueueDeclareBody> evt) throws AMQException
+        {
+        }
+
+        protected void client(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<QueueDeclareBody> evt) throws AMQException
+        {
+            set(evt.getChannelId(), evt.getMethod().queue);
+        }
+    }
+    private class QueueBindHandler extends ClusterMethodHandler<QueueBindBody>
+    {
+        protected void peer(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<QueueBindBody> evt) throws AMQException
+        {
+        }
+
+        protected void client(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<QueueBindBody> evt) throws AMQException
+        {
+            if(evt.getMethod().queue == null)
+            {
+                evt.getMethod().queue = get(evt.getChannelId());
+            }
+        }
+    }
+    private class QueueDeleteHandler extends ClusterMethodHandler<QueueDeleteBody>
+    {
+        protected void peer(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<QueueDeleteBody> evt) throws AMQException
+        {
+        }
+
+        protected void client(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<QueueDeleteBody> evt) throws AMQException
+        {
+            if(evt.getMethod().queue == null)
+            {
+                evt.getMethod().queue = get(evt.getChannelId());
+            }
+        }
+    }
+
+    private class BasicConsumeHandler extends ClusterMethodHandler<BasicConsumeBody>
+    {
+        protected void peer(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<BasicConsumeBody> evt) throws AMQException
+        {
+        }
+
+        protected void client(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<BasicConsumeBody> evt) throws AMQException
+        {
+            if(evt.getMethod().queue == null)
+            {
+                evt.getMethod().queue = get(evt.getChannelId());
+            }
+        }
+    }
+
+}

Propchange: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ChannelQueueManager.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ClusterMethodHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ClusterMethodHandler.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ClusterMethodHandler.java (added)
+++ incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ClusterMethodHandler.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,46 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.cluster.handler;
+
+import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.server.state.StateAwareMethodListener;
+import org.apache.qpid.server.state.AMQStateManager;
+import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.exchange.ExchangeRegistry;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.protocol.AMQMethodEvent;
+import org.apache.qpid.server.cluster.ClusteredProtocolSession;
+import org.apache.qpid.AMQException;
+
+public abstract class ClusterMethodHandler<A extends AMQMethodBody> implements StateAwareMethodListener<A>
+{
+    public final void methodReceived(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<A> evt) throws AMQException
+    {
+        if (ClusteredProtocolSession.isPeerSession(session))
+        {
+            peer(stateMgr, queues, exchanges, session, evt);
+        }
+        else
+        {
+            client(stateMgr, queues, exchanges, session, evt);
+        }
+    }
+
+    protected abstract void peer(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<A> evt) throws AMQException;
+    protected abstract void client(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<A> evt) throws AMQException;
+}

Propchange: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ClusterMethodHandler.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ClusterMethodHandlerFactory.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ClusterMethodHandlerFactory.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ClusterMethodHandlerFactory.java (added)
+++ incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ClusterMethodHandlerFactory.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,278 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.cluster.handler;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQFrame;
+import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.framing.BasicCancelBody;
+import org.apache.qpid.framing.BasicConsumeBody;
+import org.apache.qpid.framing.BasicPublishBody;
+import org.apache.qpid.framing.ChannelCloseBody;
+import org.apache.qpid.framing.ChannelFlowBody;
+import org.apache.qpid.framing.ChannelOpenBody;
+import org.apache.qpid.framing.ClusterJoinBody;
+import org.apache.qpid.framing.ClusterLeaveBody;
+import org.apache.qpid.framing.ClusterMembershipBody;
+import org.apache.qpid.framing.ClusterPingBody;
+import org.apache.qpid.framing.ClusterSuspectBody;
+import org.apache.qpid.framing.ConnectionCloseBody;
+import org.apache.qpid.framing.ConnectionOpenBody;
+import org.apache.qpid.framing.ConnectionSecureOkBody;
+import org.apache.qpid.framing.ConnectionStartOkBody;
+import org.apache.qpid.framing.ConnectionTuneOkBody;
+import org.apache.qpid.framing.ExchangeDeclareBody;
+import org.apache.qpid.framing.ExchangeDeleteBody;
+import org.apache.qpid.framing.QueueBindBody;
+import org.apache.qpid.framing.QueueDeclareBody;
+import org.apache.qpid.framing.QueueDeleteBody;
+import org.apache.qpid.framing.ClusterSynchBody;
+import org.apache.qpid.framing.BasicQosBody;
+import org.apache.qpid.framing.TxSelectBody;
+import org.apache.qpid.framing.TxCommitBody;
+import org.apache.qpid.framing.TxRollbackBody;
+import org.apache.qpid.server.cluster.ClusterCapability;
+import org.apache.qpid.server.cluster.ClusteredProtocolSession;
+import org.apache.qpid.server.cluster.GroupManager;
+import org.apache.qpid.server.cluster.LoadTable;
+import org.apache.qpid.server.cluster.MemberHandle;
+import org.apache.qpid.server.cluster.MethodHandlerFactory;
+import org.apache.qpid.server.cluster.MethodHandlerRegistry;
+import org.apache.qpid.server.cluster.SimpleMemberHandle;
+import org.apache.qpid.server.exchange.ExchangeRegistry;
+import org.apache.qpid.server.handler.ChannelCloseHandler;
+import org.apache.qpid.server.handler.ChannelFlowHandler;
+import org.apache.qpid.server.handler.ChannelOpenHandler;
+import org.apache.qpid.server.handler.ConnectionCloseMethodHandler;
+import org.apache.qpid.server.handler.ConnectionOpenMethodHandler;
+import org.apache.qpid.server.handler.ConnectionSecureOkMethodHandler;
+import org.apache.qpid.server.handler.ConnectionStartOkMethodHandler;
+import org.apache.qpid.server.handler.ConnectionTuneOkMethodHandler;
+import org.apache.qpid.server.handler.ExchangeDeclareHandler;
+import org.apache.qpid.server.handler.ExchangeDeleteHandler;
+import org.apache.qpid.server.handler.BasicCancelMethodHandler;
+import org.apache.qpid.server.handler.BasicPublishMethodHandler;
+import org.apache.qpid.server.handler.QueueBindHandler;
+import org.apache.qpid.server.handler.QueueDeleteHandler;
+import org.apache.qpid.server.handler.BasicQosHandler;
+import org.apache.qpid.server.handler.TxSelectHandler;
+import org.apache.qpid.server.handler.TxCommitHandler;
+import org.apache.qpid.server.handler.TxRollbackHandler;
+import org.apache.qpid.server.protocol.AMQMethodEvent;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.state.AMQState;
+import org.apache.qpid.server.state.AMQStateManager;
+import org.apache.qpid.server.state.StateAwareMethodListener;
+
+public class ClusterMethodHandlerFactory implements MethodHandlerFactory
+{
+    private final GroupManager _groupMgr;
+    private final LoadTable _loadTable;
+
+    public ClusterMethodHandlerFactory(GroupManager groupMgr, LoadTable loadTable)
+    {
+        _groupMgr = groupMgr;
+        _loadTable = loadTable;
+    }
+
+    public MethodHandlerRegistry register(AMQState state, MethodHandlerRegistry registry)
+    {
+        switch (state)
+        {
+            case CONNECTION_NOT_STARTED:
+                return registry.addHandler(ConnectionStartOkBody.class, ConnectionStartOkMethodHandler.getInstance());
+            case CONNECTION_NOT_AUTH:
+                return registry.addHandler(ConnectionSecureOkBody.class, ConnectionSecureOkMethodHandler.getInstance());
+            case CONNECTION_NOT_TUNED:
+                return registry.addHandler(ConnectionTuneOkBody.class, ConnectionTuneOkMethodHandler.getInstance());
+            case CONNECTION_NOT_OPENED:
+                //connection.open override:
+                return registry.addHandler(ConnectionOpenBody.class, new ConnectionOpenHandler());
+            case CONNECTION_OPEN:
+                return registerConnectionOpened(registry);
+        }
+        return registry;
+    }
+
+    private MethodHandlerRegistry registerConnectionOpened(MethodHandlerRegistry registry)
+    {
+        //new cluster method handlers:
+        registry.addHandler(ClusterJoinBody.class, new JoinHandler());
+        registry.addHandler(ClusterLeaveBody.class, new LeaveHandler());
+        registry.addHandler(ClusterSuspectBody.class, new SuspectHandler());
+        registry.addHandler(ClusterMembershipBody.class, new MembershipHandler());
+        registry.addHandler(ClusterPingBody.class, new PingHandler());
+        registry.addHandler(ClusterSynchBody.class, new SynchHandler());
+
+        //connection.close override:
+        registry.addHandler(ConnectionCloseBody.class, new ConnectionCloseHandler());
+
+        //replicated handlers:
+        registry.addHandler(ExchangeDeclareBody.class, replicated(ExchangeDeclareHandler.getInstance()));
+        registry.addHandler(ExchangeDeleteBody.class, replicated(ExchangeDeleteHandler.getInstance()));
+
+        ChannelQueueManager channelQueueMgr = new ChannelQueueManager();
+
+
+        LocalQueueDeclareHandler handler = new LocalQueueDeclareHandler(_groupMgr);
+        registry.addHandler(QueueDeclareBody.class,
+                            chain(new QueueNameGenerator(handler),
+                                  channelQueueMgr.createQueueDeclareHandler(),
+                                  new ReplicatingHandler<QueueDeclareBody>(_groupMgr, handler)));
+
+        registry.addHandler(QueueBindBody.class, chain(channelQueueMgr.createQueueBindHandler(), replicated(QueueBindHandler.getInstance())));
+        registry.addHandler(QueueDeleteBody.class, chain(channelQueueMgr.createQueueDeleteHandler(), replicated(alternate(new QueueDeleteHandler(false), new QueueDeleteHandler(true)))));
+        registry.addHandler(BasicConsumeBody.class, chain(channelQueueMgr.createBasicConsumeHandler(), new ReplicatingConsumeHandler(_groupMgr)));
+
+        //other modified handlers:
+        registry.addHandler(BasicCancelBody.class, alternate(new RemoteCancelHandler(), BasicCancelMethodHandler.getInstance()));
+
+        //other unaffected handlers:
+        registry.addHandler(BasicPublishBody.class, BasicPublishMethodHandler.getInstance());
+        registry.addHandler(BasicQosBody.class, BasicQosHandler.getInstance());
+        registry.addHandler(ChannelOpenBody.class, ChannelOpenHandler.getInstance());
+        registry.addHandler(ChannelCloseBody.class, ChannelCloseHandler.getInstance());
+        registry.addHandler(ChannelFlowBody.class, ChannelFlowHandler.getInstance());
+        registry.addHandler(TxSelectBody.class, TxSelectHandler.getInstance());
+        registry.addHandler(TxCommitBody.class, TxCommitHandler.getInstance());
+        registry.addHandler(TxRollbackBody.class, TxRollbackHandler.getInstance());
+
+
+        return registry;
+    }
+
+    private class SynchHandler implements StateAwareMethodListener<ClusterSynchBody>
+    {
+        public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
+                                   ExchangeRegistry exchangeRegistry, AMQProtocolSession session,
+                                   AMQMethodEvent<ClusterSynchBody> evt) throws AMQException
+        {
+            _groupMgr.handleSynch(ClusteredProtocolSession.getSessionPeer(session));
+        }
+    }
+
+    private class JoinHandler implements StateAwareMethodListener<ClusterJoinBody>
+    {
+        public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
+                                   ExchangeRegistry exchangeRegistry, AMQProtocolSession session,
+                                   AMQMethodEvent<ClusterJoinBody> evt) throws AMQException
+        {
+            _groupMgr.handleJoin(new SimpleMemberHandle(evt.getMethod().broker));
+        }
+    }
+
+    private class LeaveHandler implements StateAwareMethodListener<ClusterLeaveBody>
+    {
+        public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
+                                   ExchangeRegistry exchangeRegistry, AMQProtocolSession protocolSession,
+                                   AMQMethodEvent<ClusterLeaveBody> evt) throws AMQException
+        {
+            _groupMgr.handleLeave(new SimpleMemberHandle(evt.getMethod().broker));
+        }
+    }
+
+    private class SuspectHandler implements StateAwareMethodListener<ClusterSuspectBody>
+    {
+        public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
+                                   ExchangeRegistry exchangeRegistry, AMQProtocolSession protocolSession,
+                                   AMQMethodEvent<ClusterSuspectBody> evt) throws AMQException
+        {
+            _groupMgr.handleSuspect(new SimpleMemberHandle(evt.getMethod().broker));
+        }
+    }
+
+    private class MembershipHandler implements StateAwareMethodListener<ClusterMembershipBody>
+    {
+        public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
+                                   ExchangeRegistry exchangeRegistry, AMQProtocolSession session,
+                                   AMQMethodEvent<ClusterMembershipBody> evt) throws AMQException
+        {
+            ClusterMembershipBody body = evt.getMethod();
+            _groupMgr.handleMembershipAnnouncement(new String(body.members));
+        }
+    }
+
+    private class PingHandler implements StateAwareMethodListener<ClusterPingBody>
+    {
+        public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry,
+                                   ExchangeRegistry exchangeRegistry, AMQProtocolSession session,
+                                   AMQMethodEvent<ClusterPingBody> evt) throws AMQException
+        {
+            MemberHandle peer = new SimpleMemberHandle(evt.getMethod().broker);
+            _groupMgr.handlePing(peer, evt.getMethod().load);
+            if (evt.getMethod().responseRequired)
+            {
+                evt.getMethod().load = _loadTable.getLocalLoad();
+                session.writeFrame(new AMQFrame(evt.getChannelId(), evt.getMethod()));
+            }
+        }
+    }
+
+    private class ConnectionOpenHandler extends ExtendedHandler<ConnectionOpenBody>
+    {
+        ConnectionOpenHandler()
+        {
+            super(ConnectionOpenMethodHandler.getInstance());
+        }
+
+        void postHandle(AMQStateManager stateMgr, AMQProtocolSession session, AMQMethodEvent<ConnectionOpenBody> evt)
+        {
+            String capabilities = evt.getMethod().capabilities;
+            if (ClusterCapability.contains(capabilities))
+            {
+                ClusteredProtocolSession.setSessionPeer(session, ClusterCapability.getPeer(capabilities));
+            }
+            else
+            {
+                _loadTable.incrementLocalLoad();
+            }
+        }
+    }
+
+    private class ConnectionCloseHandler extends ExtendedHandler<ConnectionCloseBody>
+    {
+        ConnectionCloseHandler()
+        {
+            super(ConnectionCloseMethodHandler.getInstance());
+        }
+
+        void postHandle(AMQStateManager stateMgr, AMQProtocolSession session, AMQMethodEvent<ConnectionCloseBody> evt)
+        {
+            if (!ClusteredProtocolSession.isPeerSession(session))
+            {
+                _loadTable.decrementLocalLoad();
+            }
+        }
+    }
+
+    private <B extends AMQMethodBody> ReplicatingHandler<B> replicated(StateAwareMethodListener<B> handler)
+    {
+        return new ReplicatingHandler<B>(_groupMgr, handler);
+    }
+
+    private <B extends AMQMethodBody> StateAwareMethodListener<B> alternate(StateAwareMethodListener<B> peer, StateAwareMethodListener<B> client)
+    {
+        return new PeerHandler<B>(peer, client);
+    }
+
+    private <B extends AMQMethodBody> StateAwareMethodListener<B> chain(ClusterMethodHandler<B>... h)
+    {
+        return new ChainedClusterMethodHandler<B>(h);
+    }
+}

Propchange: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ClusterMethodHandlerFactory.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ExtendedHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ExtendedHandler.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ExtendedHandler.java (added)
+++ incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ExtendedHandler.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,52 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.cluster.handler;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.server.exchange.ExchangeRegistry;
+import org.apache.qpid.server.protocol.AMQMethodEvent;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.state.AMQStateManager;
+import org.apache.qpid.server.state.StateAwareMethodListener;
+
+class ExtendedHandler<A extends AMQMethodBody> implements StateAwareMethodListener<A>
+{
+    private final StateAwareMethodListener<A> _base;
+
+    ExtendedHandler(StateAwareMethodListener<A> base)
+    {
+        _base = base;
+    }
+
+    public void methodReceived(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<A> evt) throws AMQException
+    {
+        preHandle(stateMgr, session, evt);
+        _base.methodReceived(stateMgr, queues, exchanges, session, evt);
+        postHandle(stateMgr, session, evt);
+    }
+
+    void preHandle(AMQStateManager stateMgr, AMQProtocolSession session, AMQMethodEvent<A> evt) throws AMQException
+    {
+    }
+
+    void postHandle(AMQStateManager stateMgr, AMQProtocolSession session, AMQMethodEvent<A> evt) throws AMQException
+    {
+    }
+}

Propchange: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ExtendedHandler.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/HandlerUtils.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/HandlerUtils.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/HandlerUtils.java (added)
+++ incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/HandlerUtils.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,22 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.cluster.handler;
+
+public abstract class HandlerUtils
+{
+}

Propchange: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/HandlerUtils.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/LocalQueueDeclareHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/LocalQueueDeclareHandler.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/LocalQueueDeclareHandler.java (added)
+++ incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/LocalQueueDeclareHandler.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,74 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.cluster.handler;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.QueueDeclareBody;
+import org.apache.qpid.server.cluster.ClusteredProtocolSession;
+import org.apache.qpid.server.cluster.GroupManager;
+import org.apache.qpid.server.cluster.util.LogMessage;
+import org.apache.qpid.server.cluster.MemberHandle;
+import org.apache.qpid.server.handler.QueueDeclareHandler;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.ClusteredQueue;
+import org.apache.qpid.server.queue.PrivateQueue;
+import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.queue.RemoteQueueProxy;
+
+public class LocalQueueDeclareHandler extends QueueDeclareHandler
+{
+    private static final Logger _logger = Logger.getLogger(LocalQueueDeclareHandler.class);
+    private final GroupManager _groupMgr;
+
+    LocalQueueDeclareHandler(GroupManager groupMgr)
+    {
+        _groupMgr = groupMgr;
+    }
+
+    protected String createName()
+    {
+        return super.createName() + "@" + _groupMgr.getLocal().getDetails();
+    }
+
+    protected AMQQueue createQueue(QueueDeclareBody body, QueueRegistry registry, AMQProtocolSession session) throws AMQException
+    {
+        //is it private or shared:
+        if (body.exclusive)
+        {
+            if (ClusteredProtocolSession.isPeerSession(session))
+            {
+                //need to get peer from the session...
+                MemberHandle peer = ClusteredProtocolSession.getSessionPeer(session);
+                _logger.debug(new LogMessage("Creating proxied queue {0} on behalf of {1}", body.queue, peer));
+                return new RemoteQueueProxy(peer, _groupMgr, body.queue, body.durable, peer.getDetails(), body.autoDelete, registry);
+            }
+            else
+            {
+                _logger.debug(new LogMessage("Creating local private queue {0}", body.queue));
+                return new PrivateQueue(_groupMgr, body.queue, body.durable, session.getContextKey(), body.autoDelete, registry);
+            }
+        }
+        else
+        {
+            _logger.debug(new LogMessage("Creating local shared queue {0}", body.queue));
+            return new ClusteredQueue(_groupMgr, body.queue, body.durable, null, body.autoDelete, registry);
+        }
+    }
+}

Propchange: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/LocalQueueDeclareHandler.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/NullListener.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/NullListener.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/NullListener.java (added)
+++ incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/NullListener.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,35 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.cluster.handler;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.server.exchange.ExchangeRegistry;
+import org.apache.qpid.server.protocol.AMQMethodEvent;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.state.AMQStateManager;
+import org.apache.qpid.server.state.StateAwareMethodListener;
+
+public class NullListener<T extends AMQMethodBody> implements StateAwareMethodListener<T>
+{
+    public void methodReceived(AMQStateManager stateManager, QueueRegistry queueRegistry, ExchangeRegistry exchangeRegistry, AMQProtocolSession protocolSession, AMQMethodEvent<T> evt) throws AMQException
+    {
+    }
+}
+

Propchange: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/NullListener.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/PeerHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/PeerHandler.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/PeerHandler.java (added)
+++ incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/PeerHandler.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,57 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.cluster.handler;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.server.cluster.ClusteredProtocolSession;
+import org.apache.qpid.server.exchange.ExchangeRegistry;
+import org.apache.qpid.server.protocol.AMQMethodEvent;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.state.AMQStateManager;
+import org.apache.qpid.server.state.StateAwareMethodListener;
+
+/**
+ * Base for implementing handlers that carry out different actions based on whether the method they
+ * are handling was sent by a peer (i.e. another broker in the cluster) or a client (i.e. an end-user
+ * application).
+ *
+ */
+public class PeerHandler<A extends AMQMethodBody> extends ClusterMethodHandler<A>
+{
+    private final StateAwareMethodListener<A> _peer;
+    private final StateAwareMethodListener<A> _client;
+
+    PeerHandler(StateAwareMethodListener<A> peer, StateAwareMethodListener<A> client)
+    {
+        _peer = peer;
+        _client = client;
+    }
+
+    protected void peer(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<A> evt) throws AMQException
+    {
+        _peer.methodReceived(stateMgr, queues, exchanges, session, evt);
+    }
+
+    protected void client(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<A> evt) throws AMQException
+    {
+        _client.methodReceived(stateMgr, queues, exchanges, session, evt);
+    }
+
+}

Propchange: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/PeerHandler.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/QueueNameGenerator.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/QueueNameGenerator.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/QueueNameGenerator.java (added)
+++ incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/QueueNameGenerator.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,60 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.cluster.handler;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.QueueDeclareBody;
+import org.apache.qpid.server.exchange.ExchangeRegistry;
+import org.apache.qpid.server.protocol.AMQMethodEvent;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.state.AMQStateManager;
+
+/**
+ * Generates queue names for queues declared with no name.
+ *
+ */
+class QueueNameGenerator extends ClusterMethodHandler<QueueDeclareBody>
+{
+    private final LocalQueueDeclareHandler _handler;
+
+    QueueNameGenerator(LocalQueueDeclareHandler handler)
+    {
+        _handler = handler;
+    }
+
+    protected void peer(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<QueueDeclareBody> evt) throws AMQException
+    {
+    }
+
+    protected void client(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges,
+                          AMQProtocolSession session, AMQMethodEvent<QueueDeclareBody> evt)
+            throws AMQException
+    {
+        setName(evt.getMethod());//need to set the name before propagating this method
+    }
+
+    protected void setName(QueueDeclareBody body)
+    {
+        if (body.queue == null)
+        {
+            body.queue = _handler.createName();
+        }
+    }
+}
+

Propchange: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/QueueNameGenerator.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/RemoteCancelHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/RemoteCancelHandler.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/RemoteCancelHandler.java (added)
+++ incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/RemoteCancelHandler.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,50 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.cluster.handler;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.BasicCancelBody;
+import org.apache.qpid.server.cluster.ClusteredProtocolSession;
+import org.apache.qpid.server.exchange.ExchangeRegistry;
+import org.apache.qpid.server.protocol.AMQMethodEvent;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.ClusteredQueue;
+import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.state.AMQStateManager;
+import org.apache.qpid.server.state.StateAwareMethodListener;
+
+public class RemoteCancelHandler implements StateAwareMethodListener<BasicCancelBody>
+{
+    private final Logger _logger = Logger.getLogger(RemoteCancelHandler.class);
+
+    public void methodReceived(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<BasicCancelBody> evt) throws AMQException
+    {
+        //By convention, consumers setup between brokers use the queue name as the consumer tag:
+        AMQQueue queue = queues.getQueue(evt.getMethod().consumerTag);
+        if (queue instanceof ClusteredQueue)
+        {
+            ((ClusteredQueue) queue).removeRemoteSubscriber(ClusteredProtocolSession.getSessionPeer(session));
+        }
+        else
+        {
+            _logger.warn("Got remote cancel request for non-clustered queue: " + queue);
+        }
+    }
+}

Propchange: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/RemoteCancelHandler.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/RemoteConsumeHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/RemoteConsumeHandler.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/RemoteConsumeHandler.java (added)
+++ incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/RemoteConsumeHandler.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,55 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.cluster.handler;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.BasicConsumeBody;
+import org.apache.qpid.framing.BasicConsumeOkBody;
+import org.apache.qpid.server.cluster.ClusteredProtocolSession;
+import org.apache.qpid.server.exchange.ExchangeRegistry;
+import org.apache.qpid.server.protocol.AMQMethodEvent;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.ClusteredQueue;
+import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.state.AMQStateManager;
+import org.apache.qpid.server.state.StateAwareMethodListener;
+
+/**
+ * Handles consume requests from other cluster members.
+ *
+ */
+public class RemoteConsumeHandler implements StateAwareMethodListener<BasicConsumeBody>
+{
+    private final Logger _logger = Logger.getLogger(RemoteConsumeHandler.class);
+
+    public void methodReceived(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<BasicConsumeBody> evt) throws AMQException
+    {
+        AMQQueue queue = queues.getQueue(evt.getMethod().queue);
+        if (queue instanceof ClusteredQueue)
+        {
+            ((ClusteredQueue) queue).addRemoteSubcriber(ClusteredProtocolSession.getSessionPeer(session));
+            session.writeFrame(BasicConsumeOkBody.createAMQFrame(evt.getChannelId(), evt.getMethod().queue));
+        }
+        else
+        {
+            _logger.warn("Got remote consume request for non-clustered queue: " + queue);
+        }
+    }
+}

Propchange: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/RemoteConsumeHandler.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ReplicatingConsumeHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ReplicatingConsumeHandler.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ReplicatingConsumeHandler.java (added)
+++ incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ReplicatingConsumeHandler.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,81 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.cluster.handler;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.BasicConsumeBody;
+import org.apache.qpid.server.cluster.BroadcastPolicy;
+import org.apache.qpid.server.cluster.GroupManager;
+import org.apache.qpid.server.cluster.util.LogMessage;
+import org.apache.qpid.server.exchange.ExchangeRegistry;
+import org.apache.qpid.server.handler.BasicConsumeMethodHandler;
+import org.apache.qpid.server.protocol.AMQMethodEvent;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.state.AMQStateManager;
+import org.apache.qpid.server.state.StateAwareMethodListener;
+
+public class ReplicatingConsumeHandler extends ReplicatingHandler<BasicConsumeBody>
+{
+    ReplicatingConsumeHandler(GroupManager groupMgr)
+    {
+        this(groupMgr, null);
+    }
+
+    ReplicatingConsumeHandler(GroupManager groupMgr, BroadcastPolicy policy)
+    {
+        super(groupMgr, base(), policy);
+    }
+
+    protected void replicate(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<BasicConsumeBody> evt) throws AMQException
+    {
+        //only replicate if the queue in question is a shared queue
+        if (isShared(queues.getQueue(evt.getMethod().queue)))
+        {
+            super.replicate(stateMgr, queues, exchanges, session, evt);
+        }
+        else
+        {
+            _logger.info(new LogMessage("Handling consume for private queue ({0}) locally", evt.getMethod()));
+            local(stateMgr, queues, exchanges, session, evt);
+            _logger.info(new LogMessage("Handled consume for private queue ({0}) locally", evt.getMethod()));
+
+        }
+    }
+
+    protected boolean isShared(AMQQueue queue)
+    {
+        return queue != null && queue.isShared();
+    }
+
+    static StateAwareMethodListener<BasicConsumeBody> base()
+    {
+        return new PeerHandler<BasicConsumeBody>(peer(), client());
+    }
+
+    static StateAwareMethodListener<BasicConsumeBody> peer()
+    {
+        return new RemoteConsumeHandler();
+    }
+
+    static StateAwareMethodListener<BasicConsumeBody> client()
+    {
+        return BasicConsumeMethodHandler.getInstance();
+    }
+}

Propchange: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ReplicatingConsumeHandler.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ReplicatingHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ReplicatingHandler.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ReplicatingHandler.java (added)
+++ incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ReplicatingHandler.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,127 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.cluster.handler;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.server.cluster.BroadcastPolicy;
+import org.apache.qpid.server.cluster.ClusteredProtocolSession;
+import org.apache.qpid.server.cluster.GroupManager;
+import org.apache.qpid.server.cluster.GroupResponseHandler;
+import org.apache.qpid.server.cluster.util.LogMessage;
+import org.apache.qpid.server.cluster.Member;
+import org.apache.qpid.server.cluster.SimpleSendable;
+import org.apache.qpid.server.cluster.policy.StandardPolicies;
+import org.apache.qpid.server.exchange.ExchangeRegistry;
+import org.apache.qpid.server.protocol.AMQMethodEvent;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.state.AMQStateManager;
+import org.apache.qpid.server.state.StateAwareMethodListener;
+
+import java.util.List;
+
+/**
+ * Basic template for handling methods that should be broadcast to the group and
+ * processed locally after 'completion' of this broadcast.
+ *
+ */
+class ReplicatingHandler<A extends AMQMethodBody> extends ClusterMethodHandler<A> implements StandardPolicies
+{
+    protected static final Logger _logger = Logger.getLogger(ReplicatingHandler.class);
+
+    private final StateAwareMethodListener<A> _base;
+    private final GroupManager _groupMgr;
+    private final BroadcastPolicy _policy;
+
+    ReplicatingHandler(GroupManager groupMgr, StateAwareMethodListener<A> base)
+    {
+        this(groupMgr, base, null);
+    }
+
+    ReplicatingHandler(GroupManager groupMgr, StateAwareMethodListener<A> base, BroadcastPolicy policy)
+    {
+        _groupMgr = groupMgr;
+        _base = base;
+        _policy = policy;
+    }
+
+    protected void peer(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<A> evt) throws AMQException
+    {
+        local(stateMgr, queues, exchanges, session, evt);
+        _logger.debug(new LogMessage("Handled {0} locally", evt.getMethod()));
+    }
+
+    protected void client(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<A> evt) throws AMQException
+    {
+        replicate(stateMgr, queues, exchanges, session, evt);
+    }
+
+    protected void replicate(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<A> evt) throws AMQException
+    {
+        if (_policy == null)
+        {
+            //asynch delivery
+            _groupMgr.broadcast(new SimpleSendable(evt.getMethod()));
+            local(stateMgr, queues, exchanges, session, evt);
+        }
+        else
+        {
+            Callback callback = new Callback(stateMgr, queues, exchanges, session, evt);
+            _groupMgr.broadcast(new SimpleSendable(evt.getMethod()), _policy, callback);
+        }
+        _logger.debug(new LogMessage("Replicated {0} to peers", evt.getMethod()));
+    }
+
+    protected void local(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<A> evt) throws AMQException
+    {
+        _base.methodReceived(stateMgr, queues, exchanges, session, evt);
+    }
+
+    private class Callback implements GroupResponseHandler
+    {
+        private final AMQStateManager _stateMgr;
+        private final QueueRegistry _queues;
+        private final ExchangeRegistry _exchanges;
+        private final AMQProtocolSession _session;
+        private final AMQMethodEvent<A> _evt;
+
+        Callback(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<A> evt)
+        {
+            _stateMgr = stateMgr;
+            _queues = queues;
+            _exchanges = exchanges;
+            _session = session;
+            _evt = evt;
+        }
+
+        public void response(List<AMQMethodBody> responses, List<Member> members)
+        {
+            try
+            {
+                local(_stateMgr, _queues, _exchanges, _session, _evt);
+                _logger.debug(new LogMessage("Handled {0} locally, in response to completion of replication", _evt.getMethod()));
+            }
+            catch (AMQException e)
+            {
+                _logger.error(new LogMessage("Error handling {0}:{1}", _evt.getMethod(), e), e);
+            }
+        }
+    }
+}

Propchange: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/ReplicatingHandler.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/WrappedListener.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/WrappedListener.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/WrappedListener.java (added)
+++ incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/WrappedListener.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,53 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.cluster.handler;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.server.exchange.ExchangeRegistry;
+import org.apache.qpid.server.protocol.AMQMethodEvent;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.state.AMQStateManager;
+import org.apache.qpid.server.state.StateAwareMethodListener;
+
+public class WrappedListener<T extends AMQMethodBody> implements StateAwareMethodListener<T>
+{
+    private final StateAwareMethodListener<T> _primary;
+    private final StateAwareMethodListener _post;
+    private final StateAwareMethodListener _pre;
+
+    WrappedListener(StateAwareMethodListener<T> primary, StateAwareMethodListener  pre, StateAwareMethodListener post)
+    {
+        _pre = check(pre);
+        _post = check(post);
+        _primary = check(primary);
+    }
+
+    public void methodReceived(AMQStateManager stateMgr, QueueRegistry queues, ExchangeRegistry exchanges, AMQProtocolSession session, AMQMethodEvent<T> evt) throws AMQException
+    {
+        _pre.methodReceived(stateMgr, queues, exchanges, session, evt);
+        _primary.methodReceived(stateMgr, queues, exchanges, session, evt);
+        _post.methodReceived(stateMgr, queues, exchanges, session, evt);
+    }
+
+    private static <T extends AMQMethodBody> StateAwareMethodListener<T> check(StateAwareMethodListener<T> in)
+    {
+        return in == null ? new NullListener<T>() : in;
+    }
+}

Propchange: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/WrappedListener.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/WrappingMethodHandlerFactory.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/WrappingMethodHandlerFactory.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/WrappingMethodHandlerFactory.java (added)
+++ incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/WrappingMethodHandlerFactory.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,82 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.cluster.handler;
+
+import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.server.cluster.MethodHandlerFactory;
+import org.apache.qpid.server.cluster.MethodHandlerRegistry;
+import org.apache.qpid.server.state.AMQState;
+import org.apache.qpid.server.state.StateAwareMethodListener;
+
+public abstract class WrappingMethodHandlerFactory implements MethodHandlerFactory
+{
+    private final MethodHandlerFactory _delegate;
+    private final StateAwareMethodListener _pre;
+    private final StateAwareMethodListener _post;
+
+    protected WrappingMethodHandlerFactory(MethodHandlerFactory delegate,
+                                           StateAwareMethodListener pre,
+                                           StateAwareMethodListener post)
+    {
+        _delegate = delegate;
+        _pre = pre;
+        _post = post;
+    }
+
+    public MethodHandlerRegistry register(AMQState state, MethodHandlerRegistry registry)
+    {
+        if (isWrappableState(state))
+        {
+            return wrap(_delegate.register(state, registry), state);
+        }
+        else
+        {
+            return _delegate.register(state, registry);
+        }
+    }
+
+    protected abstract boolean isWrappableState(AMQState state);
+
+    protected abstract Iterable<FrameDescriptor> getWrappableFrameTypes(AMQState state);
+
+    private MethodHandlerRegistry wrap(MethodHandlerRegistry registry, AMQState state)
+    {
+        for (FrameDescriptor fd : getWrappableFrameTypes(state))
+        {
+            wrap(registry, fd.type, fd.instance);
+        }
+        return registry;
+    }
+
+    private <A extends AMQMethodBody, B extends Class<A>> void wrap(MethodHandlerRegistry r, B type, A frame)
+    {
+        r.addHandler(type, new WrappedListener<A>(r.getHandler(frame), _pre, _post));
+    }
+
+    protected static class FrameDescriptor<A extends AMQMethodBody, B extends Class<A>>
+    {
+        protected final A instance;
+        protected final B type;
+
+        public FrameDescriptor(B type, A instance)
+        {
+            this.instance = instance;
+            this.type = type;
+        }
+    }
+}

Propchange: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/handler/WrappingMethodHandlerFactory.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/policy/AsynchBroadcastPolicy.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/policy/AsynchBroadcastPolicy.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/policy/AsynchBroadcastPolicy.java (added)
+++ incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/policy/AsynchBroadcastPolicy.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,28 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.cluster.policy;
+
+import org.apache.qpid.server.cluster.BroadcastPolicy;
+
+public class AsynchBroadcastPolicy implements BroadcastPolicy
+{
+    public boolean isComplete(int responded, int members)
+    {
+        return true;
+    }
+}

Propchange: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/policy/AsynchBroadcastPolicy.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/policy/MajorityResponseBroadcastPolicy.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/policy/MajorityResponseBroadcastPolicy.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/policy/MajorityResponseBroadcastPolicy.java (added)
+++ incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/policy/MajorityResponseBroadcastPolicy.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,28 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.cluster.policy;
+
+import org.apache.qpid.server.cluster.BroadcastPolicy;
+
+public class MajorityResponseBroadcastPolicy implements BroadcastPolicy
+{
+    public boolean isComplete(int responded, int members)
+    {
+        return responded > members / 2;
+    }
+}

Propchange: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/policy/MajorityResponseBroadcastPolicy.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/policy/OneResponseBroadcastPolicy.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/policy/OneResponseBroadcastPolicy.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/policy/OneResponseBroadcastPolicy.java (added)
+++ incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/policy/OneResponseBroadcastPolicy.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,28 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.cluster.policy;
+
+import org.apache.qpid.server.cluster.BroadcastPolicy;
+
+public class OneResponseBroadcastPolicy implements BroadcastPolicy
+{
+    public boolean isComplete(int responded, int members)
+    {
+        return responded > 0;
+    }
+}

Propchange: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/policy/OneResponseBroadcastPolicy.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/policy/StandardPolicies.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/policy/StandardPolicies.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/policy/StandardPolicies.java (added)
+++ incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/policy/StandardPolicies.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,26 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.cluster.policy;
+
+import org.apache.qpid.server.cluster.BroadcastPolicy;
+
+public interface StandardPolicies
+{
+    public static final BroadcastPolicy ASYNCH_POLICY = new AsynchBroadcastPolicy();
+    public static final BroadcastPolicy SYNCH_POLICY = new SynchBroadcastPolicy();
+}

Propchange: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/policy/StandardPolicies.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/policy/SynchBroadcastPolicy.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/policy/SynchBroadcastPolicy.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/policy/SynchBroadcastPolicy.java (added)
+++ incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/policy/SynchBroadcastPolicy.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,28 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.cluster.policy;
+
+import org.apache.qpid.server.cluster.BroadcastPolicy;
+
+public class SynchBroadcastPolicy implements BroadcastPolicy
+{
+    public boolean isComplete(int responded, int members)
+    {
+        return responded == members;
+    }
+}

Propchange: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/policy/SynchBroadcastPolicy.java
------------------------------------------------------------------------------
    svn:eol-style = native

Added: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/replay/ChainedMethodRecorder.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/replay/ChainedMethodRecorder.java?view=auto&rev=447994
==============================================================================
--- incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/replay/ChainedMethodRecorder.java (added)
+++ incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/replay/ChainedMethodRecorder.java Tue Sep 19 15:06:50 2006
@@ -0,0 +1,45 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed 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.cluster.replay;
+
+import org.apache.qpid.framing.AMQMethodBody;
+
+abstract class ChainedMethodRecorder <T extends AMQMethodBody> implements MethodRecorder<T>
+{
+    private final MethodRecorder<T> _recorder;
+
+    ChainedMethodRecorder()
+    {
+        this(null);
+    }
+
+    ChainedMethodRecorder(MethodRecorder<T> recorder)
+    {
+        _recorder = recorder;
+    }
+
+    public final void record(T method)
+    {
+        if(!doRecord(method) && _recorder != null)
+        {
+            _recorder.record(method);
+        }
+    }
+
+    protected abstract boolean doRecord(T method);
+}

Propchange: incubator/qpid/trunk/qpid/java/cluster/src/org/apache/qpid/server/cluster/replay/ChainedMethodRecorder.java
------------------------------------------------------------------------------
    svn:eol-style = native