QPID-4659 : [Java Broker] move amqp 0-10 implementation into a plugin

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1503446 13f79535-47bb-0310-9956-ffa450edef68

31 files changed, 7038 insertions, 25 deletions

diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/build.xml b/qpid/java/broker-plugins/amqp-0-10-protocol/build.xml
new file mode 100644
index 0000000000..e3e6fabc87
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/build.xml
@@ -0,0 +1,33 @@
+<!--
+ - 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.
+ -->
+<project name="Qpid Broker-Plugins AMQP 0-10 Protocol" default="build">
+    <property name="module.depends" value="common broker" />
+    <property name="module.test.depends" value="common/tests broker/tests" />
+
+    <property name="module.genpom" value="true"/>
+    <property name="module.genpom.args" value="-Sqpid-common=provided -Sqpid-broker=provided"/>
+    <property name="broker-plugins-amqp-0-10-protocol.libs" value="" />
+
+    <property name="broker.plugin" value="true"/>
+
+    <import file="../../module.xml" />
+
+    <target name="bundle" depends="bundle-tasks"/>
+
+</project>
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/CreditCreditManager.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/CreditCreditManager.java
new file mode 100644
index 0000000000..cee1a04b17
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/CreditCreditManager.java
@@ -0,0 +1,176 @@
+/*
+ *
+ * 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.v0_10;
+
+
+import org.apache.qpid.server.flow.AbstractFlowCreditManager;public class CreditCreditManager extends AbstractFlowCreditManager implements FlowCreditManager_0_10
+{
+    private volatile long _bytesCredit;
+    private volatile long _messageCredit;
+
+    public CreditCreditManager(long bytesCredit, long messageCredit)
+    {
+        _bytesCredit = bytesCredit;
+        _messageCredit = messageCredit;
+        setSuspended(!hasCredit());
+
+    }
+
+
+    public synchronized void setCreditLimits(final long bytesCredit, final long messageCredit)
+    {
+        _bytesCredit = bytesCredit;
+        _messageCredit = messageCredit;
+
+        setSuspended(!hasCredit());
+
+    }
+
+
+    public long getMessageCredit()
+    {
+         return _messageCredit == -1L
+                    ? Long.MAX_VALUE
+                    : _messageCredit;
+    }
+
+    public long getBytesCredit()
+    {
+        return _bytesCredit == -1L
+                    ? Long.MAX_VALUE
+                    : _bytesCredit;
+    }
+
+    public synchronized void restoreCredit(final long messageCredit, final long bytesCredit)
+    {
+    }
+
+
+    public synchronized void addCredit(final long messageCredit, final long bytesCredit)
+    {
+        boolean notifyIncrease = true;
+        if(_messageCredit >= 0L && messageCredit > 0L)
+        {
+            notifyIncrease = _messageCredit != 0L;
+            _messageCredit += messageCredit;
+        }
+
+
+
+        if(_bytesCredit >= 0L && bytesCredit > 0L)
+        {
+            notifyIncrease = notifyIncrease && bytesCredit>0;
+            _bytesCredit += bytesCredit;
+
+
+
+            if(notifyIncrease)
+            {
+                notifyIncreaseBytesCredit();
+            }
+        }
+
+
+
+        setSuspended(!hasCredit());
+
+    }
+
+    public void clearCredit()
+    {
+        _bytesCredit = 0l;
+        _messageCredit = 0l;
+        setSuspended(true);
+    }
+
+
+    public synchronized boolean hasCredit()
+    {
+        // Note !=, if credit is < 0 that indicates infinite credit
+        return (_bytesCredit != 0L  && _messageCredit != 0L);
+    }
+
+    public synchronized boolean useCreditForMessage(long msgSize)
+    {
+        if(_messageCredit >= 0L)
+        {
+            if(_messageCredit > 0)
+            {
+                if(_bytesCredit < 0L)
+                {
+                    _messageCredit--;
+
+                    return true;
+                }
+                else if(msgSize <= _bytesCredit)
+                {
+                    _messageCredit--;
+                    _bytesCredit -= msgSize;
+
+                    return true;
+                }
+                else
+                {
+                    return false;
+                }
+            }
+            else
+            {
+                setSuspended(true);
+                return false;
+            }
+        }
+        else if(_bytesCredit >= 0L)
+        {
+            if(msgSize <= _bytesCredit)
+            {
+                 _bytesCredit -= msgSize;
+
+                return true;
+            }
+            else
+            {
+                return false;
+            }
+
+        }
+        else
+        {
+            return true;
+        }
+
+    }
+
+    public synchronized void stop()
+    {
+        if(_bytesCredit > 0)
+        {
+            _bytesCredit = 0;
+        }
+        if(_messageCredit > 0)
+        {
+            _messageCredit = 0;
+        }
+
+    }
+
+
+}
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ExplicitAcceptDispositionChangeListener.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ExplicitAcceptDispositionChangeListener.java
new file mode 100755
index 0000000000..4b38b8a1a3
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ExplicitAcceptDispositionChangeListener.java
@@ -0,0 +1,93 @@
+/*
+ *
+ * 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.v0_10;
+
+import org.apache.log4j.Logger;
+
+import org.apache.qpid.server.queue.QueueEntry;
+
+
+class ExplicitAcceptDispositionChangeListener implements ServerSession.MessageDispositionChangeListener
+{
+    private static final Logger _logger = Logger.getLogger(ExplicitAcceptDispositionChangeListener.class);
+
+
+    private final QueueEntry _entry;
+    private final Subscription_0_10 _sub;
+
+    public ExplicitAcceptDispositionChangeListener(QueueEntry entry, Subscription_0_10 subscription_0_10)
+    {
+        _entry = entry;
+        _sub = subscription_0_10;
+    }
+
+    public void onAccept()
+    {
+        final Subscription_0_10 subscription = getSubscription();
+        if(subscription != null && _entry.isAcquiredBy(_sub))
+        {
+            subscription.getSessionModel().acknowledge(subscription, _entry);
+        }
+        else
+        {
+            _logger.warn("MessageAccept received for message which has not been acquired (likely client error)");
+        }
+
+    }
+
+    public void onRelease(boolean setRedelivered)
+    {
+        final Subscription_0_10 subscription = getSubscription();
+        if(subscription != null && _entry.isAcquiredBy(_sub))
+        {
+            subscription.release(_entry, setRedelivered);
+        }
+        else
+        {
+            _logger.warn("MessageRelease received for message which has not been acquired (likely client error)");
+        }
+    }
+
+    public void onReject()
+    {
+        final Subscription_0_10 subscription = getSubscription();
+        if(subscription != null && _entry.isAcquiredBy(_sub))
+        {
+            subscription.reject(_entry);
+        }
+        else
+        {
+            _logger.warn("MessageReject received for message which has not been acquired (likely client error)");
+        }
+
+    }
+
+    public boolean acquire()
+    {
+        return _entry.acquire(getSubscription());
+    }
+
+
+    private Subscription_0_10 getSubscription()
+    {
+        return _sub;
+    }
+}
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/FlowCreditManager_0_10.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/FlowCreditManager_0_10.java
new file mode 100755
index 0000000000..7f092814da
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/FlowCreditManager_0_10.java
@@ -0,0 +1,30 @@
+/*
+ *
+ * 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.v0_10;
+
+import org.apache.qpid.server.flow.FlowCreditManager;
+
+public interface FlowCreditManager_0_10 extends FlowCreditManager
+{
+    public void addCredit(long count, long bytes);
+
+    void clearCredit();
+}
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ImplicitAcceptDispositionChangeListener.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ImplicitAcceptDispositionChangeListener.java
new file mode 100755
index 0000000000..ce0155b789
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ImplicitAcceptDispositionChangeListener.java
@@ -0,0 +1,88 @@
+/*
+ *
+ * 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.v0_10;
+
+import org.apache.log4j.Logger;
+
+import org.apache.qpid.server.queue.QueueEntry;
+
+class ImplicitAcceptDispositionChangeListener implements ServerSession.MessageDispositionChangeListener
+{
+    private static final Logger _logger = Logger.getLogger(ImplicitAcceptDispositionChangeListener.class);
+
+
+    private final QueueEntry _entry;
+    private Subscription_0_10 _sub;
+
+    public ImplicitAcceptDispositionChangeListener(QueueEntry entry, Subscription_0_10 subscription_0_10)
+    {
+        _entry = entry;
+        _sub = subscription_0_10;
+    }
+
+    public void onAccept()
+    {
+        _logger.warn("MessageAccept received for message which is using NONE as the accept mode (likely client error)");
+    }
+
+    public void onRelease(boolean setRedelivered)
+    {
+        if(_entry.isAcquiredBy(_sub))
+        {
+            getSubscription().release(_entry, setRedelivered);
+        }
+        else
+        {
+            _logger.warn("MessageRelease received for message which has not been acquired (likely client error)");
+        }
+    }
+
+    public void onReject()
+    {
+        if(_entry.isAcquiredBy(_sub))
+        {
+            getSubscription().reject(_entry);
+        }
+        else
+        {
+            _logger.warn("MessageReject received for message which has not been acquired (likely client error)");
+        }
+
+    }
+
+    public boolean acquire()
+    {
+        boolean acquired = _entry.acquire(getSubscription());
+        if(acquired)
+        {
+            getSubscription().recordUnacknowledged(_entry);
+        }
+        return acquired;
+
+    }
+
+    public Subscription_0_10 getSubscription()
+    {
+        return _sub;
+    }
+
+
+}
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageAcceptCompletionListener.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageAcceptCompletionListener.java
new file mode 100755
index 0000000000..f5f2a8d43f
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageAcceptCompletionListener.java
@@ -0,0 +1,56 @@
+/*
+ *
+ * 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.v0_10;
+
+import org.apache.qpid.server.queue.QueueEntry;
+import org.apache.qpid.transport.Method;
+
+public class MessageAcceptCompletionListener implements Method.CompletionListener
+{
+    private final Subscription_0_10 _sub;
+    private final QueueEntry _entry;
+    private final ServerSession _session;
+    private boolean _restoreCredit;
+
+    public MessageAcceptCompletionListener(Subscription_0_10 sub, ServerSession session, QueueEntry entry, boolean restoreCredit)
+    {
+        super();
+        _sub = sub;
+        _entry = entry;
+        _session = session;
+        _restoreCredit = restoreCredit;
+    }
+
+    public void onComplete(Method method)
+    {
+        if(_restoreCredit)
+        {
+            _sub.restoreCredit(_entry);
+        }
+        if(_entry.isAcquiredBy(_sub))
+        {
+            _session.acknowledge(_sub, _entry);
+        }
+
+        _session.removeDispositionListener(method);
+    }
+}
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_v0_10.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_v0_10.java
new file mode 100644
index 0000000000..c6ae0c6e47
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_v0_10.java
@@ -0,0 +1,138 @@
+/*
+ *
+ * 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.v0_10;
+
+import java.nio.ByteBuffer;
+import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.plugin.MessageConverter;
+import org.apache.qpid.server.store.StoreFuture;
+import org.apache.qpid.server.store.StoredMessage;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.transport.DeliveryProperties;
+import org.apache.qpid.transport.Header;
+import org.apache.qpid.transport.MessageDeliveryPriority;
+import org.apache.qpid.transport.MessageProperties;
+
+public class MessageConverter_v0_10 implements MessageConverter<ServerMessage, MessageTransferMessage>
+{
+    @Override
+    public Class<ServerMessage> getInputClass()
+    {
+        return ServerMessage.class;
+    }
+
+    @Override
+    public Class<MessageTransferMessage> getOutputClass()
+    {
+        return MessageTransferMessage.class;
+    }
+
+    @Override
+    public MessageTransferMessage convert(ServerMessage serverMsg, VirtualHost vhost)
+    {
+        return new MessageTransferMessage(convertToStoredMessage(serverMsg), null);
+    }
+
+    private StoredMessage<MessageMetaData_0_10> convertToStoredMessage(final ServerMessage serverMsg)
+    {
+        final MessageMetaData_0_10 messageMetaData_0_10 = convertMetaData(serverMsg);
+
+        return new StoredMessage<MessageMetaData_0_10>()
+                {
+                    @Override
+                    public MessageMetaData_0_10 getMetaData()
+                    {
+                        return messageMetaData_0_10;
+                    }
+
+                    @Override
+                    public long getMessageNumber()
+                    {
+                        return serverMsg.getMessageNumber();
+                    }
+
+                    @Override
+                    public void addContent(int offsetInMessage, ByteBuffer src)
+                    {
+                        throw new UnsupportedOperationException();
+                    }
+
+                    @Override
+                    public int getContent(int offsetInMessage, ByteBuffer dst)
+                    {
+                        return serverMsg.getContent(dst, offsetInMessage);
+                    }
+
+                    @Override
+                    public ByteBuffer getContent(int offsetInMessage, int size)
+                    {
+                        return serverMsg.getContent(offsetInMessage, size);
+                    }
+
+                    @Override
+                    public StoreFuture flushToStore()
+                    {
+                        return StoreFuture.IMMEDIATE_FUTURE;
+                    }
+
+                    @Override
+                    public void remove()
+                    {
+                        throw new UnsupportedOperationException();
+                    }
+                };
+    }
+
+    private MessageMetaData_0_10 convertMetaData(ServerMessage serverMsg)
+    {
+        DeliveryProperties deliveryProps = new DeliveryProperties();
+        MessageProperties messageProps = new MessageProperties();
+
+        int size = (int) serverMsg.getSize();
+        ByteBuffer body = ByteBuffer.allocate(size);
+        serverMsg.getContent(body, 0);
+        body.flip();
+
+
+        deliveryProps.setExpiration(serverMsg.getExpiration());
+        deliveryProps.setImmediate(serverMsg.isImmediate());
+        deliveryProps.setPriority(MessageDeliveryPriority.get(serverMsg.getMessageHeader().getPriority()));
+        deliveryProps.setRoutingKey(serverMsg.getRoutingKey());
+        deliveryProps.setTimestamp(serverMsg.getMessageHeader().getTimestamp());
+
+        messageProps.setContentEncoding(serverMsg.getMessageHeader().getEncoding());
+        messageProps.setContentLength(size);
+        messageProps.setContentType(serverMsg.getMessageHeader().getMimeType());
+        if(serverMsg.getMessageHeader().getCorrelationId() != null)
+        {
+            messageProps.setCorrelationId(serverMsg.getMessageHeader().getCorrelationId().getBytes());
+        }
+
+        Header header = new Header(deliveryProps, messageProps, null);
+        return new MessageMetaData_0_10(header, size, serverMsg.getArrivalTime());
+    }
+
+    @Override
+    public String getType()
+    {
+        return "Unknown to v0-10";
+    }
+}
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageMetaDataType_0_10.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageMetaDataType_0_10.java
new file mode 100644
index 0000000000..90fb443f5b
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageMetaDataType_0_10.java
@@ -0,0 +1,67 @@
+/*
+ *
+ * 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.v0_10;
+
+import java.nio.ByteBuffer;
+import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.plugin.MessageMetaDataType;
+import org.apache.qpid.server.protocol.AmqpProtocolVersion;
+import org.apache.qpid.server.store.StoredMessage;
+
+public class MessageMetaDataType_0_10 implements MessageMetaDataType<MessageMetaData_0_10>
+{
+
+    public static final int TYPE = 1;
+
+    @Override
+    public int ordinal()
+    {
+        return TYPE;
+    }
+
+    @Override
+    public MessageMetaData_0_10 createMetaData(ByteBuffer buf)
+    {
+        return MessageMetaData_0_10.FACTORY.createMetaData(buf);
+    }
+
+    @Override
+    public ServerMessage<MessageMetaData_0_10> createMessage(StoredMessage<MessageMetaData_0_10> msg)
+    {
+        return new MessageTransferMessage(msg, null);
+    }
+
+    public int hashCode()
+    {
+        return ordinal();
+    }
+
+    public boolean equals(Object o)
+    {
+        return o != null && o.getClass() == getClass();
+    }
+
+    @Override
+    public String getType()
+    {
+        return AmqpProtocolVersion.v0_10.toString();
+    }
+}
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageMetaData_0_10.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageMetaData_0_10.java
new file mode 100755
index 0000000000..ee2a40a5b2
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageMetaData_0_10.java
@@ -0,0 +1,281 @@
+/*
+*
+* 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.v0_10;
+
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.server.message.AMQMessageHeader;
+import org.apache.qpid.server.message.InboundMessage;
+import org.apache.qpid.server.plugin.MessageMetaDataType;
+import org.apache.qpid.server.store.StorableMessageMetaData;
+import org.apache.qpid.transport.DeliveryProperties;
+import org.apache.qpid.transport.Header;
+import org.apache.qpid.transport.MessageDeliveryMode;
+import org.apache.qpid.transport.MessageProperties;
+import org.apache.qpid.transport.MessageTransfer;
+import org.apache.qpid.transport.Struct;
+import org.apache.qpid.transport.codec.BBDecoder;
+import org.apache.qpid.transport.codec.BBEncoder;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+
+public class MessageMetaData_0_10 implements StorableMessageMetaData, InboundMessage
+{
+    private Header _header;
+    private DeliveryProperties _deliveryProps;
+    private MessageProperties _messageProps;
+    private MessageTransferHeader _messageHeader;
+    private long _arrivalTime;
+    private int _bodySize;
+
+    private static final int ENCODER_SIZE = 1 << 10;
+
+    public static final MessageMetaDataType.Factory<MessageMetaData_0_10> FACTORY = new MetaDataFactory();
+
+    private static final MessageMetaDataType_0_10 TYPE = new MessageMetaDataType_0_10();
+
+    private volatile ByteBuffer _encoded;
+    private Object _connectionReference;
+
+
+    public MessageMetaData_0_10(MessageTransfer xfr)
+    {
+        this(xfr.getHeader(), xfr.getBodySize(), System.currentTimeMillis());
+    }
+
+    public MessageMetaData_0_10(Header header, int bodySize, long arrivalTime)
+    {
+        _header = header;
+        if(_header != null)
+        {
+            _deliveryProps = _header.getDeliveryProperties();
+            _messageProps = _header.getMessageProperties();
+        }
+        else
+        {
+            _deliveryProps = null;
+            _messageProps = null;
+        }
+        _messageHeader = new MessageTransferHeader(_deliveryProps, _messageProps);
+        _arrivalTime = arrivalTime;
+        _bodySize = bodySize;
+
+    }
+
+
+
+    public MessageMetaDataType getType()
+    {
+        return TYPE;
+    }
+
+    public int getStorableSize()
+    {
+        ByteBuffer buf = _encoded;
+
+        if(buf == null)
+        {
+            buf = encodeAsBuffer();
+            _encoded = buf;
+        }
+
+        //TODO -- need to add stuff
+        return buf.limit();
+    }
+
+    private ByteBuffer encodeAsBuffer()
+    {
+        BBEncoder encoder = new BBEncoder(ENCODER_SIZE);
+
+        encoder.writeInt64(_arrivalTime);
+        encoder.writeInt32(_bodySize);
+        int headersLength = 0;
+        if(_header.getDeliveryProperties() != null)
+        {
+            headersLength++;
+        }
+        if(_header.getMessageProperties() != null)
+        {
+            headersLength++;
+        }
+        if(_header.getNonStandardProperties() != null)
+        {
+            headersLength += _header.getNonStandardProperties().size();
+        }
+
+        encoder.writeInt32(headersLength);
+
+        if(_header.getDeliveryProperties() != null)
+        {
+            encoder.writeStruct32(_header.getDeliveryProperties());
+        }
+        if(_header.getMessageProperties() != null)
+        {
+            encoder.writeStruct32(_header.getMessageProperties());
+        }
+        if(_header.getNonStandardProperties() != null)
+        {
+
+            for(Struct header : _header.getNonStandardProperties())
+            {
+                encoder.writeStruct32(header);
+            }
+
+        }
+        ByteBuffer buf = encoder.buffer();
+        return buf;
+    }
+
+    public int writeToBuffer(int offsetInMetaData, ByteBuffer dest)
+    {
+        ByteBuffer buf = _encoded;
+
+        if(buf == null)
+        {
+            buf = encodeAsBuffer();
+            _encoded = buf;
+        }
+
+        buf = buf.duplicate();
+
+        buf.position(offsetInMetaData);
+
+        if(dest.remaining() < buf.limit())
+        {
+            buf.limit(dest.remaining());
+        }
+        dest.put(buf);
+        return buf.limit();
+    }
+
+    public int getContentSize()
+    {
+        return _bodySize;
+    }
+
+    public boolean isPersistent()
+    {
+        return _deliveryProps == null ? false : _deliveryProps.getDeliveryMode() == MessageDeliveryMode.PERSISTENT;
+    }
+
+    public String getRoutingKey()
+    {
+        return _deliveryProps == null ? null : _deliveryProps.getRoutingKey();
+    }
+
+    public AMQShortString getRoutingKeyShortString()
+    {
+        return AMQShortString.valueOf(getRoutingKey());
+    }
+
+    public AMQMessageHeader getMessageHeader()
+    {
+        return _messageHeader;
+    }
+
+    public long getSize()
+    {
+
+        return _bodySize;
+    }
+
+    public boolean isImmediate()
+    {
+        return _deliveryProps != null && _deliveryProps.getImmediate();
+    }
+
+    public long getExpiration()
+    {
+        return _deliveryProps == null ? 0L : _deliveryProps.getExpiration();
+    }
+
+    public boolean isRedelivered()
+    {
+        // The *Message* is never redelivered, only queue entries are...
+        return false;
+    }
+
+    public long getArrivalTime()
+    {
+        return _arrivalTime;
+    }
+
+    public Header getHeader()
+    {
+        return _header;
+    }
+
+    public void setConnectionReference(Object connectionReference)
+    {
+        _connectionReference = connectionReference;
+    }
+
+    public Object getConnectionReference()
+    {
+        return _connectionReference;
+    }
+
+    private static class MetaDataFactory implements MessageMetaDataType.Factory<MessageMetaData_0_10>
+    {
+        public MessageMetaData_0_10 createMetaData(ByteBuffer buf)
+        {
+            BBDecoder decoder = new BBDecoder();
+            decoder.init(buf);
+
+            long arrivalTime = decoder.readInt64();
+            int bodySize = decoder.readInt32();
+            int headerCount = decoder.readInt32();
+
+            DeliveryProperties deliveryProperties = null;
+            MessageProperties messageProperties = null;
+            List<Struct> otherProps = null;
+
+            for(int i = 0 ; i < headerCount; i++)
+            {
+                Struct struct = decoder.readStruct32();
+                if(struct instanceof DeliveryProperties && deliveryProperties == null)
+                {
+                    deliveryProperties = (DeliveryProperties) struct;
+                }
+                else if(struct instanceof MessageProperties && messageProperties == null)
+                {
+                    messageProperties = (MessageProperties) struct;
+                }
+                else
+                {
+                    if(otherProps == null)
+                    {
+                        otherProps = new ArrayList<Struct>();
+
+                    }
+                    otherProps.add(struct);
+                }
+            }
+            Header header = new Header(deliveryProperties,messageProperties,otherProps);
+
+            return new MessageMetaData_0_10(header, bodySize, arrivalTime);
+
+        }
+    }
+
+
+}
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageTransferHeader.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageTransferHeader.java
new file mode 100644
index 0000000000..1b506d9bf8
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageTransferHeader.java
@@ -0,0 +1,171 @@
+/*
+ *
+ * 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.v0_10;
+
+import java.util.*;
+import org.apache.qpid.server.message.AMQMessageHeader;
+import org.apache.qpid.transport.DeliveryProperties;
+import org.apache.qpid.transport.MessageDeliveryPriority;
+import org.apache.qpid.transport.MessageProperties;
+
+class MessageTransferHeader implements AMQMessageHeader
+{
+
+
+    public static final String JMS_TYPE = "x-jms-type";
+
+    private final DeliveryProperties _deliveryProps;
+    private final MessageProperties _messageProps;
+
+    public MessageTransferHeader(DeliveryProperties deliveryProps, MessageProperties messageProps)
+    {
+        _deliveryProps = deliveryProps;
+        _messageProps = messageProps;
+    }
+
+    public String getCorrelationId()
+    {
+        if (_messageProps != null && _messageProps.getCorrelationId() != null)
+        {
+            return new String(_messageProps.getCorrelationId());
+        }
+        else
+        {
+            return null;
+        }
+    }
+
+    public long getExpiration()
+    {
+        return _deliveryProps == null ? 0L : _deliveryProps.getExpiration();
+    }
+
+    public String getUserId()
+    {
+        byte[] userIdBytes = _messageProps == null ? null : _messageProps.getUserId();
+        return userIdBytes == null ? null : new String(userIdBytes);
+    }
+
+    public String getAppId()
+    {
+        byte[] appIdBytes = _messageProps == null ? null : _messageProps.getAppId();
+        return appIdBytes == null ? null : new String(appIdBytes);
+    }
+
+    public String getMessageId()
+    {
+        UUID id = _messageProps == null ? null : _messageProps.getMessageId();
+
+        return id == null ? null : String.valueOf(id);
+    }
+
+    public String getMimeType()
+    {
+        return _messageProps == null ? null : _messageProps.getContentType();
+    }
+
+    public String getEncoding()
+    {
+        return _messageProps == null ? null : _messageProps.getContentEncoding();
+    }
+
+    public byte getPriority()
+    {
+        MessageDeliveryPriority priority = _deliveryProps == null || !_deliveryProps.hasPriority()
+                                           ? MessageDeliveryPriority.MEDIUM
+                                           : _deliveryProps.getPriority();
+        return (byte) priority.getValue();
+    }
+
+    public long getTimestamp()
+    {
+        return _deliveryProps == null ? 0L : _deliveryProps.getTimestamp();
+    }
+
+    public String getType()
+    {
+        Object type = getHeader(JMS_TYPE);
+        return type instanceof String ? (String) type : null;
+    }
+
+    public String getReplyTo()
+    {
+        if (_messageProps != null && _messageProps.getReplyTo() != null)
+        {
+            return _messageProps.getReplyTo().toString();
+        }
+        else
+        {
+            return null;
+        }
+    }
+
+    public String getReplyToExchange()
+    {
+        if (_messageProps != null && _messageProps.getReplyTo() != null)
+        {
+            return _messageProps.getReplyTo().getExchange();
+        }
+        else
+        {
+            return null;
+        }
+    }
+
+    public String getReplyToRoutingKey()
+    {
+        if (_messageProps != null && _messageProps.getReplyTo() != null)
+        {
+            return _messageProps.getReplyTo().getRoutingKey();
+        }
+        else
+        {
+            return null;
+        }
+    }
+
+    public Object getHeader(String name)
+    {
+        Map<String, Object> appHeaders = _messageProps == null ? null : _messageProps.getApplicationHeaders();
+        return appHeaders == null ? null : appHeaders.get(name);
+    }
+
+    public boolean containsHeaders(Set<String> names)
+    {
+        Map<String, Object> appHeaders = _messageProps == null ? null : _messageProps.getApplicationHeaders();
+        return appHeaders != null && appHeaders.keySet().containsAll(names);
+
+    }
+
+    @Override
+    public Collection<String> getHeaderNames()
+    {
+        Map<String, Object> appHeaders = _messageProps == null ? null : _messageProps.getApplicationHeaders();
+        return appHeaders != null ? Collections.unmodifiableCollection(appHeaders.keySet()) : Collections.EMPTY_SET ;
+
+    }
+
+    public boolean containsHeader(String name)
+    {
+        Map<String, Object> appHeaders = _messageProps == null ? null : _messageProps.getApplicationHeaders();
+        return appHeaders != null && appHeaders.containsKey(name);
+    }
+}
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageTransferMessage.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageTransferMessage.java
new file mode 100644
index 0000000000..4e8bfcb652
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageTransferMessage.java
@@ -0,0 +1,136 @@
+/*
+ *
+ * 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.v0_10;
+
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.server.message.AMQMessageHeader;
+import org.apache.qpid.server.message.AbstractServerMessageImpl;
+import org.apache.qpid.server.message.InboundMessage;
+import org.apache.qpid.server.message.MessageReference;
+import org.apache.qpid.server.store.StoredMessage;
+import org.apache.qpid.transport.Header;
+
+import java.nio.ByteBuffer;
+
+
+public class MessageTransferMessage extends AbstractServerMessageImpl<MessageMetaData_0_10> implements InboundMessage
+{
+
+    private Object _connectionRef;
+
+    public MessageTransferMessage(StoredMessage<MessageMetaData_0_10> storeMessage, Object connectionRef)
+    {
+        super(storeMessage);
+        _connectionRef = connectionRef;
+    }
+
+    private MessageMetaData_0_10 getMetaData()
+    {
+        return getStoredMessage().getMetaData();
+    }
+
+    public String getRoutingKey()
+    {
+        return getMetaData().getRoutingKey();
+    }
+
+    public AMQShortString getRoutingKeyShortString()
+    {
+        return AMQShortString.valueOf(getRoutingKey());
+    }
+
+    public AMQMessageHeader getMessageHeader()
+    {
+        return getMetaData().getMessageHeader();
+    }
+
+    public boolean isPersistent()
+    {
+        return getMetaData().isPersistent();
+    }
+
+
+    public boolean isRedelivered()
+    {
+        // The *Message* is never redelivered, only queue entries are... this is here so that filters
+        // can run against the message on entry to an exchange
+        return false;
+    }
+
+    public long getSize()
+    {
+
+        return getMetaData().getSize();
+    }
+
+    public boolean isImmediate()
+    {
+        return getMetaData().isImmediate();
+    }
+
+    public long getExpiration()
+    {
+        return getMetaData().getExpiration();
+    }
+
+    public MessageReference newReference()
+    {
+        return new TransferMessageReference(this);
+    }
+
+    public long getMessageNumber()
+    {
+        return getStoredMessage().getMessageNumber();
+    }
+
+    public long getArrivalTime()
+    {
+        return getMetaData().getArrivalTime();
+    }
+
+    public int getContent(ByteBuffer buf, int offset)
+    {
+        return getStoredMessage().getContent(offset, buf);
+    }
+
+
+    public ByteBuffer getContent(int offset, int size)
+    {
+        return getStoredMessage().getContent(offset,size);
+    }
+
+    public Header getHeader()
+    {
+        return getMetaData().getHeader();
+    }
+
+    public ByteBuffer getBody()
+    {
+
+        return  getContent(0, (int)getSize());
+    }
+
+    public Object getConnectionReference()
+    {
+        return _connectionRef;
+    }
+
+}
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngineCreator_0_10.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngineCreator_0_10.java
new file mode 100644
index 0000000000..ab50a33b9b
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngineCreator_0_10.java
@@ -0,0 +1,100 @@
+/*
+ *
+ * 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.v0_10;
+
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import org.apache.qpid.protocol.ServerProtocolEngine;
+import org.apache.qpid.server.model.Broker;
+import org.apache.qpid.server.model.Port;
+import org.apache.qpid.server.model.Transport;
+import org.apache.qpid.server.protocol.AmqpProtocolVersion;
+import org.apache.qpid.server.plugin.ProtocolEngineCreator;
+import org.apache.qpid.transport.ConnectionDelegate;
+import org.apache.qpid.transport.network.NetworkConnection;
+
+public class ProtocolEngineCreator_0_10 implements ProtocolEngineCreator
+{
+
+    private static final byte[] AMQP_0_10_HEADER =
+            new byte[] { (byte) 'A',
+                         (byte) 'M',
+                         (byte) 'Q',
+                         (byte) 'P',
+                         (byte) 1,
+                         (byte) 1,
+                         (byte) 0,
+                         (byte) 10
+            };
+
+
+    public ProtocolEngineCreator_0_10()
+    {
+    }
+
+    public AmqpProtocolVersion getVersion()
+    {
+        return AmqpProtocolVersion.v0_10;
+    }
+
+
+    public byte[] getHeaderIdentifier()
+    {
+        return AMQP_0_10_HEADER;
+    }
+
+    public ServerProtocolEngine newProtocolEngine(Broker broker,
+                                                  NetworkConnection network,
+                                                  Port port,
+                                                  Transport transport,
+                                                  long id)
+    {
+        String fqdn = null;
+        SocketAddress address = network.getLocalAddress();
+        if (address instanceof InetSocketAddress)
+        {
+            fqdn = ((InetSocketAddress) address).getHostName();
+        }
+        final ConnectionDelegate connDelegate = new ServerConnectionDelegate(broker,
+                fqdn, broker.getSubjectCreator(address));
+
+        ServerConnection conn = new ServerConnection(id,broker);
+
+        conn.setConnectionDelegate(connDelegate);
+        conn.setRemoteAddress(network.getRemoteAddress());
+        conn.setLocalAddress(network.getLocalAddress());
+        return new ProtocolEngine_0_10( conn, network, port, transport);
+    }
+
+
+    private static ProtocolEngineCreator INSTANCE = new ProtocolEngineCreator_0_10();
+
+    public static ProtocolEngineCreator getInstance()
+    {
+        return INSTANCE;
+    }
+
+    @Override
+    public String getType()
+    {
+        return getVersion().toString();
+    }
+}
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngine_0_10.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngine_0_10.java
new file mode 100755
index 0000000000..73708d9841
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngine_0_10.java
@@ -0,0 +1,200 @@
+/*
+ *
+ * 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.v0_10;
+
+import org.apache.qpid.protocol.ServerProtocolEngine;
+import org.apache.qpid.server.logging.messages.ConnectionMessages;
+import org.apache.qpid.server.model.Port;
+import org.apache.qpid.server.model.Transport;
+import org.apache.qpid.server.protocol.v0_10.ServerConnection;
+import org.apache.qpid.transport.Sender;
+import org.apache.qpid.transport.network.Assembler;
+import org.apache.qpid.transport.network.Disassembler;
+import org.apache.qpid.transport.network.InputHandler;
+import org.apache.qpid.transport.network.NetworkConnection;
+
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+
+
+public class ProtocolEngine_0_10  extends InputHandler implements ServerProtocolEngine
+{
+    public static final int MAX_FRAME_SIZE = 64 * 1024 - 1;
+
+    private NetworkConnection _network;
+    private long _readBytes;
+    private long _writtenBytes;
+    private ServerConnection _connection;
+
+    private long _createTime = System.currentTimeMillis();
+    private long _lastReadTime;
+    private long _lastWriteTime;
+
+    public ProtocolEngine_0_10(ServerConnection conn,
+                               NetworkConnection network,
+                               Port port,
+                               Transport transport)
+    {
+        super(new Assembler(conn));
+        _connection = conn;
+        _connection.setPort(port);
+        _connection.setTransport(transport);
+
+        if(network != null)
+        {
+            setNetworkConnection(network);
+        }
+
+
+    }
+
+    public void setNetworkConnection(NetworkConnection network)
+    {
+        setNetworkConnection(network, network.getSender());
+    }
+
+    public void setNetworkConnection(NetworkConnection network, Sender<ByteBuffer> sender)
+    {
+        _network = network;
+
+        _connection.setNetworkConnection(network);
+        _connection.setSender(new Disassembler(wrapSender(sender), MAX_FRAME_SIZE));
+        _connection.setPeerPrincipal(_network.getPeerPrincipal());
+        // FIXME Two log messages to maintain compatibility with earlier protocol versions
+        _connection.getLogActor().message(ConnectionMessages.OPEN(null, null, null, false, false, false));
+        _connection.getLogActor().message(ConnectionMessages.OPEN(null, "0-10", null, false, true, false));
+    }
+
+    private Sender<ByteBuffer> wrapSender(final Sender<ByteBuffer> sender)
+    {
+        return new Sender<ByteBuffer>()
+        {
+            @Override
+            public void setIdleTimeout(int i)
+            {
+                sender.setIdleTimeout(i);
+
+            }
+
+            @Override
+            public void send(ByteBuffer msg)
+            {
+                _lastWriteTime = System.currentTimeMillis();
+                sender.send(msg);
+
+            }
+
+            @Override
+            public void flush()
+            {
+                sender.flush();
+
+            }
+
+            @Override
+            public void close()
+            {
+                sender.close();
+
+            }
+        };
+    }
+
+    @Override
+    public long getLastReadTime()
+    {
+        return _lastReadTime;
+    }
+
+    @Override
+    public long getLastWriteTime()
+    {
+        return _lastWriteTime;
+    }
+
+    public SocketAddress getRemoteAddress()
+    {
+        return _network.getRemoteAddress();
+    }
+
+    public SocketAddress getLocalAddress()
+    {
+        return _network.getLocalAddress();
+    }
+
+    public void received(final ByteBuffer buf)
+    {
+        _lastReadTime = System.currentTimeMillis();
+        super.received(buf);
+        _connection.receivedComplete();
+    }
+
+    public long getReadBytes()
+    {
+        return _readBytes;
+    }
+
+    public long getWrittenBytes()
+    {
+        return _writtenBytes;
+    }
+
+    public void writerIdle()
+    {
+        _connection.doHeartbeat();
+    }
+
+    public void readerIdle()
+    {
+        //Todo
+    }
+
+    public String getAddress()
+    {
+        return getRemoteAddress().toString();
+    }
+
+    public String getAuthId()
+    {
+        return _connection.getAuthorizedPrincipal() == null ? null : _connection.getAuthorizedPrincipal().getName();
+    }
+
+    public boolean isDurable()
+    {
+        return false;
+    }
+
+    @Override
+    public void closed()
+    {
+        super.closed();
+    }
+
+    public long getCreateTime()
+    {
+        return _createTime;
+    }
+
+    public long getConnectionId()
+    {
+        return _connection.getConnectionId();
+    }
+}
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
new file mode 100644
index 0000000000..0015988ab7
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
@@ -0,0 +1,560 @@
+/*
+ *
+ * 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.v0_10;
+
+import java.net.SocketAddress;
+import java.security.Principal;
+import java.text.MessageFormat;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import javax.security.auth.Subject;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.protocol.AMQConstant;
+import org.apache.qpid.server.logging.LogActor;
+import org.apache.qpid.server.logging.LogSubject;
+import org.apache.qpid.server.logging.actors.AMQPConnectionActor;
+import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.logging.actors.GenericActor;
+import org.apache.qpid.server.logging.messages.ConnectionMessages;
+import org.apache.qpid.server.model.Broker;
+import org.apache.qpid.server.model.Port;
+import org.apache.qpid.server.model.Transport;
+import org.apache.qpid.server.protocol.AMQConnectionModel;
+import org.apache.qpid.server.protocol.AMQSessionModel;
+import org.apache.qpid.server.security.AuthorizationHolder;
+import org.apache.qpid.server.security.auth.AuthenticatedPrincipal;
+import org.apache.qpid.server.stats.StatisticsCounter;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.transport.Connection;
+import org.apache.qpid.transport.ConnectionCloseCode;
+import org.apache.qpid.transport.ExecutionErrorCode;
+import org.apache.qpid.transport.ExecutionException;
+import org.apache.qpid.transport.Method;
+import org.apache.qpid.transport.ProtocolEvent;
+import org.apache.qpid.transport.Session;
+import org.apache.qpid.transport.network.NetworkConnection;
+
+import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.CONNECTION_FORMAT;
+import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.SOCKET_FORMAT;
+import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.USER_FORMAT;
+
+public class ServerConnection extends Connection implements AMQConnectionModel, LogSubject, AuthorizationHolder
+{
+    private Runnable _onOpenTask;
+    private AtomicBoolean _logClosed = new AtomicBoolean(false);
+    private LogActor _actor;
+
+    private Subject _authorizedSubject = null;
+    private Principal _authorizedPrincipal = null;
+    private StatisticsCounter _messagesDelivered, _dataDelivered, _messagesReceived, _dataReceived;
+    private final long _connectionId;
+    private final Object _reference = new Object();
+    private VirtualHost _virtualHost;
+    private Port _port;
+    private AtomicLong _lastIoTime = new AtomicLong();
+    private boolean _blocking;
+    private Principal _peerPrincipal;
+    private NetworkConnection _networkConnection;
+    private Transport _transport;
+    private volatile boolean _stopped;
+
+    public ServerConnection(final long connectionId, Broker broker)
+    {
+        _connectionId = connectionId;
+        _actor = new AMQPConnectionActor(this, broker.getRootMessageLogger());
+    }
+
+    public Object getReference()
+    {
+        return _reference;
+    }
+
+    @Override
+    protected void invoke(Method method)
+    {
+        super.invoke(method);
+    }
+
+    @Override
+    protected void setState(State state)
+    {
+        super.setState(state);
+
+        if (state == State.OPEN)
+        {
+            if (_onOpenTask != null)
+            {
+                _onOpenTask.run();
+            }
+            _actor.message(ConnectionMessages.OPEN(getClientId(), "0-10", getClientVersion(), true, true, true));
+
+            getVirtualHost().getConnectionRegistry().registerConnection(this);
+        }
+
+        if (state == State.CLOSE_RCVD || state == State.CLOSED || state == State.CLOSING)
+        {
+            if(_virtualHost != null)
+            {
+                _virtualHost.getConnectionRegistry().deregisterConnection(this);
+            }
+        }
+
+        if (state == State.CLOSED)
+        {
+            logClosed();
+        }
+    }
+
+    protected void logClosed()
+    {
+        if(_logClosed.compareAndSet(false, true))
+        {
+            CurrentActor.get().message(this, ConnectionMessages.CLOSE());
+        }
+    }
+
+    @Override
+    public ServerConnectionDelegate getConnectionDelegate()
+    {
+        return (ServerConnectionDelegate) super.getConnectionDelegate();
+    }
+
+    public void setConnectionDelegate(ServerConnectionDelegate delegate)
+    {
+        super.setConnectionDelegate(delegate);
+    }
+
+    public VirtualHost getVirtualHost()
+    {
+        return _virtualHost;
+    }
+
+    public void setVirtualHost(VirtualHost virtualHost)
+    {
+        _virtualHost = virtualHost;
+
+        initialiseStatistics();
+    }
+
+    @Override
+    public String getVirtualHostName()
+    {
+        return _virtualHost == null ? null : _virtualHost.getName();
+    }
+
+    @Override
+    public Port getPort()
+    {
+        return _port;
+    }
+
+    public void setPort(Port port)
+    {
+        _port = port;
+    }
+
+    @Override
+    public Transport getTransport()
+    {
+        return _transport;
+    }
+
+    @Override
+    public void stop()
+    {
+        _stopped = true;
+    }
+
+    @Override
+    public boolean isStopped()
+    {
+        return _stopped;
+    }
+
+    public void setTransport(Transport transport)
+    {
+        _transport = transport;
+    }
+
+    public void onOpen(final Runnable task)
+    {
+        _onOpenTask = task;
+    }
+
+    public void closeSession(AMQSessionModel session, AMQConstant cause, String message) throws AMQException
+    {
+        ExecutionException ex = new ExecutionException();
+        ExecutionErrorCode code = ExecutionErrorCode.INTERNAL_ERROR;
+        try
+        {
+	        code = ExecutionErrorCode.get(cause.getCode());
+        }
+        catch (IllegalArgumentException iae)
+        {
+            // Ignore, already set to INTERNAL_ERROR
+        }
+        ex.setErrorCode(code);
+        ex.setDescription(message);
+        ((ServerSession)session).invoke(ex);
+
+        ((ServerSession)session).close(cause, message);
+    }
+
+    public LogSubject getLogSubject()
+    {
+        return (LogSubject) this;
+    }
+
+    @Override
+    public void received(ProtocolEvent event)
+    {
+        _lastIoTime.set(System.currentTimeMillis());
+        if (event.isConnectionControl())
+        {
+            CurrentActor.set(_actor);
+        }
+        else
+        {
+            ServerSession channel = (ServerSession) getSession(event.getChannel());
+            LogActor channelActor = null;
+
+            if (channel != null)
+            {
+                channelActor = channel.getLogActor();
+            }
+
+            CurrentActor.set(channelActor == null ? _actor : channelActor);
+        }
+
+        try
+        {
+            super.received(event);
+        }
+        finally
+        {
+            CurrentActor.remove();
+        }
+    }
+
+    public String toLogString()
+    {
+        boolean hasVirtualHost = (null != this.getVirtualHost());
+        boolean hasClientId = (null != getClientId());
+
+        if (hasClientId && hasVirtualHost)
+        {
+            return "[" +
+                    MessageFormat.format(CONNECTION_FORMAT,
+                                         getConnectionId(),
+                                         getClientId(),
+                                         getRemoteAddressString(),
+                                         getVirtualHost().getName())
+                 + "] ";
+        }
+        else if (hasClientId)
+        {
+            return "[" +
+                    MessageFormat.format(USER_FORMAT,
+                                         getConnectionId(),
+                                         getClientId(),
+                                         getRemoteAddressString())
+                 + "] ";
+
+        }
+        else
+        {
+            return "[" +
+                    MessageFormat.format(SOCKET_FORMAT,
+                                         getConnectionId(),
+                                         getRemoteAddressString())
+                 + "] ";
+        }
+    }
+
+    public LogActor getLogActor()
+    {
+        return _actor;
+    }
+
+    public void close(AMQConstant cause, String message) throws AMQException
+    {
+        closeSubscriptions();
+        ConnectionCloseCode replyCode = ConnectionCloseCode.NORMAL;
+        try
+        {
+            replyCode = ConnectionCloseCode.get(cause.getCode());
+        }
+        catch (IllegalArgumentException iae)
+        {
+            // Ignore
+        }
+        close(replyCode, message);
+    }
+
+    public synchronized void block()
+    {
+        if(!_blocking)
+        {
+            _blocking = true;
+            for(AMQSessionModel ssn : getSessionModels())
+            {
+                ssn.block();
+            }
+        }
+    }
+
+    public synchronized void unblock()
+    {
+        if(_blocking)
+        {
+            _blocking = false;
+            for(AMQSessionModel ssn : getSessionModels())
+            {
+                ssn.unblock();
+            }
+        }
+    }
+
+    @Override
+    public synchronized void registerSession(final Session ssn)
+    {
+        super.registerSession(ssn);
+        if(_blocking)
+        {
+            ((ServerSession)ssn).block();
+        }
+    }
+
+    @Override
+    public synchronized void removeSession(final Session ssn)
+    {
+        super.removeSession(ssn);
+    }
+
+    public List<AMQSessionModel> getSessionModels()
+    {
+        List<AMQSessionModel> sessions = new ArrayList<AMQSessionModel>();
+        for (Session ssn : getChannels())
+        {
+            sessions.add((AMQSessionModel) ssn);
+        }
+        return sessions;
+    }
+
+    public void registerMessageDelivered(long messageSize)
+    {
+        _messagesDelivered.registerEvent(1L);
+        _dataDelivered.registerEvent(messageSize);
+        _virtualHost.registerMessageDelivered(messageSize);
+    }
+
+    public void registerMessageReceived(long messageSize, long timestamp)
+    {
+        _messagesReceived.registerEvent(1L, timestamp);
+        _dataReceived.registerEvent(messageSize, timestamp);
+        _virtualHost.registerMessageReceived(messageSize, timestamp);
+    }
+
+    public StatisticsCounter getMessageReceiptStatistics()
+    {
+        return _messagesReceived;
+    }
+
+    public StatisticsCounter getDataReceiptStatistics()
+    {
+        return _dataReceived;
+    }
+
+    public StatisticsCounter getMessageDeliveryStatistics()
+    {
+        return _messagesDelivered;
+    }
+
+    public StatisticsCounter getDataDeliveryStatistics()
+    {
+        return _dataDelivered;
+    }
+
+    public void resetStatistics()
+    {
+        _messagesDelivered.reset();
+        _dataDelivered.reset();
+        _messagesReceived.reset();
+        _dataReceived.reset();
+    }
+
+    public void initialiseStatistics()
+    {
+        _messagesDelivered = new StatisticsCounter("messages-delivered-" + getConnectionId());
+        _dataDelivered = new StatisticsCounter("data-delivered-" + getConnectionId());
+        _messagesReceived = new StatisticsCounter("messages-received-" + getConnectionId());
+        _dataReceived = new StatisticsCounter("data-received-" + getConnectionId());
+    }
+
+    /**
+     * @return authorizedSubject
+     */
+    public Subject getAuthorizedSubject()
+    {
+        return _authorizedSubject;
+    }
+
+    /**
+     * Sets the authorized subject.  It also extracts the UsernamePrincipal from the subject
+     * and caches it for optimisation purposes.
+     *
+     * @param authorizedSubject
+     */
+    public void setAuthorizedSubject(final Subject authorizedSubject)
+    {
+        if (authorizedSubject == null)
+        {
+            _authorizedSubject = null;
+            _authorizedPrincipal = null;
+        }
+        else
+        {
+            _authorizedSubject = authorizedSubject;
+            _authorizedPrincipal = AuthenticatedPrincipal.getAuthenticatedPrincipalFromSubject(authorizedSubject);
+        }
+    }
+
+    public Principal getAuthorizedPrincipal()
+    {
+        return _authorizedPrincipal;
+    }
+
+    public long getConnectionId()
+    {
+        return _connectionId;
+    }
+
+    public boolean isSessionNameUnique(byte[] name)
+    {
+        return !super.hasSessionWithName(name);
+    }
+
+    public String getRemoteAddressString()
+    {
+        return String.valueOf(getRemoteAddress());
+    }
+
+    public String getUserName()
+    {
+        return _authorizedPrincipal.getName();
+    }
+
+    @Override
+    public void closed()
+    {
+        closeSubscriptions();
+        super.closed();
+    }
+
+    private void closeSubscriptions()
+    {
+        for (Session ssn : getChannels())
+        {
+            ((ServerSession)ssn).unregisterSubscriptions();
+        }
+    }
+
+    public void receivedComplete()
+    {
+        for (Session ssn : getChannels())
+        {
+            ((ServerSession)ssn).receivedComplete();
+        }
+    }
+
+    @Override
+    public void send(ProtocolEvent event)
+    {
+        _lastIoTime.set(System.currentTimeMillis());
+        super.send(event);
+    }
+
+    public long getLastIoTime()
+    {
+        return _lastIoTime.longValue();
+    }
+
+
+    public String getClientId()
+    {
+        return getConnectionDelegate().getClientId();
+    }
+
+    public String getClientVersion()
+    {
+        return getConnectionDelegate().getClientVersion();
+    }
+
+    public String getPrincipalAsString()
+    {
+        return getAuthorizedPrincipal() == null ? null : getAuthorizedPrincipal().getName();
+    }
+
+    public long getSessionCountLimit()
+    {
+        return getChannelMax();
+    }
+
+    public Principal getPeerPrincipal()
+    {
+        return _peerPrincipal;
+    }
+
+    public void setPeerPrincipal(Principal peerPrincipal)
+    {
+        _peerPrincipal = peerPrincipal;
+    }
+
+    @Override
+    public void setRemoteAddress(SocketAddress remoteAddress)
+    {
+        super.setRemoteAddress(remoteAddress);
+    }
+
+    @Override
+    public void setLocalAddress(SocketAddress localAddress)
+    {
+        super.setLocalAddress(localAddress);
+    }
+
+    public void setNetworkConnection(NetworkConnection network)
+    {
+        _networkConnection = network;
+    }
+
+    public NetworkConnection getNetworkConnection()
+    {
+        return _networkConnection;
+    }
+
+    public void doHeartbeat()
+    {
+        super.doHeartBeat();
+
+    }
+}
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java
new file mode 100644
index 0000000000..6634627805
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java
@@ -0,0 +1,350 @@
+/*
+ *
+ * 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.v0_10;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.StringTokenizer;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
+import org.apache.qpid.common.QpidProperties;
+import org.apache.qpid.common.ServerPropertyNames;
+import org.apache.qpid.properties.ConnectionStartProperties;
+import org.apache.qpid.server.configuration.BrokerProperties;
+import org.apache.qpid.server.model.Broker;
+import org.apache.qpid.server.protocol.AMQConnectionModel;
+import org.apache.qpid.server.security.SecurityManager;
+import org.apache.qpid.server.security.SubjectCreator;
+import org.apache.qpid.server.security.auth.AuthenticationResult.AuthenticationStatus;
+import org.apache.qpid.server.security.auth.SubjectAuthenticationResult;
+import org.apache.qpid.server.virtualhost.State;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.transport.*;
+import org.apache.qpid.transport.network.NetworkConnection;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.qpid.transport.Connection.State.CLOSE_RCVD;
+
+public class ServerConnectionDelegate extends ServerDelegate
+{
+    private static final Logger LOGGER = LoggerFactory.getLogger(ServerConnectionDelegate.class);
+
+    private final Broker _broker;
+    private final String _localFQDN;
+    private int _maxNoOfChannels;
+    private Map<String,Object> _clientProperties;
+    private final SubjectCreator _subjectCreator;
+
+    public ServerConnectionDelegate(Broker broker, String localFQDN, SubjectCreator subjectCreator)
+    {
+        this(createConnectionProperties(broker), Collections.singletonList((Object)"en_US"), broker, localFQDN, subjectCreator);
+    }
+
+    private ServerConnectionDelegate(Map<String, Object> properties,
+                                    List<Object> locales,
+                                    Broker broker,
+                                    String localFQDN,
+                                    SubjectCreator subjectCreator)
+    {
+        super(properties, parseToList(subjectCreator.getMechanisms()), locales);
+
+        _broker = broker;
+        _localFQDN = localFQDN;
+        _maxNoOfChannels = (Integer)broker.getAttribute(Broker.CONNECTION_SESSION_COUNT_LIMIT);
+        _subjectCreator = subjectCreator;
+    }
+
+    private static List<String> getFeatures(Broker broker)
+    {
+        String brokerDisabledFeatures = System.getProperty(BrokerProperties.PROPERTY_DISABLED_FEATURES);
+        final List<String> features = new ArrayList<String>();
+        if (brokerDisabledFeatures == null || !brokerDisabledFeatures.contains(ServerPropertyNames.FEATURE_QPID_JMS_SELECTOR))
+        {
+            features.add(ServerPropertyNames.FEATURE_QPID_JMS_SELECTOR);
+        }
+
+        return Collections.unmodifiableList(features);
+    }
+
+    private static Map<String, Object> createConnectionProperties(final Broker broker)
+    {
+        final Map<String,Object> map = new HashMap<String,Object>();
+        // Federation tag is used by the client to identify the broker instance
+        map.put(ServerPropertyNames.FEDERATION_TAG, broker.getId().toString());
+        final List<String> features = getFeatures(broker);
+        if (features != null && features.size() > 0)
+        {
+            map.put(ServerPropertyNames.QPID_FEATURES, features);
+        }
+
+        map.put(ServerPropertyNames.PRODUCT, QpidProperties.getProductName());
+        map.put(ServerPropertyNames.VERSION, QpidProperties.getReleaseVersion());
+        map.put(ServerPropertyNames.QPID_BUILD, QpidProperties.getBuildVersion());
+        map.put(ServerPropertyNames.QPID_INSTANCE_NAME, broker.getName());
+
+        return map;
+    }
+
+    private static List<Object> parseToList(String mechanisms)
+    {
+        List<Object> list = new ArrayList<Object>();
+        StringTokenizer tokenizer = new StringTokenizer(mechanisms, " ");
+        while(tokenizer.hasMoreTokens())
+        {
+            list.add(tokenizer.nextToken());
+        }
+        return list;
+    }
+
+    public ServerSession getSession(Connection conn, SessionAttach atc)
+    {
+        SessionDelegate serverSessionDelegate = new ServerSessionDelegate();
+
+        ServerSession ssn = new ServerSession(conn, serverSessionDelegate,  new Binary(atc.getName()), 0);
+
+        return ssn;
+    }
+
+    protected SaslServer createSaslServer(Connection conn, String mechanism) throws SaslException
+    {
+        return _subjectCreator.createSaslServer(mechanism, _localFQDN, ((ServerConnection) conn).getPeerPrincipal());
+
+    }
+
+    protected void secure(final SaslServer ss, final Connection conn, final byte[] response)
+    {
+        final ServerConnection sconn = (ServerConnection) conn;
+        final SubjectAuthenticationResult authResult = _subjectCreator.authenticate(ss, response);
+
+        if (AuthenticationStatus.SUCCESS.equals(authResult.getStatus()))
+        {
+            tuneAuthorizedConnection(sconn);
+            sconn.setAuthorizedSubject(authResult.getSubject());
+        }
+        else if (AuthenticationStatus.CONTINUE.equals(authResult.getStatus()))
+        {
+            connectionAuthContinue(sconn, authResult.getChallenge());
+        }
+        else
+        {
+            connectionAuthFailed(sconn, authResult.getCause());
+        }
+    }
+
+    @Override
+    public void connectionClose(Connection conn, ConnectionClose close)
+    {
+        final ServerConnection sconn = (ServerConnection) conn;
+        try
+        {
+            sconn.logClosed();
+        }
+        finally
+        {
+            sconn.closeCode(close);
+            sconn.setState(CLOSE_RCVD);
+            sendConnectionCloseOkAndCloseSender(conn);
+        }
+    }
+
+    public void connectionOpen(Connection conn, ConnectionOpen open)
+    {
+        final ServerConnection sconn = (ServerConnection) conn;
+
+        VirtualHost vhost;
+        String vhostName;
+        if(open.hasVirtualHost())
+        {
+            vhostName = open.getVirtualHost();
+        }
+        else
+        {
+            vhostName = "";
+        }
+        vhost = _broker.getVirtualHostRegistry().getVirtualHost(vhostName);
+
+        SecurityManager.setThreadSubject(sconn.getAuthorizedSubject());
+
+        if(vhost != null)
+        {
+            sconn.setVirtualHost(vhost);
+
+            if (!vhost.getSecurityManager().accessVirtualhost(vhostName, sconn.getRemoteAddress()))
+            {
+                sconn.setState(Connection.State.CLOSING);
+                sconn.invoke(new ConnectionClose(ConnectionCloseCode.CONNECTION_FORCED, "Permission denied '"+vhostName+"'"));
+            }
+            else if (vhost.getState() != State.ACTIVE)
+            {
+                sconn.setState(Connection.State.CLOSING);
+                sconn.invoke(new ConnectionClose(ConnectionCloseCode.CONNECTION_FORCED, "Virtual host '"+vhostName+"' is not active"));
+            }
+            else
+            {
+                sconn.setState(Connection.State.OPEN);
+                sconn.invoke(new ConnectionOpenOk(Collections.emptyList()));
+            }
+        }
+        else
+        {
+            sconn.setState(Connection.State.CLOSING);
+            sconn.invoke(new ConnectionClose(ConnectionCloseCode.INVALID_PATH, "Unknown virtualhost '"+vhostName+"'"));
+        }
+
+    }
+
+    @Override
+    public void connectionTuneOk(final Connection conn, final ConnectionTuneOk ok)
+    {
+        ServerConnection sconn = (ServerConnection) conn;
+        int okChannelMax = ok.getChannelMax();
+
+        if (okChannelMax > getChannelMax())
+        {
+            LOGGER.error("Connection '" + sconn.getConnectionId() + "' being severed, " +
+                    "client connectionTuneOk returned a channelMax (" + okChannelMax +
+                    ") above the server's offered limit (" + getChannelMax() +")");
+
+            //Due to the error we must forcefully close the connection without negotiation
+            sconn.getSender().close();
+            return;
+        }
+
+        final NetworkConnection networkConnection = sconn.getNetworkConnection();
+
+        if(ok.hasHeartbeat())
+        {
+            int heartbeat = ok.getHeartbeat();
+            if(heartbeat < 0)
+            {
+                heartbeat = 0;
+            }
+
+            networkConnection.setMaxReadIdle(2 * heartbeat);
+            networkConnection.setMaxWriteIdle(heartbeat);
+
+        }
+        else
+        {
+            networkConnection.setMaxReadIdle(0);
+            networkConnection.setMaxWriteIdle(0);
+        }
+
+        setConnectionTuneOkChannelMax(sconn, okChannelMax);
+    }
+
+    @Override
+    public int getChannelMax()
+    {
+        return _maxNoOfChannels;
+    }
+
+    protected void setChannelMax(int channelMax)
+    {
+        _maxNoOfChannels = channelMax;
+    }
+
+    @Override public void sessionDetach(Connection conn, SessionDetach dtc)
+    {
+        // To ensure a clean detach, we stop any remaining subscriptions. Stop ensures
+        // that any in-progress delivery (SubFlushRunner/QueueRunner) is completed before the stop
+        // completes.
+        stopAllSubscriptions(conn, dtc);
+        Session ssn = conn.getSession(dtc.getChannel());
+        ((ServerSession)ssn).setClose(true);
+        super.sessionDetach(conn, dtc);
+    }
+
+    private void stopAllSubscriptions(Connection conn, SessionDetach dtc)
+    {
+        final ServerSession ssn = (ServerSession) conn.getSession(dtc.getChannel());
+        final Collection<Subscription_0_10> subs = ssn.getSubscriptions();
+        for (Subscription_0_10 subscription_0_10 : subs)
+        {
+            subscription_0_10.stop();
+        }
+    }
+
+
+    @Override
+    public void sessionAttach(final Connection conn, final SessionAttach atc)
+    {
+        final Session ssn;
+
+        if(isSessionNameUnique(atc.getName(), conn))
+        {
+            super.sessionAttach(conn, atc);
+        }
+        else
+        {
+            ssn = getSession(conn, atc);
+            ssn.invoke(new SessionDetached(atc.getName(), SessionDetachCode.SESSION_BUSY));
+            ssn.closed();
+        }
+    }
+
+    private boolean isSessionNameUnique(final byte[] name, final Connection conn)
+    {
+        final ServerConnection sconn = (ServerConnection) conn;
+        final String userId = sconn.getUserName();
+
+        final Iterator<AMQConnectionModel> connections =
+                        ((ServerConnection)conn).getVirtualHost().getConnectionRegistry().getConnections().iterator();
+        while(connections.hasNext())
+        {
+            final AMQConnectionModel amqConnectionModel = (AMQConnectionModel) connections.next();
+            if (userId.equals(amqConnectionModel.getUserName()) && !amqConnectionModel.isSessionNameUnique(name))
+            {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    @Override
+    public void connectionStartOk(Connection conn, ConnectionStartOk ok)
+    {
+        _clientProperties = ok.getClientProperties();
+        super.connectionStartOk(conn, ok);
+    }
+
+    public Map<String,Object> getClientProperties()
+    {
+        return _clientProperties;
+    }
+
+    public String getClientId()
+    {
+        return _clientProperties == null ? null : (String) _clientProperties.get(ConnectionStartProperties.CLIENT_ID_0_10);
+    }
+
+    public String getClientVersion()
+    {
+        return _clientProperties == null ? null : (String) _clientProperties.get(ConnectionStartProperties.VERSION_0_10);
+    }
+}
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
new file mode 100644
index 0000000000..abe784cefa
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
@@ -0,0 +1,1000 @@
+/*
+ *
+ * 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.v0_10;
+
+import java.security.Principal;
+import java.text.MessageFormat;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.SortedMap;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+
+import javax.security.auth.Subject;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.AMQStoreException;
+import org.apache.qpid.protocol.AMQConstant;
+import org.apache.qpid.server.TransactionTimeoutHelper;
+import org.apache.qpid.server.TransactionTimeoutHelper.CloseAction;
+import org.apache.qpid.server.logging.LogActor;
+import org.apache.qpid.server.logging.LogMessage;
+import org.apache.qpid.server.logging.LogSubject;
+import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.logging.actors.GenericActor;
+import org.apache.qpid.server.logging.messages.ChannelMessages;
+import org.apache.qpid.server.logging.subjects.ChannelLogSubject;
+import org.apache.qpid.server.message.InboundMessage;
+import org.apache.qpid.server.message.MessageReference;
+import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.protocol.AMQConnectionModel;
+import org.apache.qpid.server.protocol.AMQSessionModel;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.BaseQueue;
+import org.apache.qpid.server.queue.QueueEntry;
+import org.apache.qpid.server.security.AuthorizationHolder;
+import org.apache.qpid.server.store.MessageStore;
+import org.apache.qpid.server.store.StoreFuture;
+import org.apache.qpid.server.txn.AlreadyKnownDtxException;
+import org.apache.qpid.server.txn.AsyncAutoCommitTransaction;
+import org.apache.qpid.server.txn.DistributedTransaction;
+import org.apache.qpid.server.txn.DtxNotSelectedException;
+import org.apache.qpid.server.txn.IncorrectDtxStateException;
+import org.apache.qpid.server.txn.JoinAndResumeDtxException;
+import org.apache.qpid.server.txn.LocalTransaction;
+import org.apache.qpid.server.txn.NotAssociatedDtxException;
+import org.apache.qpid.server.txn.RollbackOnlyDtxException;
+import org.apache.qpid.server.txn.ServerTransaction;
+import org.apache.qpid.server.txn.SuspendAndFailDtxException;
+import org.apache.qpid.server.txn.TimeoutDtxException;
+import org.apache.qpid.server.txn.UnknownDtxBranchException;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.transport.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.CHANNEL_FORMAT;
+import static org.apache.qpid.util.Serial.gt;
+
+public class ServerSession extends Session
+        implements AuthorizationHolder,
+                   AMQSessionModel, LogSubject, AsyncAutoCommitTransaction.FutureRecorder
+{
+    private static final Logger _logger = LoggerFactory.getLogger(ServerSession.class);
+
+    private static final String NULL_DESTINTATION = UUID.randomUUID().toString();
+    private static final int PRODUCER_CREDIT_TOPUP_THRESHOLD = 1 << 30;
+    private static final int UNFINISHED_COMMAND_QUEUE_THRESHOLD = 500;
+
+    private final UUID _id = UUID.randomUUID();
+    private long _createTime = System.currentTimeMillis();
+    private LogActor _actor = GenericActor.getInstance(this);
+
+    private final Set<Object> _blockingEntities = Collections.synchronizedSet(new HashSet<Object>());
+
+    private final AtomicBoolean _blocking = new AtomicBoolean(false);
+    private ChannelLogSubject _logSubject;
+    private final AtomicInteger _outstandingCredit = new AtomicInteger(UNLIMITED_CREDIT);
+
+    public static interface MessageDispositionChangeListener
+    {
+        public void onAccept();
+
+        public void onRelease(boolean setRedelivered);
+
+        public void onReject();
+
+        public boolean acquire();
+
+
+    }
+
+    public static interface Task
+    {
+        public void doTask(ServerSession session);
+    }
+
+
+    private final SortedMap<Integer, MessageDispositionChangeListener> _messageDispositionListenerMap =
+            new ConcurrentSkipListMap<Integer, MessageDispositionChangeListener>();
+
+    private ServerTransaction _transaction;
+
+    private final AtomicLong _txnStarts = new AtomicLong(0);
+    private final AtomicLong _txnCommits = new AtomicLong(0);
+    private final AtomicLong _txnRejects = new AtomicLong(0);
+    private final AtomicLong _txnCount = new AtomicLong(0);
+
+    private Map<String, Subscription_0_10> _subscriptions = new ConcurrentHashMap<String, Subscription_0_10>();
+
+    private final List<Task> _taskList = new CopyOnWriteArrayList<Task>();
+
+    private final TransactionTimeoutHelper _transactionTimeoutHelper;
+
+    private AtomicReference<LogMessage> _forcedCloseLogMessage = new AtomicReference<LogMessage>();
+
+    public ServerSession(Connection connection, SessionDelegate delegate, Binary name, long expiry)
+    {
+        super(connection, delegate, name, expiry);
+        _transaction = new AsyncAutoCommitTransaction(this.getMessageStore(),this);
+        _logSubject = new ChannelLogSubject(this);
+
+        _transactionTimeoutHelper = new TransactionTimeoutHelper(_logSubject, new CloseAction()
+        {
+            @Override
+            public void doTimeoutAction(String reason) throws AMQException
+            {
+                getConnectionModel().closeSession(ServerSession.this, AMQConstant.RESOURCE_ERROR, reason);
+            }
+        });
+    }
+
+    protected void setState(State state)
+    {
+        super.setState(state);
+
+        if (state == State.OPEN)
+        {
+            _actor.message(ChannelMessages.CREATE());
+            if(_blocking.get())
+            {
+                invokeBlock();
+            }
+        }
+    }
+
+    private void invokeBlock()
+    {
+        invoke(new MessageSetFlowMode("", MessageFlowMode.CREDIT));
+        invoke(new MessageStop(""));
+    }
+
+    @Override
+    protected boolean isFull(int id)
+    {
+        return isCommandsFull(id);
+    }
+
+    public void enqueue(final ServerMessage message, final List<? extends BaseQueue> queues)
+    {
+        if(_outstandingCredit.get() != UNLIMITED_CREDIT
+                && _outstandingCredit.decrementAndGet() == (Integer.MAX_VALUE - PRODUCER_CREDIT_TOPUP_THRESHOLD))
+        {
+            _outstandingCredit.addAndGet(PRODUCER_CREDIT_TOPUP_THRESHOLD);
+            invoke(new MessageFlow("",MessageCreditUnit.MESSAGE, PRODUCER_CREDIT_TOPUP_THRESHOLD));
+        }
+        getConnectionModel().registerMessageReceived(message.getSize(), message.getArrivalTime());
+        PostEnqueueAction postTransactionAction = new PostEnqueueAction(queues, message, isTransactional()) ;
+        _transaction.enqueue(queues,message, postTransactionAction);
+        incrementOutstandingTxnsIfNecessary();
+    }
+
+
+    public void sendMessage(MessageTransfer xfr,
+                            Runnable postIdSettingAction)
+    {
+        getConnectionModel().registerMessageDelivered(xfr.getBodySize());
+        invoke(xfr, postIdSettingAction);
+    }
+
+    public void onMessageDispositionChange(MessageTransfer xfr, MessageDispositionChangeListener acceptListener)
+    {
+        _messageDispositionListenerMap.put(xfr.getId(), acceptListener);
+    }
+
+
+    private static interface MessageDispositionAction
+    {
+        void performAction(MessageDispositionChangeListener  listener);
+    }
+
+    public void accept(RangeSet ranges)
+    {
+        dispositionChange(ranges, new MessageDispositionAction()
+        {
+            public void performAction(MessageDispositionChangeListener listener)
+            {
+                listener.onAccept();
+            }
+        });
+    }
+
+
+    public void release(RangeSet ranges, final boolean setRedelivered)
+    {
+        dispositionChange(ranges, new MessageDispositionAction()
+                                      {
+                                          public void performAction(MessageDispositionChangeListener listener)
+                                          {
+                                              listener.onRelease(setRedelivered);
+                                          }
+                                      });
+    }
+
+    public void reject(RangeSet ranges)
+    {
+        dispositionChange(ranges, new MessageDispositionAction()
+                                      {
+                                          public void performAction(MessageDispositionChangeListener listener)
+                                          {
+                                              listener.onReject();
+                                          }
+                                      });
+    }
+
+    public RangeSet acquire(RangeSet transfers)
+    {
+        RangeSet acquired = RangeSetFactory.createRangeSet();
+
+        if(!_messageDispositionListenerMap.isEmpty())
+        {
+            Iterator<Integer> unacceptedMessages = _messageDispositionListenerMap.keySet().iterator();
+            Iterator<Range> rangeIter = transfers.iterator();
+
+            if(rangeIter.hasNext())
+            {
+                Range range = rangeIter.next();
+
+                while(range != null && unacceptedMessages.hasNext())
+                {
+                    int next = unacceptedMessages.next();
+                    while(gt(next, range.getUpper()))
+                    {
+                        if(rangeIter.hasNext())
+                        {
+                            range = rangeIter.next();
+                        }
+                        else
+                        {
+                            range = null;
+                            break;
+                        }
+                    }
+                    if(range != null && range.includes(next))
+                    {
+                        MessageDispositionChangeListener changeListener = _messageDispositionListenerMap.get(next);
+                        if(changeListener != null && changeListener.acquire())
+                        {
+                            acquired.add(next);
+                        }
+                    }
+
+
+                }
+
+            }
+
+
+        }
+
+        return acquired;
+    }
+
+    public void dispositionChange(RangeSet ranges, MessageDispositionAction action)
+    {
+        if(ranges != null)
+        {
+
+            if(ranges.size() == 1)
+            {
+                Range r = ranges.getFirst();
+                for(int i = r.getLower(); i <= r.getUpper(); i++)
+                {
+                    MessageDispositionChangeListener changeListener = _messageDispositionListenerMap.remove(i);
+                    if(changeListener != null)
+                    {
+                        action.performAction(changeListener);
+                    }
+                }
+            }
+            else if(!_messageDispositionListenerMap.isEmpty())
+            {
+                Iterator<Integer> unacceptedMessages = _messageDispositionListenerMap.keySet().iterator();
+                Iterator<Range> rangeIter = ranges.iterator();
+
+                if(rangeIter.hasNext())
+                {
+                    Range range = rangeIter.next();
+
+                    while(range != null && unacceptedMessages.hasNext())
+                    {
+                        int next = unacceptedMessages.next();
+                        while(gt(next, range.getUpper()))
+                        {
+                            if(rangeIter.hasNext())
+                            {
+                                range = rangeIter.next();
+                            }
+                            else
+                            {
+                                range = null;
+                                break;
+                            }
+                        }
+                        if(range != null && range.includes(next))
+                        {
+                            MessageDispositionChangeListener changeListener = _messageDispositionListenerMap.remove(next);
+                            action.performAction(changeListener);
+                        }
+
+
+                    }
+
+                }
+            }
+        }
+    }
+
+    public void removeDispositionListener(Method method)
+    {
+        _messageDispositionListenerMap.remove(method.getId());
+    }
+
+    public void onClose()
+    {
+        if(_transaction instanceof LocalTransaction)
+        {
+            _transaction.rollback();
+        }
+        else if(_transaction instanceof DistributedTransaction)
+        {
+            getVirtualHost().getDtxRegistry().endAssociations(this);
+        }
+
+        for(MessageDispositionChangeListener listener : _messageDispositionListenerMap.values())
+        {
+            listener.onRelease(true);
+        }
+        _messageDispositionListenerMap.clear();
+
+        for (Task task : _taskList)
+        {
+            task.doTask(this);
+        }
+
+        LogMessage operationalLoggingMessage = _forcedCloseLogMessage.get();
+        if (operationalLoggingMessage == null)
+        {
+            operationalLoggingMessage = ChannelMessages.CLOSE();
+        }
+        CurrentActor.get().message(getLogSubject(), operationalLoggingMessage);
+    }
+
+    @Override
+    protected void awaitClose()
+    {
+        // Broker shouldn't block awaiting close - thus do override this method to do nothing
+    }
+
+    public void acknowledge(final Subscription_0_10 sub, final QueueEntry entry)
+    {
+        _transaction.dequeue(entry.getQueue(), entry.getMessage(),
+                             new ServerTransaction.Action()
+                             {
+
+                                 public void postCommit()
+                                 {
+                                     sub.acknowledge(entry);
+                                 }
+
+                                 public void onRollback()
+                                 {
+                                     // The client has acknowledge the message and therefore have seen it.
+                                     // In the event of rollback, the message must be marked as redelivered.
+                                     entry.setRedelivered();
+                                     entry.release();
+                                 }
+                             });
+    }
+
+    public Collection<Subscription_0_10> getSubscriptions()
+    {
+        return _subscriptions.values();
+    }
+
+    public void register(String destination, Subscription_0_10 sub)
+    {
+        _subscriptions.put(destination == null ? NULL_DESTINTATION : destination, sub);
+    }
+
+    public Subscription_0_10 getSubscription(String destination)
+    {
+        return _subscriptions.get(destination == null ? NULL_DESTINTATION : destination);
+    }
+
+    public void unregister(Subscription_0_10 sub)
+    {
+        _subscriptions.remove(sub.getName());
+        try
+        {
+            sub.getSendLock();
+            AMQQueue queue = sub.getQueue();
+            if(queue != null)
+            {
+                queue.unregisterSubscription(sub);
+            }
+        }
+        catch (AMQException e)
+        {
+            // TODO
+            _logger.error("Failed to unregister subscription :" + e.getMessage(), e);
+        }
+        finally
+        {
+            sub.releaseSendLock();
+        }
+    }
+
+    public boolean isTransactional()
+    {
+        return _transaction.isTransactional();
+    }
+
+    public void selectTx()
+    {
+        _transaction = new LocalTransaction(this.getMessageStore());
+        _txnStarts.incrementAndGet();
+    }
+
+    public void selectDtx()
+    {
+        _transaction = new DistributedTransaction(this, getMessageStore(), getVirtualHost());
+
+    }
+
+
+    public void startDtx(Xid xid, boolean join, boolean resume)
+            throws JoinAndResumeDtxException,
+                   UnknownDtxBranchException,
+                   AlreadyKnownDtxException,
+                   DtxNotSelectedException
+    {
+        DistributedTransaction distributedTransaction = assertDtxTransaction();
+        distributedTransaction.start(xid, join, resume);
+    }
+
+
+    public void endDtx(Xid xid, boolean fail, boolean suspend)
+            throws NotAssociatedDtxException,
+            UnknownDtxBranchException,
+            DtxNotSelectedException,
+            SuspendAndFailDtxException, TimeoutDtxException
+    {
+        DistributedTransaction distributedTransaction = assertDtxTransaction();
+        distributedTransaction.end(xid, fail, suspend);
+    }
+
+
+    public long getTimeoutDtx(Xid xid)
+            throws UnknownDtxBranchException
+    {
+        return getVirtualHost().getDtxRegistry().getTimeout(xid);
+    }
+
+
+    public void setTimeoutDtx(Xid xid, long timeout)
+            throws UnknownDtxBranchException
+    {
+        getVirtualHost().getDtxRegistry().setTimeout(xid, timeout);
+    }
+
+
+    public void prepareDtx(Xid xid)
+            throws UnknownDtxBranchException,
+            IncorrectDtxStateException, AMQStoreException, RollbackOnlyDtxException, TimeoutDtxException
+    {
+        getVirtualHost().getDtxRegistry().prepare(xid);
+    }
+
+    public void commitDtx(Xid xid, boolean onePhase)
+            throws UnknownDtxBranchException,
+            IncorrectDtxStateException, AMQStoreException, RollbackOnlyDtxException, TimeoutDtxException
+    {
+        getVirtualHost().getDtxRegistry().commit(xid, onePhase);
+    }
+
+
+    public void rollbackDtx(Xid xid)
+            throws UnknownDtxBranchException,
+            IncorrectDtxStateException, AMQStoreException, TimeoutDtxException
+    {
+        getVirtualHost().getDtxRegistry().rollback(xid);
+    }
+
+
+    public void forgetDtx(Xid xid) throws UnknownDtxBranchException, IncorrectDtxStateException
+    {
+        getVirtualHost().getDtxRegistry().forget(xid);
+    }
+
+    public List<Xid> recoverDtx()
+    {
+        return getVirtualHost().getDtxRegistry().recover();
+    }
+
+    private DistributedTransaction assertDtxTransaction() throws DtxNotSelectedException
+    {
+        if(_transaction instanceof DistributedTransaction)
+        {
+            return (DistributedTransaction) _transaction;
+        }
+        else
+        {
+            throw new DtxNotSelectedException();
+        }
+    }
+
+
+    public void commit()
+    {
+        _transaction.commit();
+
+        _txnCommits.incrementAndGet();
+        _txnStarts.incrementAndGet();
+        decrementOutstandingTxnsIfNecessary();
+    }
+
+    public void rollback()
+    {
+        _transaction.rollback();
+
+        _txnRejects.incrementAndGet();
+        _txnStarts.incrementAndGet();
+        decrementOutstandingTxnsIfNecessary();
+    }
+
+
+    private void incrementOutstandingTxnsIfNecessary()
+    {
+        if(isTransactional())
+        {
+            //There can currently only be at most one outstanding transaction
+            //due to only having LocalTransaction support. Set value to 1 if 0.
+            _txnCount.compareAndSet(0,1);
+        }
+    }
+
+    private void decrementOutstandingTxnsIfNecessary()
+    {
+        if(isTransactional())
+        {
+            //There can currently only be at most one outstanding transaction
+            //due to only having LocalTransaction support. Set value to 0 if 1.
+            _txnCount.compareAndSet(1,0);
+        }
+    }
+
+    public Long getTxnCommits()
+    {
+        return _txnCommits.get();
+    }
+
+    public Long getTxnRejects()
+    {
+        return _txnRejects.get();
+    }
+
+    public int getChannelId()
+    {
+        return getChannel();
+    }
+
+    public Long getTxnCount()
+    {
+        return _txnCount.get();
+    }
+
+    public Long getTxnStart()
+    {
+        return _txnStarts.get();
+    }
+
+    public Principal getAuthorizedPrincipal()
+    {
+        return getConnection().getAuthorizedPrincipal();
+    }
+
+    public Subject getAuthorizedSubject()
+    {
+        return getConnection().getAuthorizedSubject();
+    }
+
+    public void addSessionCloseTask(Task task)
+    {
+        _taskList.add(task);
+    }
+
+    public void removeSessionCloseTask(Task task)
+    {
+        _taskList.remove(task);
+    }
+
+    public Object getReference()
+    {
+        return getConnection().getReference();
+    }
+
+    public MessageStore getMessageStore()
+    {
+        return getVirtualHost().getMessageStore();
+    }
+
+    public VirtualHost getVirtualHost()
+    {
+        return getConnection().getVirtualHost();
+    }
+
+    public boolean isDurable()
+    {
+        return false;
+    }
+
+
+    public long getCreateTime()
+    {
+        return _createTime;
+    }
+
+    @Override
+    public UUID getId()
+    {
+        return _id;
+    }
+
+    public AMQConnectionModel getConnectionModel()
+    {
+        return getConnection();
+    }
+
+    public String getClientID()
+    {
+        return getConnection().getClientId();
+    }
+
+    @Override
+    public ServerConnection getConnection()
+    {
+        return (ServerConnection) super.getConnection();
+    }
+
+    public LogActor getLogActor()
+    {
+        return _actor;
+    }
+
+    public LogSubject getLogSubject()
+    {
+        return (LogSubject) this;
+    }
+
+    public void checkTransactionStatus(long openWarn, long openClose, long idleWarn, long idleClose) throws AMQException
+    {
+        _transactionTimeoutHelper.checkIdleOrOpenTimes(_transaction, openWarn, openClose, idleWarn, idleClose);
+    }
+
+    public void block(AMQQueue queue)
+    {
+        block(queue, queue.getName());
+    }
+
+    public void block()
+    {
+        block(this, "** All Queues **");
+    }
+
+
+    private void block(Object queue, String name)
+    {
+        synchronized (_blockingEntities)
+        {
+            if(_blockingEntities.add(queue))
+            {
+
+                if(_blocking.compareAndSet(false,true))
+                {
+                    if(getState() == State.OPEN)
+                    {
+                        invokeBlock();
+                    }
+                    _actor.message(_logSubject, ChannelMessages.FLOW_ENFORCED(name));
+                }
+
+
+            }
+        }
+    }
+
+    public void unblock(AMQQueue queue)
+    {
+        unblock((Object)queue);
+    }
+
+    public void unblock()
+    {
+        unblock(this);
+    }
+
+    private void unblock(Object queue)
+    {
+        synchronized(_blockingEntities)
+        {
+            if(_blockingEntities.remove(queue) && _blockingEntities.isEmpty())
+            {
+                if(_blocking.compareAndSet(true,false) && !isClosing())
+                {
+
+                    _actor.message(_logSubject, ChannelMessages.FLOW_REMOVED());
+                    MessageFlow mf = new MessageFlow();
+                    mf.setUnit(MessageCreditUnit.MESSAGE);
+                    mf.setDestination("");
+                    _outstandingCredit.set(Integer.MAX_VALUE);
+                    mf.setValue(Integer.MAX_VALUE);
+                    invoke(mf);
+
+
+                }
+            }
+        }
+    }
+
+    public boolean onSameConnection(InboundMessage inbound)
+    {
+        return ((inbound instanceof MessageTransferMessage)
+                && ((MessageTransferMessage)inbound).getConnectionReference() == getConnection().getReference())
+                || ((inbound instanceof MessageMetaData_0_10)
+                    && (((MessageMetaData_0_10)inbound).getConnectionReference())== getConnection().getReference());
+    }
+
+
+    public String toLogString()
+    {
+        long connectionId = super.getConnection() instanceof ServerConnection
+                            ? getConnection().getConnectionId()
+                            : -1;
+
+        String remoteAddress = String.valueOf(getConnection().getRemoteAddress());
+        return "[" +
+               MessageFormat.format(CHANNEL_FORMAT,
+                                    connectionId,
+                                   getClientID(),
+                                   remoteAddress,
+                                   getVirtualHost().getName(),
+                                   getChannel())
+            + "] ";
+    }
+
+    @Override
+    public void close(AMQConstant cause, String message)
+    {
+        if (cause == null)
+        {
+            close();
+        }
+        else
+        {
+            close(cause.getCode(), message);
+        }
+    }
+
+    void close(int cause, String message)
+    {
+        _forcedCloseLogMessage.compareAndSet(null, ChannelMessages.CLOSE_FORCED(cause, message));
+        close();
+    }
+
+    @Override
+    public void close()
+    {
+        // unregister subscriptions in order to prevent sending of new messages
+        // to subscriptions with closing session
+        unregisterSubscriptions();
+        super.close();
+    }
+
+    void unregisterSubscriptions()
+    {
+        final Collection<Subscription_0_10> subscriptions = getSubscriptions();
+        for (Subscription_0_10 subscription_0_10 : subscriptions)
+        {
+            unregister(subscription_0_10);
+        }
+    }
+
+    void stopSubscriptions()
+    {
+        final Collection<Subscription_0_10> subscriptions = getSubscriptions();
+        for (Subscription_0_10 subscription_0_10 : subscriptions)
+        {
+            subscription_0_10.stop();
+        }
+    }
+
+
+    public void receivedComplete()
+    {
+        final Collection<Subscription_0_10> subscriptions = getSubscriptions();
+        for (Subscription_0_10 subscription_0_10 : subscriptions)
+        {
+            subscription_0_10.flushCreditState(false);
+        }
+        awaitCommandCompletion();
+    }
+
+    private class PostEnqueueAction implements ServerTransaction.Action
+    {
+
+        private List<? extends BaseQueue> _queues;
+        private ServerMessage _message;
+        private final boolean _transactional;
+
+        public PostEnqueueAction(List<? extends BaseQueue> queues, ServerMessage message, final boolean transactional)
+        {
+            _transactional = transactional;
+            setState(queues, message);
+        }
+
+        public void setState(List<? extends BaseQueue> queues, ServerMessage message)
+        {
+            _message = message;
+            _queues = queues;
+        }
+
+        public void postCommit()
+        {
+            MessageReference<?> ref = _message.newReference();
+            for(int i = 0; i < _queues.size(); i++)
+            {
+                try
+                {
+                    BaseQueue queue = _queues.get(i);
+                    queue.enqueue(_message, _transactional, null);
+                    if(queue instanceof AMQQueue)
+                    {
+                        ((AMQQueue)queue).checkCapacity(ServerSession.this);
+                    }
+
+                }
+                catch (AMQException e)
+                {
+                    // TODO
+                    throw new RuntimeException(e);
+                }
+            }
+            ref.release();
+        }
+
+        public void onRollback()
+        {
+            // NO-OP
+        }
+    }
+
+    public int getUnacknowledgedMessageCount()
+    {
+        return _messageDispositionListenerMap.size();
+    }
+
+    public boolean getBlocking()
+    {
+        return _blocking.get();
+    }
+
+    private final LinkedList<AsyncCommand> _unfinishedCommandsQueue = new LinkedList<AsyncCommand>();
+
+    public void completeAsyncCommands()
+    {
+        AsyncCommand cmd;
+        while((cmd = _unfinishedCommandsQueue.peek()) != null && cmd.isReadyForCompletion())
+        {
+            cmd.complete();
+            _unfinishedCommandsQueue.poll();
+        }
+        while(_unfinishedCommandsQueue.size() > UNFINISHED_COMMAND_QUEUE_THRESHOLD)
+        {
+            cmd = _unfinishedCommandsQueue.poll();
+            cmd.awaitReadyForCompletion();
+            cmd.complete();
+        }
+    }
+
+
+    public void awaitCommandCompletion()
+    {
+        AsyncCommand cmd;
+        while((cmd = _unfinishedCommandsQueue.poll()) != null)
+        {
+            cmd.awaitReadyForCompletion();
+            cmd.complete();
+        }
+    }
+
+
+    public Object getAsyncCommandMark()
+    {
+        return _unfinishedCommandsQueue.isEmpty() ? null : _unfinishedCommandsQueue.getLast();
+    }
+
+    public void recordFuture(final StoreFuture future, final ServerTransaction.Action action)
+    {
+        _unfinishedCommandsQueue.add(new AsyncCommand(future, action));
+    }
+
+    private static class AsyncCommand
+    {
+        private final StoreFuture _future;
+        private ServerTransaction.Action _action;
+
+        public AsyncCommand(final StoreFuture future, final ServerTransaction.Action action)
+        {
+            _future = future;
+            _action = action;
+        }
+
+        void awaitReadyForCompletion()
+        {
+            _future.waitForCompletion();
+        }
+
+        void complete()
+        {
+            if(!_future.isComplete())
+            {
+                _future.waitForCompletion();
+            }
+            _action.postCommit();
+            _action = null;
+        }
+
+        boolean isReadyForCompletion()
+        {
+            return _future.isComplete();
+        }
+    }
+
+    protected void setClose(boolean close)
+    {
+        super.setClose(close);
+    }
+
+    @Override
+    public int getConsumerCount()
+    {
+        return _subscriptions.values().size();
+    }
+
+    @Override
+    public int compareTo(AMQSessionModel o)
+    {
+        return getId().compareTo(o.getId());
+    }
+
+}
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java
new file mode 100644
index 0000000000..8e79813216
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java
@@ -0,0 +1,1582 @@
+/*
+ *
+ * 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.v0_10;
+
+import org.apache.log4j.Logger;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.AMQStoreException;
+import org.apache.qpid.AMQUnknownExchangeType;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.server.exchange.Exchange;
+import org.apache.qpid.server.exchange.ExchangeInUseException;
+import org.apache.qpid.server.exchange.HeadersExchange;
+import org.apache.qpid.server.filter.FilterManager;
+import org.apache.qpid.server.filter.FilterManagerFactory;
+import org.apache.qpid.server.logging.messages.ExchangeMessages;
+import org.apache.qpid.server.model.UUIDGenerator;
+import org.apache.qpid.server.plugin.ExchangeType;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.AMQQueueFactory;
+import org.apache.qpid.server.queue.BaseQueue;
+import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.security.SecurityManager;
+import org.apache.qpid.server.store.DurableConfigurationStoreHelper;
+import org.apache.qpid.server.store.DurableConfigurationStore;
+import org.apache.qpid.server.store.MessageStore;
+import org.apache.qpid.server.store.StoreFuture;
+import org.apache.qpid.server.store.StoredMessage;
+import org.apache.qpid.server.txn.AlreadyKnownDtxException;
+import org.apache.qpid.server.txn.DtxNotSelectedException;
+import org.apache.qpid.server.txn.IncorrectDtxStateException;
+import org.apache.qpid.server.txn.JoinAndResumeDtxException;
+import org.apache.qpid.server.txn.NotAssociatedDtxException;
+import org.apache.qpid.server.txn.RollbackOnlyDtxException;
+import org.apache.qpid.server.txn.ServerTransaction;
+import org.apache.qpid.server.txn.SuspendAndFailDtxException;
+import org.apache.qpid.server.txn.TimeoutDtxException;
+import org.apache.qpid.server.txn.UnknownDtxBranchException;
+import org.apache.qpid.server.virtualhost.ExchangeExistsException;
+import org.apache.qpid.server.virtualhost.ExchangeIsAlternateException;
+import org.apache.qpid.server.virtualhost.RequiredExchangeException;
+import org.apache.qpid.server.virtualhost.ReservedExchangeNameException;
+import org.apache.qpid.server.virtualhost.UnknownExchangeException;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.transport.*;
+
+import java.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+
+public class ServerSessionDelegate extends SessionDelegate
+{
+    private static final Logger LOGGER = Logger.getLogger(ServerSessionDelegate.class);
+
+    /**
+     * No-local queue argument is used to support the no-local feature of Durable Subscribers.
+     */
+    private static final String QUEUE_ARGUMENT_NO_LOCAL = "no-local";
+
+    public ServerSessionDelegate()
+    {
+
+    }
+
+    @Override
+    public void command(Session session, Method method)
+    {
+        try
+        {
+            setThreadSubject(session);
+
+            if(!session.isClosing())
+            {
+                Object asyncCommandMark = ((ServerSession)session).getAsyncCommandMark();
+                super.command(session, method, false);
+                Object newOutstanding = ((ServerSession)session).getAsyncCommandMark();
+                if(newOutstanding == null || newOutstanding == asyncCommandMark)
+                {
+                    session.processed(method);
+                }
+
+                if(newOutstanding != null)
+                {
+                    ((ServerSession)session).completeAsyncCommands();
+                }
+
+                if (method.isSync())
+                {
+                    ((ServerSession)session).awaitCommandCompletion();
+                    session.flushProcessed();
+                }
+            }
+        }
+        catch(RuntimeException e)
+        {
+            LOGGER.error("Exception processing command", e);
+            exception(session, method, ExecutionErrorCode.INTERNAL_ERROR, "Exception processing command: " + e);
+        }
+    }
+
+    @Override
+    public void messageAccept(Session session, MessageAccept method)
+    {
+        final ServerSession serverSession = (ServerSession) session;
+        serverSession.accept(method.getTransfers());
+        if(!serverSession.isTransactional())
+        {
+            serverSession.recordFuture(StoreFuture.IMMEDIATE_FUTURE,
+                                       new CommandProcessedAction(serverSession, method));
+        }
+    }
+
+    @Override
+    public void messageReject(Session session, MessageReject method)
+    {
+        ((ServerSession)session).reject(method.getTransfers());
+    }
+
+    @Override
+    public void messageRelease(Session session, MessageRelease method)
+    {
+        ((ServerSession)session).release(method.getTransfers(), method.getSetRedelivered());
+    }
+
+    @Override
+    public void messageAcquire(Session session, MessageAcquire method)
+    {
+        RangeSet acquiredRanges = ((ServerSession)session).acquire(method.getTransfers());
+
+        Acquired result = new Acquired(acquiredRanges);
+
+
+        session.executionResult((int) method.getId(), result);
+
+
+    }
+
+    @Override
+    public void messageResume(Session session, MessageResume method)
+    {
+        super.messageResume(session, method);
+    }
+
+    @Override
+    public void messageSubscribe(Session session, MessageSubscribe method)
+    {
+        /*
+          TODO - work around broken Python tests
+          Correct code should read like
+          if not hasAcceptMode() exception ILLEGAL_ARGUMENT "Accept-mode not supplied"
+          else if not method.hasAcquireMode() exception ExecutionErrorCode.ILLEGAL_ARGUMENT, "Acquire-mode not supplied"
+        */
+        if(!method.hasAcceptMode())
+        {
+            method.setAcceptMode(MessageAcceptMode.EXPLICIT);
+        }
+        if(!method.hasAcquireMode())
+        {
+            method.setAcquireMode(MessageAcquireMode.PRE_ACQUIRED);
+
+        }
+
+        if(!method.hasQueue())
+        {
+            exception(session, method, ExecutionErrorCode.ILLEGAL_ARGUMENT, "queue not supplied");
+        }
+        else
+        {
+            String destination = method.getDestination();
+
+            if(((ServerSession)session).getSubscription(destination)!=null)
+            {
+                exception(session, method, ExecutionErrorCode.NOT_ALLOWED, "Subscription already exists with destination '"+destination+"'");
+            }
+            else
+            {
+                String queueName = method.getQueue();
+                QueueRegistry queueRegistry = getQueueRegistry(session);
+
+
+                final AMQQueue queue = queueRegistry.getQueue(queueName);
+
+                if(queue == null)
+                {
+                    exception(session,method,ExecutionErrorCode.NOT_FOUND, "Queue: " + queueName + " not found");
+                }
+                else if(queue.getAuthorizationHolder() != null && queue.getAuthorizationHolder() != session)
+                {
+                    exception(session,method,ExecutionErrorCode.RESOURCE_LOCKED, "Exclusive Queue: " + queueName + " owned exclusively by another session");
+                }
+                else if(queue.isExclusive() && queue.getExclusiveOwningSession() != null && queue.getExclusiveOwningSession() != session)
+                {
+                    exception(session,method,ExecutionErrorCode.RESOURCE_LOCKED, "Exclusive Queue: " + queueName + " owned exclusively by another session");
+                }
+                else
+                {
+                    if(queue.isExclusive())
+                    {
+                        ServerSession s = (ServerSession) session;
+                        queue.setExclusiveOwningSession(s);
+                        if(queue.getAuthorizationHolder() == null)
+                        {
+                            queue.setAuthorizationHolder(s);
+                            queue.setExclusiveOwningSession(s);
+                            ((ServerSession) session).addSessionCloseTask(new ServerSession.Task()
+                            {
+                                public void doTask(ServerSession session)
+                                {
+                                    if(queue.getAuthorizationHolder() == session)
+                                    {
+                                        queue.setAuthorizationHolder(null);
+                                        queue.setExclusiveOwningSession(null);
+                                    }
+                                }
+                            });
+                        }
+                    }
+
+                    FlowCreditManager_0_10 creditManager = new WindowCreditManager(0L,0L);
+
+                    FilterManager filterManager = null;
+                    try
+                    {
+                        filterManager = FilterManagerFactory.createManager(method.getArguments());
+                    }
+                    catch (AMQException amqe)
+                    {
+                        exception(session, method, ExecutionErrorCode.ILLEGAL_ARGUMENT, "Exception Creating FilterManager");
+                        return;
+                    }
+
+                    Subscription_0_10 sub = new Subscription_0_10((ServerSession)session,
+                                                                  destination,
+                                                                  method.getAcceptMode(),
+                                                                  method.getAcquireMode(),
+                                                                  MessageFlowMode.WINDOW,
+                                                                  creditManager,
+                                                                  filterManager,
+                                                                  method.getArguments());
+
+                    ((ServerSession)session).register(destination, sub);
+                    try
+                    {
+                        queue.registerSubscription(sub, method.getExclusive());
+                    }
+                    catch (AMQQueue.ExistingExclusiveSubscription existing)
+                    {
+                        exception(session, method, ExecutionErrorCode.RESOURCE_LOCKED, "Queue has an exclusive consumer");
+                    }
+                    catch (AMQQueue.ExistingSubscriptionPreventsExclusive exclusive)
+                    {
+                        exception(session, method, ExecutionErrorCode.RESOURCE_LOCKED, "Queue has an existing consumer - can't subscribe exclusively");
+                    }
+                    catch (AMQException e)
+                    {
+                        exception(session, method, e, "Cannot subscribe to queue '" + queueName + "' with destination '" + destination);
+                    }
+                }
+            }
+        }
+    }
+
+    @Override
+    public void messageTransfer(Session ssn, final MessageTransfer xfr)
+    {
+        final Exchange exchange = getExchangeForMessage(ssn, xfr);
+
+        DeliveryProperties delvProps = null;
+        if(xfr.getHeader() != null && (delvProps = xfr.getHeader().getDeliveryProperties()) != null && delvProps.hasTtl() && !delvProps
+                .hasExpiration())
+        {
+            delvProps.setExpiration(System.currentTimeMillis() + delvProps.getTtl());
+        }
+
+        final MessageMetaData_0_10 messageMetaData = new MessageMetaData_0_10(xfr);
+        messageMetaData.setConnectionReference(((ServerSession)ssn).getReference());
+
+        if (!getVirtualHost(ssn).getSecurityManager().authorisePublish(messageMetaData.isImmediate(), messageMetaData.getRoutingKey(), exchange.getName()))
+        {
+            ExecutionErrorCode errorCode = ExecutionErrorCode.UNAUTHORIZED_ACCESS;
+            String description = "Permission denied: exchange-name '" + exchange.getName() + "'";
+            exception(ssn, xfr, errorCode, description);
+
+            return;
+        }
+
+        final Exchange exchangeInUse;
+        List<? extends BaseQueue> queues = exchange.route(messageMetaData);
+        if(queues.isEmpty() && exchange.getAlternateExchange() != null)
+        {
+            final Exchange alternateExchange = exchange.getAlternateExchange();
+            queues = alternateExchange.route(messageMetaData);
+            if (!queues.isEmpty())
+            {
+                exchangeInUse = alternateExchange;
+            }
+            else
+            {
+                exchangeInUse = exchange;
+            }
+        }
+        else
+        {
+            exchangeInUse = exchange;
+        }
+
+        final ServerSession serverSession = (ServerSession) ssn;
+        if(!queues.isEmpty())
+        {
+            final MessageStore store = getVirtualHost(ssn).getMessageStore();
+            final StoredMessage<MessageMetaData_0_10> storeMessage = createStoreMessage(xfr, messageMetaData, store);
+            MessageTransferMessage message = new MessageTransferMessage(storeMessage, serverSession.getReference());
+            serverSession.enqueue(message, queues);
+            storeMessage.flushToStore();
+        }
+        else
+        {
+            if((delvProps == null || !delvProps.getDiscardUnroutable()) && xfr.getAcceptMode() == MessageAcceptMode.EXPLICIT)
+            {
+                RangeSet rejects = RangeSetFactory.createRangeSet();
+                rejects.add(xfr.getId());
+                MessageReject reject = new MessageReject(rejects, MessageRejectCode.UNROUTABLE, "Unroutable");
+                ssn.invoke(reject);
+            }
+            else
+            {
+                serverSession.getLogActor().message(ExchangeMessages.DISCARDMSG(exchangeInUse.getName(), messageMetaData.getRoutingKey()));
+            }
+        }
+
+
+        if(serverSession.isTransactional())
+        {
+            serverSession.processed(xfr);
+        }
+        else
+        {
+            serverSession.recordFuture(StoreFuture.IMMEDIATE_FUTURE, new CommandProcessedAction(serverSession, xfr));
+        }
+    }
+
+    private StoredMessage<MessageMetaData_0_10> createStoreMessage(final MessageTransfer xfr,
+                                                                   final MessageMetaData_0_10 messageMetaData, final MessageStore store)
+    {
+        final StoredMessage<MessageMetaData_0_10> storeMessage = store.addMessage(messageMetaData);
+        ByteBuffer body = xfr.getBody();
+        if(body != null)
+        {
+            storeMessage.addContent(0, body);
+        }
+        return storeMessage;
+    }
+
+    @Override
+    public void messageCancel(Session session, MessageCancel method)
+    {
+        String destination = method.getDestination();
+
+        Subscription_0_10 sub = ((ServerSession)session).getSubscription(destination);
+
+        if(sub == null)
+        {
+            exception(session, method, ExecutionErrorCode.NOT_FOUND, "not-found: destination '"+destination+"'");
+        }
+        else
+        {
+            AMQQueue queue = sub.getQueue();
+            ((ServerSession)session).unregister(sub);
+            if(!queue.isDeleted() && queue.isExclusive() && queue.getConsumerCount() == 0)
+            {
+                queue.setAuthorizationHolder(null);
+            }
+        }
+    }
+
+    @Override
+    public void messageFlush(Session session, MessageFlush method)
+    {
+        String destination = method.getDestination();
+
+        Subscription_0_10 sub = ((ServerSession)session).getSubscription(destination);
+
+        if(sub == null)
+        {
+            exception(session, method, ExecutionErrorCode.NOT_FOUND, "not-found: destination '"+destination+"'");
+        }
+        else
+        {
+
+            try
+            {
+                sub.flush();
+            }
+            catch (AMQException e)
+            {
+                exception(session, method, e, "Cannot flush subscription '" + destination);
+            }
+        }
+    }
+
+    @Override
+    public void txSelect(Session session, TxSelect method)
+    {
+        // TODO - check current tx mode
+        ((ServerSession)session).selectTx();
+    }
+
+    @Override
+    public void txCommit(Session session, TxCommit method)
+    {
+        // TODO - check current tx mode
+        ((ServerSession)session).commit();
+    }
+
+    @Override
+    public void txRollback(Session session, TxRollback method)
+    {
+        // TODO - check current tx mode
+        ((ServerSession)session).rollback();
+    }
+
+    @Override
+    public void dtxSelect(Session session, DtxSelect method)
+    {
+        // TODO - check current tx mode
+        ((ServerSession)session).selectDtx();
+    }
+
+    @Override
+    public void dtxStart(Session session, DtxStart method)
+    {
+        XaResult result = new XaResult();
+        result.setStatus(DtxXaStatus.XA_OK);
+        try
+        {
+            ((ServerSession)session).startDtx(method.getXid(), method.getJoin(), method.getResume());
+            session.executionResult(method.getId(), result);
+        }
+        catch(JoinAndResumeDtxException e)
+        {
+            exception(session, method, ExecutionErrorCode.COMMAND_INVALID, e.getMessage());
+        }
+        catch(UnknownDtxBranchException e)
+        {
+            exception(session, method, ExecutionErrorCode.NOT_ALLOWED, "Unknown xid " + method.getXid());
+        }
+        catch(AlreadyKnownDtxException e)
+        {
+            exception(session, method, ExecutionErrorCode.NOT_ALLOWED, "Xid already started an neither join nor " +
+                                                                       "resume set" + method.getXid());
+        }
+        catch(DtxNotSelectedException e)
+        {
+            exception(session, method, ExecutionErrorCode.COMMAND_INVALID, e.getMessage());
+        }
+
+    }
+
+    @Override
+    public void dtxEnd(Session session, DtxEnd method)
+    {
+        XaResult result = new XaResult();
+        result.setStatus(DtxXaStatus.XA_OK);
+        try
+        {
+            try
+            {
+                ((ServerSession)session).endDtx(method.getXid(), method.getFail(), method.getSuspend());
+            }
+            catch (TimeoutDtxException e)
+            {
+                result.setStatus(DtxXaStatus.XA_RBTIMEOUT);
+            }
+            session.executionResult(method.getId(), result);
+        }
+        catch(UnknownDtxBranchException e)
+        {
+            exception(session, method, ExecutionErrorCode.ILLEGAL_STATE, e.getMessage());
+        }
+        catch(NotAssociatedDtxException e)
+        {
+            exception(session, method, ExecutionErrorCode.ILLEGAL_STATE, e.getMessage());
+        }
+        catch(DtxNotSelectedException e)
+        {
+            exception(session, method, ExecutionErrorCode.ILLEGAL_STATE, e.getMessage());
+        }
+        catch(SuspendAndFailDtxException e)
+        {
+            exception(session, method, ExecutionErrorCode.COMMAND_INVALID, e.getMessage());
+        }
+
+    }
+
+    @Override
+    public void dtxCommit(Session session, DtxCommit method)
+    {
+        XaResult result = new XaResult();
+        result.setStatus(DtxXaStatus.XA_OK);
+        try
+        {
+            try
+            {
+                ((ServerSession)session).commitDtx(method.getXid(), method.getOnePhase());
+            }
+            catch (RollbackOnlyDtxException e)
+            {
+                result.setStatus(DtxXaStatus.XA_RBROLLBACK);
+            }
+            catch (TimeoutDtxException e)
+            {
+                result.setStatus(DtxXaStatus.XA_RBTIMEOUT);
+            }
+            session.executionResult(method.getId(), result);
+        }
+        catch(UnknownDtxBranchException e)
+        {
+            exception(session, method, ExecutionErrorCode.NOT_FOUND, e.getMessage());
+        }
+        catch(IncorrectDtxStateException e)
+        {
+            exception(session, method, ExecutionErrorCode.ILLEGAL_STATE, e.getMessage());
+        }
+        catch(AMQStoreException e)
+        {
+            exception(session, method, ExecutionErrorCode.INTERNAL_ERROR, e.getMessage());
+        }
+    }
+
+    @Override
+    public void dtxForget(Session session, DtxForget method)
+    {
+        try
+        {
+            ((ServerSession)session).forgetDtx(method.getXid());
+        }
+        catch(UnknownDtxBranchException e)
+        {
+            exception(session, method, ExecutionErrorCode.NOT_FOUND, e.getMessage());
+        }
+        catch(IncorrectDtxStateException e)
+        {
+            exception(session, method, ExecutionErrorCode.ILLEGAL_STATE, e.getMessage());
+        }
+
+    }
+
+    @Override
+    public void dtxGetTimeout(Session session, DtxGetTimeout method)
+    {
+        GetTimeoutResult result = new GetTimeoutResult();
+        try
+        {
+            result.setTimeout(((ServerSession) session).getTimeoutDtx(method.getXid()));
+            session.executionResult(method.getId(), result);
+        }
+        catch(UnknownDtxBranchException e)
+        {
+            exception(session, method, ExecutionErrorCode.NOT_FOUND, e.getMessage());
+        }
+    }
+
+    @Override
+    public void dtxPrepare(Session session, DtxPrepare method)
+    {
+        XaResult result = new XaResult();
+        result.setStatus(DtxXaStatus.XA_OK);
+        try
+        {
+            try
+            {
+                ((ServerSession)session).prepareDtx(method.getXid());
+            }
+            catch (RollbackOnlyDtxException e)
+            {
+                result.setStatus(DtxXaStatus.XA_RBROLLBACK);
+            }
+            catch (TimeoutDtxException e)
+            {
+                result.setStatus(DtxXaStatus.XA_RBTIMEOUT);
+            }
+            session.executionResult((int) method.getId(), result);
+        }
+        catch(UnknownDtxBranchException e)
+        {
+            exception(session, method, ExecutionErrorCode.NOT_FOUND, e.getMessage());
+        }
+        catch(IncorrectDtxStateException e)
+        {
+            exception(session, method, ExecutionErrorCode.ILLEGAL_STATE, e.getMessage());
+        }
+        catch(AMQStoreException e)
+        {
+            exception(session, method, ExecutionErrorCode.INTERNAL_ERROR, e.getMessage());
+        }
+    }
+
+    @Override
+    public void dtxRecover(Session session, DtxRecover method)
+    {
+        RecoverResult result = new RecoverResult();
+        List inDoubt = ((ServerSession)session).recoverDtx();
+        result.setInDoubt(inDoubt);
+        session.executionResult(method.getId(), result);
+    }
+
+    @Override
+    public void dtxRollback(Session session, DtxRollback method)
+    {
+
+        XaResult result = new XaResult();
+        result.setStatus(DtxXaStatus.XA_OK);
+        try
+        {
+            try
+            {
+                ((ServerSession)session).rollbackDtx(method.getXid());
+            }
+            catch (TimeoutDtxException e)
+            {
+                result.setStatus(DtxXaStatus.XA_RBTIMEOUT);
+            }
+            session.executionResult(method.getId(), result);
+        }
+        catch(UnknownDtxBranchException e)
+        {
+            exception(session, method, ExecutionErrorCode.NOT_FOUND, e.getMessage());
+        }
+        catch(IncorrectDtxStateException e)
+        {
+            exception(session, method, ExecutionErrorCode.ILLEGAL_STATE, e.getMessage());
+        }
+        catch(AMQStoreException e)
+        {
+            exception(session, method, ExecutionErrorCode.INTERNAL_ERROR, e.getMessage());
+        }
+    }
+
+    @Override
+    public void dtxSetTimeout(Session session, DtxSetTimeout method)
+    {
+        try
+        {
+            ((ServerSession)session).setTimeoutDtx(method.getXid(), method.getTimeout());
+        }
+        catch(UnknownDtxBranchException e)
+        {
+            exception(session, method, ExecutionErrorCode.NOT_FOUND, e.getMessage());
+        }
+    }
+
+    @Override
+    public void executionSync(final Session ssn, final ExecutionSync sync)
+    {
+        ((ServerSession)ssn).awaitCommandCompletion();
+        super.executionSync(ssn, sync);
+    }
+
+    @Override
+    public void exchangeDeclare(Session session, ExchangeDeclare method)
+    {
+        String exchangeName = method.getExchange();
+        VirtualHost virtualHost = getVirtualHost(session);
+
+        //we must check for any unsupported arguments present and throw not-implemented
+        if(method.hasArguments())
+        {
+            Map<String,Object> args = method.getArguments();
+            //QPID-3392: currently we don't support any!
+            if(!args.isEmpty())
+            {
+                exception(session, method, ExecutionErrorCode.NOT_IMPLEMENTED, "Unsupported exchange argument(s) found " + args.keySet().toString());
+                return;
+            }
+        }
+
+        if(method.getPassive())
+        {
+            Exchange exchange = getExchange(session, exchangeName);
+
+            if(exchange == null)
+            {
+                exception(session, method, ExecutionErrorCode.NOT_FOUND, "not-found: exchange-name '" + exchangeName + "'");
+            }
+            else
+            {
+                if (!exchange.getTypeShortString().toString().equals(method.getType())
+                        && (method.getType() != null && method.getType().length() > 0))
+                {
+                    exception(session, method, ExecutionErrorCode.NOT_ALLOWED, "Attempt to redeclare exchange: "
+                            + exchangeName + " of type " + exchange.getTypeShortString() + " to " + method.getType() + ".");
+                }
+            }
+        }
+        else
+        {
+
+            try
+            {
+                virtualHost.createExchange(null,
+                        method.getExchange(),
+                        method.getType(),
+                        method.getDurable(),
+                        method.getAutoDelete(),
+                        method.getAlternateExchange());
+            }
+            catch(ReservedExchangeNameException e)
+            {
+                exception(session, method, ExecutionErrorCode.NOT_ALLOWED, "Attempt to declare exchange: "
+                                            + exchangeName + " which begins with reserved name or prefix.");
+            }
+            catch(UnknownExchangeException e)
+            {
+                exception(session, method, ExecutionErrorCode.NOT_FOUND,
+                                                            "Unknown alternate exchange " + e.getExchangeName());
+            }
+            catch(AMQUnknownExchangeType e)
+            {
+                exception(session, method, ExecutionErrorCode.NOT_FOUND, "Unknown Exchange Type: " + method.getType());
+            }
+            catch(ExchangeExistsException e)
+            {
+                Exchange exchange = e.getExistingExchange();
+                if(!exchange.getTypeShortString().toString().equals(method.getType()))
+                {
+                    exception(session, method, ExecutionErrorCode.NOT_ALLOWED,
+                            "Attempt to redeclare exchange: " + exchangeName
+                                    + " of type " + exchange.getTypeShortString()
+                                    + " to " + method.getType() +".");
+                }
+                else if(method.hasAlternateExchange()
+                          && (exchange.getAlternateExchange() == null ||
+                              !method.getAlternateExchange().equals(exchange.getAlternateExchange().getName())))
+                {
+                    exception(session, method, ExecutionErrorCode.NOT_ALLOWED,
+                            "Attempt to change alternate exchange of: " + exchangeName
+                                    + " from " + exchange.getAlternateExchange()
+                                    + " to " + method.getAlternateExchange() +".");
+                }
+            }
+            catch (AMQException e)
+            {
+                exception(session, method, e, "Cannot declare exchange '" + exchangeName);
+            }
+
+
+        }
+
+    }
+
+    // TODO decouple AMQException and AMQConstant error codes
+    private void exception(Session session, Method method, AMQException exception, String message)
+    {
+        ExecutionErrorCode errorCode = ExecutionErrorCode.INTERNAL_ERROR;
+        if (exception.getErrorCode() != null)
+        {
+            try
+            {
+                errorCode = ExecutionErrorCode.get(exception.getErrorCode().getCode());
+            }
+            catch (IllegalArgumentException iae)
+            {
+                // ignore, already set to INTERNAL_ERROR
+            }
+        }
+        String description = message + "': " + exception.getMessage();
+
+        exception(session, method, errorCode, description);
+    }
+
+    private void exception(Session session, Method method, ExecutionErrorCode errorCode, String description)
+    {
+        ExecutionException ex = new ExecutionException();
+        ex.setErrorCode(errorCode);
+        ex.setCommandId(method.getId());
+        ex.setDescription(description);
+
+        session.invoke(ex);
+
+        ((ServerSession)session).close(errorCode.getValue(), description);
+    }
+
+    private Exchange getExchange(Session session, String exchangeName)
+    {
+        return getVirtualHost(session).getExchange(exchangeName);
+    }
+
+    private Exchange getExchangeForMessage(Session ssn, MessageTransfer xfr)
+    {
+        VirtualHost virtualHost = getVirtualHost(ssn);
+
+        Exchange exchange;
+        if(xfr.hasDestination())
+        {
+            exchange = virtualHost.getExchange(xfr.getDestination());
+            if(exchange == null)
+            {
+                exchange = virtualHost.getDefaultExchange();
+            }
+        }
+        else
+        {
+            exchange = virtualHost.getDefaultExchange();
+        }
+        return exchange;
+    }
+
+    private VirtualHost getVirtualHost(Session session)
+    {
+        ServerConnection conn = getServerConnection(session);
+        VirtualHost vhost = conn.getVirtualHost();
+        return vhost;
+    }
+
+    private ServerConnection getServerConnection(Session session)
+    {
+        ServerConnection conn = (ServerConnection) session.getConnection();
+        return conn;
+    }
+
+    @Override
+    public void exchangeDelete(Session session, ExchangeDelete method)
+    {
+        VirtualHost virtualHost = getVirtualHost(session);
+
+        try
+        {
+            if (nameNullOrEmpty(method.getExchange()))
+            {
+                exception(session, method, ExecutionErrorCode.INVALID_ARGUMENT, "Delete not allowed for default exchange");
+                return;
+            }
+
+            Exchange exchange = getExchange(session, method.getExchange());
+
+            if(exchange == null)
+            {
+                exception(session, method, ExecutionErrorCode.NOT_FOUND, "No such exchange '" + method.getExchange() + "'");
+            }
+            else
+            {
+                virtualHost.removeExchange(exchange, !method.getIfUnused());
+            }
+        }
+        catch (ExchangeInUseException e)
+        {
+            exception(session, method, ExecutionErrorCode.PRECONDITION_FAILED, "Exchange in use");
+        }
+        catch (ExchangeIsAlternateException e)
+        {
+            exception(session, method, ExecutionErrorCode.NOT_ALLOWED, "Exchange in use as an alternate exchange");
+        }
+        catch (RequiredExchangeException e)
+        {
+            exception(session, method, ExecutionErrorCode.NOT_ALLOWED, "Exchange '"+method.getExchange()+"' cannot be deleted");
+        }
+        catch (AMQException e)
+        {
+            exception(session, method, e, "Cannot delete exchange '" + method.getExchange() );
+        }
+    }
+
+    private boolean nameNullOrEmpty(String name)
+    {
+        if(name == null || name.length() == 0)
+        {
+            return true;
+        }
+
+        return false;
+    }
+
+    private boolean isStandardExchange(Exchange exchange, Collection<ExchangeType<? extends Exchange>> registeredTypes)
+    {
+        for(ExchangeType type : registeredTypes)
+        {
+            if(type.getDefaultExchangeName().toString().equals( exchange.getName() ))
+            {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    @Override
+    public void exchangeQuery(Session session, ExchangeQuery method)
+    {
+
+        ExchangeQueryResult result = new ExchangeQueryResult();
+
+        Exchange exchange = getExchange(session, method.getName());
+
+        if(exchange != null)
+        {
+            result.setDurable(exchange.isDurable());
+            result.setType(exchange.getTypeShortString().toString());
+            result.setNotFound(false);
+        }
+        else
+        {
+            result.setNotFound(true);
+        }
+
+        session.executionResult((int) method.getId(), result);
+    }
+
+    @Override
+    public void exchangeBind(Session session, ExchangeBind method)
+    {
+
+        VirtualHost virtualHost = getVirtualHost(session);
+        QueueRegistry queueRegistry = virtualHost.getQueueRegistry();
+
+        if (!method.hasQueue())
+        {
+            exception(session, method, ExecutionErrorCode.ILLEGAL_ARGUMENT, "queue not set");
+        }
+        else if (nameNullOrEmpty(method.getExchange()))
+        {
+            exception(session, method, ExecutionErrorCode.INVALID_ARGUMENT, "Bind not allowed for default exchange");
+        }
+        else
+        {
+            //TODO - here because of non-compiant python tests
+            // should raise exception ILLEGAL_ARGUMENT "binding-key not set"
+            if (!method.hasBindingKey())
+            {
+                method.setBindingKey(method.getQueue());
+            }
+            AMQQueue queue = queueRegistry.getQueue(method.getQueue());
+            Exchange exchange = virtualHost.getExchange(method.getExchange());
+            if(queue == null)
+            {
+                exception(session, method, ExecutionErrorCode.NOT_FOUND, "Queue: '" + method.getQueue() + "' not found");
+            }
+            else if(exchange == null)
+            {
+                exception(session, method, ExecutionErrorCode.NOT_FOUND, "Exchange: '" + method.getExchange() + "' not found");
+            }
+            else if(exchange.getTypeShortString().equals(HeadersExchange.TYPE.getName()) && (!method.hasArguments() || method.getArguments() == null || !method.getArguments().containsKey("x-match")))
+            {
+                exception(session, method, ExecutionErrorCode.INTERNAL_ERROR, "Bindings to an exchange of type " + HeadersExchange.TYPE.getName() + " require an x-match header");
+            }
+            else
+            {
+                if (!exchange.isBound(method.getBindingKey(), method.getArguments(), queue))
+                {
+                    try
+                    {
+                        exchange.addBinding(method.getBindingKey(), queue, method.getArguments());
+                    }
+                    catch (AMQException e)
+                    {
+                        exception(session, method, e, "Cannot add binding '" + method.getBindingKey());
+                    }
+                }
+                else
+                {
+                    // todo
+                }
+            }
+
+
+        }
+
+
+
+    }
+
+    @Override
+    public void exchangeUnbind(Session session, ExchangeUnbind method)
+    {
+        VirtualHost virtualHost = getVirtualHost(session);
+        QueueRegistry queueRegistry = virtualHost.getQueueRegistry();
+
+        if (!method.hasQueue())
+        {
+            exception(session, method, ExecutionErrorCode.ILLEGAL_ARGUMENT, "queue not set");
+        }
+        else if (nameNullOrEmpty(method.getExchange()))
+        {
+            exception(session, method, ExecutionErrorCode.INVALID_ARGUMENT, "Unbind not allowed for default exchange");
+        }
+        else if (!method.hasBindingKey())
+        {
+            exception(session, method, ExecutionErrorCode.ILLEGAL_ARGUMENT, "binding-key not set");
+        }
+        else
+        {
+            AMQQueue queue = queueRegistry.getQueue(method.getQueue());
+            Exchange exchange = virtualHost.getExchange(method.getExchange());
+            if(queue == null)
+            {
+                exception(session, method, ExecutionErrorCode.NOT_FOUND, "Queue: '" + method.getQueue() + "' not found");
+            }
+            else if(exchange == null)
+            {
+                exception(session, method, ExecutionErrorCode.NOT_FOUND, "Exchange: '" + method.getExchange() + "' not found");
+            }
+            else
+            {
+                try
+                {
+                    exchange.removeBinding(method.getBindingKey(), queue, null);
+                }
+                catch (AMQException e)
+                {
+                    exception(session, method, e, "Cannot remove binding '" + method.getBindingKey());
+                }
+            }
+        }
+    }
+
+    @Override
+    public void exchangeBound(Session session, ExchangeBound method)
+    {
+
+        ExchangeBoundResult result = new ExchangeBoundResult();
+        VirtualHost virtualHost = getVirtualHost(session);
+        Exchange exchange;
+        AMQQueue queue;
+        if(method.hasExchange())
+        {
+            exchange = virtualHost.getExchange(method.getExchange());
+
+            if(exchange == null)
+            {
+                result.setExchangeNotFound(true);
+            }
+        }
+        else
+        {
+            exchange = virtualHost.getDefaultExchange();
+        }
+
+
+        if(method.hasQueue())
+        {
+
+            queue = getQueue(session, method.getQueue());
+            if(queue == null)
+            {
+                result.setQueueNotFound(true);
+            }
+
+
+            if(exchange != null && queue != null)
+            {
+
+                boolean queueMatched = exchange.isBound(queue);
+
+                result.setQueueNotMatched(!queueMatched);
+
+
+                if(method.hasBindingKey())
+                {
+
+                    if(queueMatched)
+                    {
+                        final boolean keyMatched = exchange.isBound(method.getBindingKey(), queue);
+                        result.setKeyNotMatched(!keyMatched);
+                        if(method.hasArguments())
+                        {
+                            if(keyMatched)
+                            {
+                                result.setArgsNotMatched(!exchange.isBound(method.getBindingKey(), method.getArguments(), queue));
+                            }
+                            else
+                            {
+                                result.setArgsNotMatched(!exchange.isBound(method.getArguments(), queue));
+                            }
+                        }
+                    }
+                    else
+                    {
+                        boolean keyMatched = exchange.isBound(method.getBindingKey());
+                        result.setKeyNotMatched(!keyMatched);
+                        if(method.hasArguments())
+                        {
+                            if(keyMatched)
+                            {
+                                result.setArgsNotMatched(!exchange.isBound(method.getBindingKey(), method.getArguments()));
+                            }
+                            else
+                            {
+                                result.setArgsNotMatched(!exchange.isBound(method.getArguments()));
+                            }
+                        }
+                    }
+
+                }
+                else if (method.hasArguments())
+                {
+                    if(queueMatched)
+                    {
+                        result.setArgsNotMatched(!exchange.isBound(method.getArguments(), queue));
+                    }
+                    else
+                    {
+                        result.setArgsNotMatched(!exchange.isBound(method.getArguments()));
+                    }
+                }
+
+            }
+            else if(exchange != null && method.hasBindingKey())
+            {
+                final boolean keyMatched = exchange.isBound(method.getBindingKey());
+                result.setKeyNotMatched(!keyMatched);
+
+                if(method.hasArguments())
+                {
+                    if(keyMatched)
+                    {
+                        result.setArgsNotMatched(!exchange.isBound(method.getBindingKey(), method.getArguments()));
+                    }
+                    else
+                    {
+                        result.setArgsNotMatched(!exchange.isBound(method.getArguments()));
+                    }
+                }
+
+
+            }
+
+        }
+        else if(exchange != null && method.hasBindingKey())
+        {
+            final boolean keyMatched = exchange.isBound(method.getBindingKey());
+            result.setKeyNotMatched(!keyMatched);
+
+            if(method.hasArguments())
+            {
+                if(keyMatched)
+                {
+                    result.setArgsNotMatched(!exchange.isBound(method.getBindingKey(), method.getArguments()));
+                }
+                else
+                {
+                    result.setArgsNotMatched(!exchange.isBound(method.getArguments()));
+                }
+            }
+
+        }
+        else if(exchange != null && method.hasArguments())
+        {
+            result.setArgsNotMatched(!exchange.isBound(method.getArguments()));
+        }
+
+
+        session.executionResult((int) method.getId(), result);
+
+
+    }
+
+    private AMQQueue getQueue(Session session, String queue)
+    {
+        QueueRegistry queueRegistry = getQueueRegistry(session);
+        return queueRegistry.getQueue(queue);
+    }
+
+    private QueueRegistry getQueueRegistry(Session session)
+    {
+        return getVirtualHost(session).getQueueRegistry();
+    }
+
+    @Override
+    public void queueDeclare(Session session, final QueueDeclare method)
+    {
+
+        VirtualHost virtualHost = getVirtualHost(session);
+        DurableConfigurationStore store = virtualHost.getDurableConfigurationStore();
+
+        String queueName = method.getQueue();
+        AMQQueue queue;
+        QueueRegistry queueRegistry = getQueueRegistry(session);
+        //TODO: do we need to check that the queue already exists with exactly the same "configuration"?
+
+        synchronized (queueRegistry)
+        {
+
+            if (((queue = queueRegistry.getQueue(queueName)) == null))
+            {
+
+                if (method.getPassive())
+                {
+                    String description = "Queue: " + queueName + " not found on VirtualHost(" + virtualHost + ").";
+                    ExecutionErrorCode errorCode = ExecutionErrorCode.NOT_FOUND;
+
+                    exception(session, method, errorCode, description);
+
+                    return;
+                }
+                else
+                {
+                    try
+                    {
+                        queue = createQueue(queueName, method, virtualHost, (ServerSession)session);
+                        if(!method.getExclusive() && method.getAutoDelete())
+                        {
+                            queue.setDeleteOnNoConsumers(true);
+                        }
+
+                        final String alternateExchangeName = method.getAlternateExchange();
+                        if(alternateExchangeName != null && alternateExchangeName.length() != 0)
+                        {
+                            Exchange alternate = getExchange(session, alternateExchangeName);
+                            queue.setAlternateExchange(alternate);
+                        }
+
+                        if(method.hasArguments()  && method.getArguments() != null)
+                        {
+                            if(method.getArguments().containsKey(QUEUE_ARGUMENT_NO_LOCAL))
+                            {
+                                Object noLocal = method.getArguments().get(QUEUE_ARGUMENT_NO_LOCAL);
+                                queue.setNoLocal(convertBooleanValue(noLocal));
+                            }
+                        }
+
+
+                        if (queue.isDurable() && !queue.isAutoDelete())
+                        {
+                            if(method.hasArguments() && method.getArguments() != null)
+                            {
+                                Map<String,Object> args = method.getArguments();
+                                FieldTable ftArgs = new FieldTable();
+                                for(Map.Entry<String, Object> entry : args.entrySet())
+                                {
+                                    ftArgs.put(new AMQShortString(entry.getKey()), entry.getValue());
+                                }
+                                DurableConfigurationStoreHelper.createQueue(store, queue, ftArgs);
+                            }
+                            else
+                            {
+                                DurableConfigurationStoreHelper.createQueue(store, queue, null);
+                            }
+                        }
+                        queueRegistry.registerQueue(queue);
+
+                        if (method.hasAutoDelete()
+                            && method.getAutoDelete()
+                            && method.hasExclusive()
+                            && method.getExclusive())
+                        {
+                            final AMQQueue q = queue;
+                            final ServerSession.Task deleteQueueTask = new ServerSession.Task()
+                                {
+                                    public void doTask(ServerSession session)
+                                    {
+                                        try
+                                        {
+                                            q.delete();
+                                        }
+                                        catch (AMQException e)
+                                        {
+                                            exception(session, method, e, "Cannot delete '" + method.getQueue());
+                                        }
+                                    }
+                                };
+                            final ServerSession s = (ServerSession) session;
+                            s.addSessionCloseTask(deleteQueueTask);
+                            queue.addQueueDeleteTask(new AMQQueue.Task()
+                                {
+                                    public void doTask(AMQQueue queue) throws AMQException
+                                    {
+                                        s.removeSessionCloseTask(deleteQueueTask);
+                                    }
+                                });
+                        }
+                        if (method.hasExclusive()
+                            && method.getExclusive())
+                        {
+                            final AMQQueue q = queue;
+                            final ServerSession.Task removeExclusive = new ServerSession.Task()
+                            {
+                                public void doTask(ServerSession session)
+                                {
+                                    q.setAuthorizationHolder(null);
+                                    q.setExclusiveOwningSession(null);
+                                }
+                            };
+                            final ServerSession s = (ServerSession) session;
+                            q.setExclusiveOwningSession(s);
+                            s.addSessionCloseTask(removeExclusive);
+                            queue.addQueueDeleteTask(new AMQQueue.Task()
+                            {
+                                public void doTask(AMQQueue queue) throws AMQException
+                                {
+                                    s.removeSessionCloseTask(removeExclusive);
+                                }
+                            });
+                        }
+                    }
+                    catch (AMQException e)
+                    {
+                        exception(session, method, e, "Cannot declare queue '" + queueName);
+                    }
+                }
+            }
+            else if (method.getExclusive() && (queue.getExclusiveOwningSession() != null && !queue.getExclusiveOwningSession().equals(session)))
+            {
+                    String description = "Cannot declare queue('" + queueName + "'),"
+                                                                           + " as exclusive queue with same name "
+                                                                           + "declared on another session";
+                    ExecutionErrorCode errorCode = ExecutionErrorCode.RESOURCE_LOCKED;
+
+                    exception(session, method, errorCode, description);
+
+                    return;
+            }
+        }
+    }
+
+    /**
+     * Converts a queue argument into a boolean value.  For compatibility with the C++
+     * and the clients, accepts with Boolean, String, or Number types.
+     * @param argValue  argument value.
+     *
+     * @return true if set
+     */
+    private boolean convertBooleanValue(Object argValue)
+    {
+        if(argValue instanceof Boolean && ((Boolean)argValue))
+        {
+            return true;
+        }
+        else if (argValue instanceof String && Boolean.parseBoolean((String)argValue))
+        {
+            return true;
+        }
+        else if (argValue instanceof Number && ((Number)argValue).intValue() != 0)
+        {
+            return true;
+        }
+        return false;
+    }
+
+    protected AMQQueue createQueue(final String queueName,
+                                   final QueueDeclare body,
+                                   VirtualHost virtualHost,
+                                   final ServerSession session)
+            throws AMQException
+    {
+        String owner = body.getExclusive() ? session.getClientID() : null;
+
+        final AMQQueue queue = AMQQueueFactory.createAMQQueueImpl(UUIDGenerator.generateQueueUUID(queueName, virtualHost.getName()), queueName, body.getDurable(), owner,
+                                                                  body.getAutoDelete(), body.getExclusive(), virtualHost, body.getArguments());
+
+        return queue;
+    }
+
+    @Override
+    public void queueDelete(Session session, QueueDelete method)
+    {
+        String queueName = method.getQueue();
+        if(queueName == null || queueName.length()==0)
+        {
+            exception(session, method, ExecutionErrorCode.INVALID_ARGUMENT, "No queue name supplied");
+
+        }
+        else
+        {
+            AMQQueue queue = getQueue(session, queueName);
+
+
+            if (queue == null)
+            {
+                exception(session, method, ExecutionErrorCode.NOT_FOUND, "No queue " + queueName + " found");
+            }
+            else
+            {
+                if(queue.getAuthorizationHolder() != null && queue.getAuthorizationHolder() != session)
+                {
+                    exception(session,method,ExecutionErrorCode.RESOURCE_LOCKED, "Exclusive Queue: " + queueName + " owned exclusively by another session");
+                }
+                else if(queue.isExclusive() && queue.getExclusiveOwningSession()  != null && queue.getExclusiveOwningSession() != session)
+                {
+                    exception(session,method,ExecutionErrorCode.RESOURCE_LOCKED, "Exclusive Queue: " + queueName + " owned exclusively by another session");
+                }
+                else if (method.getIfEmpty() && !queue.isEmpty())
+                {
+                    exception(session, method, ExecutionErrorCode.PRECONDITION_FAILED, "Queue " + queueName + " not empty");
+                }
+                else if (method.getIfUnused() && !queue.isUnused())
+                {
+                    // TODO - Error code
+                    exception(session, method, ExecutionErrorCode.PRECONDITION_FAILED, "Queue " + queueName + " in use");
+
+                }
+                else
+                {
+                    VirtualHost virtualHost = getVirtualHost(session);
+
+                    try
+                    {
+                        queue.delete();
+                        if (queue.isDurable() && !queue.isAutoDelete())
+                        {
+                            DurableConfigurationStore store = virtualHost.getDurableConfigurationStore();
+                            DurableConfigurationStoreHelper.removeQueue(store,queue);
+                        }
+                    }
+                    catch (AMQException e)
+                    {
+                        exception(session, method, e, "Cannot delete queue '" + queueName);
+                    }
+                }
+            }
+        }
+    }
+
+    @Override
+    public void queuePurge(Session session, QueuePurge method)
+    {
+        String queueName = method.getQueue();
+        if(queueName == null || queueName.length()==0)
+        {
+            exception(session, method, ExecutionErrorCode.ILLEGAL_ARGUMENT, "No queue name supplied");
+        }
+        else
+        {
+            AMQQueue queue = getQueue(session, queueName);
+
+            if (queue == null)
+            {
+                exception(session, method, ExecutionErrorCode.NOT_FOUND, "No queue " + queueName + " found");
+            }
+            else
+            {
+                try
+                {
+                    queue.clearQueue();
+                }
+                catch (AMQException e)
+                {
+                    exception(session, method, e, "Cannot purge queue '" + queueName);
+                }
+            }
+        }
+    }
+
+    @Override
+    public void queueQuery(Session session, QueueQuery method)
+    {
+        QueueQueryResult result = new QueueQueryResult();
+
+        AMQQueue queue = getQueue(session, method.getQueue());
+
+        if(queue != null)
+        {
+            result.setQueue(queue.getNameShortString().toString());
+            result.setDurable(queue.isDurable());
+            result.setExclusive(queue.isExclusive());
+            result.setAutoDelete(queue.isAutoDelete());
+            result.setArguments(queue.getArguments());
+            result.setMessageCount(queue.getMessageCount());
+            result.setSubscriberCount(queue.getConsumerCount());
+
+        }
+
+
+        session.executionResult((int) method.getId(), result);
+
+    }
+
+    @Override
+    public void messageSetFlowMode(Session session, MessageSetFlowMode sfm)
+    {
+        String destination = sfm.getDestination();
+
+        Subscription_0_10 sub = ((ServerSession)session).getSubscription(destination);
+
+        if(sub == null)
+        {
+            exception(session, sfm, ExecutionErrorCode.NOT_FOUND, "not-found: destination '"+destination+"'");
+        }
+        else if(sub.isStopped())
+        {
+            sub.setFlowMode(sfm.getFlowMode());
+        }
+    }
+
+    @Override
+    public void messageStop(Session session, MessageStop stop)
+    {
+        String destination = stop.getDestination();
+
+        Subscription_0_10 sub = ((ServerSession)session).getSubscription(destination);
+
+        if(sub == null)
+        {
+            exception(session, stop, ExecutionErrorCode.NOT_FOUND, "not-found: destination '"+destination+"'");
+        }
+        else
+        {
+            sub.stop();
+        }
+
+    }
+
+    @Override
+    public void messageFlow(Session session, MessageFlow flow)
+    {
+        String destination = flow.getDestination();
+
+        Subscription_0_10 sub = ((ServerSession)session).getSubscription(destination);
+
+        if(sub == null)
+        {
+            exception(session, flow, ExecutionErrorCode.NOT_FOUND, "not-found: destination '"+destination+"'");
+        }
+        else
+        {
+            sub.addCredit(flow.getUnit(), flow.getValue());
+        }
+
+    }
+
+    @Override
+    public void closed(Session session)
+    {
+        setThreadSubject(session);
+
+        ServerSession serverSession = (ServerSession)session;
+
+        serverSession.stopSubscriptions();
+        serverSession.onClose();
+        serverSession.unregisterSubscriptions();
+    }
+
+    @Override
+    public void detached(Session session)
+    {
+        closed(session);
+    }
+
+    private void setThreadSubject(Session session)
+    {
+        final ServerConnection scon = (ServerConnection) session.getConnection();
+        SecurityManager.setThreadSubject(scon.getAuthorizedSubject());
+    }
+
+    private static class CommandProcessedAction implements ServerTransaction.Action
+    {
+        private final ServerSession _serverSession;
+        private final Method _method;
+
+        public CommandProcessedAction(final ServerSession serverSession, final Method xfr)
+        {
+            _serverSession = serverSession;
+            _method = xfr;
+        }
+
+        public void postCommit()
+        {
+            _serverSession.processed(_method);
+        }
+
+        public void onRollback()
+        {
+        }
+    }
+}
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/Subscription_0_10.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/Subscription_0_10.java
new file mode 100644
index 0000000000..c6bceb6ac7
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/Subscription_0_10.java
@@ -0,0 +1,952 @@
+/*
+ *
+ * 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.v0_10;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.BasicContentHeaderProperties;
+import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.server.exchange.Exchange;
+import org.apache.qpid.server.filter.FilterManager;
+import org.apache.qpid.server.flow.FlowCreditManager;
+import org.apache.qpid.server.logging.LogActor;
+import org.apache.qpid.server.logging.LogSubject;
+import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.logging.actors.GenericActor;
+import org.apache.qpid.server.logging.messages.ChannelMessages;
+import org.apache.qpid.server.logging.messages.SubscriptionMessages;
+import org.apache.qpid.server.plugin.MessageConverter;
+import org.apache.qpid.server.protocol.MessageConverterRegistry;
+import org.apache.qpid.server.message.InboundMessage;
+import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.BaseQueue;
+import org.apache.qpid.server.queue.InboundMessageAdapter;
+import org.apache.qpid.server.queue.QueueEntry;
+import org.apache.qpid.server.subscription.Subscription;
+import org.apache.qpid.server.txn.AutoCommitTransaction;
+import org.apache.qpid.server.txn.ServerTransaction;
+import org.apache.qpid.transport.DeliveryProperties;
+import org.apache.qpid.transport.Header;
+import org.apache.qpid.transport.MessageAcceptMode;
+import org.apache.qpid.transport.MessageAcquireMode;
+import org.apache.qpid.transport.MessageCreditUnit;
+import org.apache.qpid.transport.MessageFlowMode;
+import org.apache.qpid.transport.MessageProperties;
+import org.apache.qpid.transport.MessageTransfer;
+import org.apache.qpid.transport.Method;
+import org.apache.qpid.transport.Option;
+import org.apache.qpid.transport.Struct;
+
+import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.QUEUE_FORMAT;
+import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.SUBSCRIPTION_FORMAT;
+
+import java.text.MessageFormat;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+public class Subscription_0_10 implements Subscription, FlowCreditManager.FlowCreditManagerListener, LogSubject
+{
+    private final long _subscriptionID;
+
+    private final QueueEntry.SubscriptionAcquiredState _owningState = new QueueEntry.SubscriptionAcquiredState(this);
+
+    private static final Option[] BATCHED = new Option[] { Option.BATCH };
+
+    private final Lock _stateChangeLock = new ReentrantLock();
+
+    private final AtomicReference<State> _state = new AtomicReference<State>(State.ACTIVE);
+    private volatile AMQQueue.Context _queueContext;
+    private final AtomicBoolean _deleted = new AtomicBoolean(false);
+
+
+    private FlowCreditManager_0_10 _creditManager;
+
+    private StateListener _stateListener = new StateListener()
+                                            {
+
+                                                public void stateChange(Subscription sub, State oldState, State newState)
+                                                {
+                                                    CurrentActor.get().message(SubscriptionMessages.STATE(newState.toString()));
+                                                }
+                                            };
+    private AMQQueue _queue;
+    private final String _destination;
+    private boolean _noLocal;
+    private final FilterManager _filters;
+    private final MessageAcceptMode _acceptMode;
+    private final MessageAcquireMode _acquireMode;
+    private MessageFlowMode _flowMode;
+    private final ServerSession _session;
+    private final AtomicBoolean _stopped = new AtomicBoolean(true);
+    private static final Struct[] EMPTY_STRUCT_ARRAY = new Struct[0];
+
+    private LogActor _logActor;
+    private final Map<String, Object> _properties = new ConcurrentHashMap<String, Object>();
+    private String _traceExclude;
+    private String _trace;
+    private final long _createTime = System.currentTimeMillis();
+    private final AtomicLong _deliveredCount = new AtomicLong(0);
+    private final AtomicLong _deliveredBytes = new AtomicLong(0);
+    private final AtomicLong _unacknowledgedCount = new AtomicLong(0);
+    private final AtomicLong _unacknowledgedBytes = new AtomicLong(0);
+
+    private final Map<String, Object> _arguments;
+    private int _deferredMessageCredit;
+    private long _deferredSizeCredit;
+
+
+    public Subscription_0_10(ServerSession session, String destination, MessageAcceptMode acceptMode,
+                             MessageAcquireMode acquireMode,
+                             MessageFlowMode flowMode,
+                             FlowCreditManager_0_10 creditManager,
+                             FilterManager filters,Map<String, Object> arguments)
+    {
+        _subscriptionID = SUB_ID_GENERATOR.getAndIncrement();
+        _session = session;
+        _postIdSettingAction = new AddMessageDispositionListenerAction(session);
+        _destination = destination;
+        _acceptMode = acceptMode;
+        _acquireMode = acquireMode;
+        _creditManager = creditManager;
+        _flowMode = flowMode;
+        _filters = filters;
+        _creditManager.addStateListener(this);
+        _arguments = arguments == null ? Collections.<String, Object> emptyMap() :
+                                         Collections.<String, Object> unmodifiableMap(arguments);
+        _state.set(_creditManager.hasCredit() ? State.ACTIVE : State.SUSPENDED);
+
+    }
+
+    public void setNoLocal(boolean noLocal)
+    {
+        _noLocal = noLocal;
+    }
+
+    public AMQQueue getQueue()
+    {
+        return _queue;
+    }
+
+    public QueueEntry.SubscriptionAcquiredState getOwningState()
+    {
+        return _owningState;
+    }
+
+    public void setQueue(AMQQueue queue, boolean exclusive)
+    {
+        if(getQueue() != null)
+        {
+            throw new IllegalStateException("Attempt to set queue for subscription " + this + " to " + queue + "when already set to " + getQueue());
+        }
+        _queue = queue;
+
+        Map<String, Object> arguments = queue.getArguments();
+        _traceExclude = (String) arguments.get("qpid.trace.exclude");
+        _trace = (String) arguments.get("qpid.trace.id");
+        String filterLogString = null;
+
+        _logActor = GenericActor.getInstance(this);
+        if (CurrentActor.get().getRootMessageLogger().isMessageEnabled(_logActor, this, SubscriptionMessages.CREATE_LOG_HIERARCHY))
+        {
+            filterLogString = getFilterLogString();
+            CurrentActor.get().message(this, SubscriptionMessages.CREATE(filterLogString, queue.isDurable() && exclusive,
+                    filterLogString.length() > 0));
+        }
+    }
+
+    public String getConsumerName()
+    {
+        return _destination;
+    }
+
+    public boolean isSuspended()
+    {
+        return !isActive() || _deleted.get() || _session.isClosing() || _session.getConnectionModel().isStopped(); // TODO check for Session suspension
+    }
+
+    public boolean hasInterest(QueueEntry entry)
+    {
+
+
+
+        //check that the message hasn't been rejected
+        if (entry.isRejectedBy(getSubscriptionID()))
+        {
+
+            return false;
+        }
+
+        if (entry.getMessage() instanceof MessageTransferMessage)
+        {
+            if(_noLocal)
+            {
+                Object connectionRef = ((MessageTransferMessage)entry.getMessage()).getConnectionReference();
+                if (connectionRef != null && connectionRef == _session.getReference())
+                {
+                    return false;
+                }
+            }
+        }
+        else
+        {
+            // no interest in messages we can't convert
+            if(MessageConverterRegistry.getConverter(entry.getMessage().getClass(), MessageTransferMessage.class)==null)
+            {
+                return false;
+            }
+        }
+
+
+        return checkFilters(entry);
+
+
+    }
+
+    private boolean checkFilters(QueueEntry entry)
+    {
+        return (_filters == null) || _filters.allAllow(entry);
+    }
+
+    public boolean isClosed()
+    {
+        return getState() == State.CLOSED;
+    }
+
+    public boolean isBrowser()
+    {
+        return _acquireMode == MessageAcquireMode.NOT_ACQUIRED;
+    }
+
+    public boolean seesRequeues()
+    {
+        return _acquireMode != MessageAcquireMode.NOT_ACQUIRED || _acceptMode == MessageAcceptMode.EXPLICIT;
+    }
+
+    public void close()
+    {
+        boolean closed = false;
+        State state = getState();
+
+        _stateChangeLock.lock();
+        try
+        {
+            while(!closed && state != State.CLOSED)
+            {
+                closed = _state.compareAndSet(state, State.CLOSED);
+                if(!closed)
+                {
+                    state = getState();
+                }
+                else
+                {
+                    _stateListener.stateChange(this,state, State.CLOSED);
+                }
+            }
+            _creditManager.removeListener(this);
+            CurrentActor.get().message(getLogSubject(), SubscriptionMessages.CLOSE());
+        }
+        finally
+        {
+            _stateChangeLock.unlock();
+        }
+
+
+
+    }
+
+    public Long getDelivered()
+    {
+        return _deliveredCount.get();
+    }
+
+    public void creditStateChanged(boolean hasCredit)
+    {
+
+        if(hasCredit)
+        {
+            if(_state.compareAndSet(State.SUSPENDED, State.ACTIVE))
+            {
+                _stateListener.stateChange(this, State.SUSPENDED, State.ACTIVE);
+            }
+            else
+            {
+                // this is a hack to get round the issue of increasing bytes credit
+                _stateListener.stateChange(this, State.ACTIVE, State.ACTIVE);
+            }
+        }
+        else
+        {
+            if(_state.compareAndSet(State.ACTIVE, State.SUSPENDED))
+            {
+                _stateListener.stateChange(this, State.ACTIVE, State.SUSPENDED);
+            }
+        }
+    }
+
+
+    public static class AddMessageDispositionListenerAction implements Runnable
+    {
+        private MessageTransfer _xfr;
+        private ServerSession.MessageDispositionChangeListener _action;
+        private ServerSession _session;
+
+        public AddMessageDispositionListenerAction(ServerSession session)
+        {
+            _session = session;
+        }
+
+        public void setXfr(MessageTransfer xfr)
+        {
+            _xfr = xfr;
+        }
+
+        public void setAction(ServerSession.MessageDispositionChangeListener action)
+        {
+            _action = action;
+        }
+
+        public void run()
+        {
+            if(_action != null)
+            {
+                _session.onMessageDispositionChange(_xfr, _action);
+            }
+        }
+    }
+
+    private final AddMessageDispositionListenerAction _postIdSettingAction;
+
+    public void send(final QueueEntry entry, boolean batch) throws AMQException
+    {
+        ServerMessage serverMsg = entry.getMessage();
+
+
+        MessageTransfer xfr;
+
+        DeliveryProperties deliveryProps;
+        MessageProperties messageProps = null;
+
+        MessageTransferMessage msg;
+
+        if(serverMsg instanceof MessageTransferMessage)
+        {
+
+            msg = (MessageTransferMessage) serverMsg;
+
+        }
+        else
+        {
+            MessageConverter converter =
+                    MessageConverterRegistry.getConverter(serverMsg.getClass(), MessageTransferMessage.class);
+
+
+            msg = (MessageTransferMessage) converter.convert(serverMsg, getQueue().getVirtualHost());
+        }
+        DeliveryProperties origDeliveryProps = msg.getHeader() == null ? null : msg.getHeader().getDeliveryProperties();
+        messageProps = msg.getHeader() == null ? null : msg.getHeader().getMessageProperties();
+
+        deliveryProps = new DeliveryProperties();
+        if(origDeliveryProps != null)
+        {
+            if(origDeliveryProps.hasDeliveryMode())
+            {
+                deliveryProps.setDeliveryMode(origDeliveryProps.getDeliveryMode());
+            }
+            if(origDeliveryProps.hasExchange())
+            {
+                deliveryProps.setExchange(origDeliveryProps.getExchange());
+            }
+            if(origDeliveryProps.hasExpiration())
+            {
+                deliveryProps.setExpiration(origDeliveryProps.getExpiration());
+            }
+            if(origDeliveryProps.hasPriority())
+            {
+                deliveryProps.setPriority(origDeliveryProps.getPriority());
+            }
+            if(origDeliveryProps.hasRoutingKey())
+            {
+                deliveryProps.setRoutingKey(origDeliveryProps.getRoutingKey());
+            }
+            if(origDeliveryProps.hasTimestamp())
+            {
+                deliveryProps.setTimestamp(origDeliveryProps.getTimestamp());
+            }
+            if(origDeliveryProps.hasTtl())
+            {
+                deliveryProps.setTtl(origDeliveryProps.getTtl());
+            }
+
+
+        }
+
+        deliveryProps.setRedelivered(entry.isRedelivered());
+
+        if(_trace != null && messageProps == null)
+        {
+            messageProps = new MessageProperties();
+        }
+
+        Header header = new Header(deliveryProps, messageProps, msg.getHeader() == null ? null : msg.getHeader().getNonStandardProperties());
+
+
+        xfr = batch ? new MessageTransfer(_destination,_acceptMode,_acquireMode,header,msg.getBody(), BATCHED)
+                    : new MessageTransfer(_destination,_acceptMode,_acquireMode,header,msg.getBody());
+
+        boolean excludeDueToFederation = false;
+
+        if(_trace != null)
+        {
+            if(!messageProps.hasApplicationHeaders())
+            {
+                messageProps.setApplicationHeaders(new HashMap<String,Object>());
+            }
+            Map<String,Object> appHeaders = messageProps.getApplicationHeaders();
+            String trace = (String) appHeaders.get("x-qpid.trace");
+            if(trace == null)
+            {
+                trace = _trace;
+            }
+            else
+            {
+                if(_traceExclude != null)
+                {
+                    excludeDueToFederation = Arrays.asList(trace.split(",")).contains(_traceExclude);
+                }
+                trace+=","+_trace;
+            }
+            appHeaders.put("x-qpid.trace",trace);
+        }
+
+        if(!excludeDueToFederation)
+        {
+            if(_acceptMode == MessageAcceptMode.NONE && _acquireMode != MessageAcquireMode.PRE_ACQUIRED)
+            {
+                xfr.setCompletionListener(new MessageAcceptCompletionListener(this, _session, entry, _flowMode == MessageFlowMode.WINDOW));
+            }
+            else if(_flowMode == MessageFlowMode.WINDOW)
+            {
+                xfr.setCompletionListener(new Method.CompletionListener()
+                                            {
+                                                public void onComplete(Method method)
+                                                {
+                                                    deferredAddCredit(1, entry.getSize());
+                                                }
+                                            });
+            }
+
+
+            _postIdSettingAction.setXfr(xfr);
+            if(_acceptMode == MessageAcceptMode.EXPLICIT)
+            {
+                _postIdSettingAction.setAction(new ExplicitAcceptDispositionChangeListener(entry, this));
+            }
+            else if(_acquireMode != MessageAcquireMode.PRE_ACQUIRED)
+            {
+                _postIdSettingAction.setAction(new ImplicitAcceptDispositionChangeListener(entry, this));
+            }
+            else
+            {
+                _postIdSettingAction.setAction(null);
+            }
+
+
+            _session.sendMessage(xfr, _postIdSettingAction);
+            entry.incrementDeliveryCount();
+            _deliveredCount.incrementAndGet();
+            _deliveredBytes.addAndGet(entry.getSize());
+            if(_acceptMode == MessageAcceptMode.NONE && _acquireMode == MessageAcquireMode.PRE_ACQUIRED)
+            {
+                forceDequeue(entry, false);
+            }
+            else if(_acquireMode == MessageAcquireMode.PRE_ACQUIRED)
+            {
+                recordUnacknowledged(entry);
+            }
+        }
+        else
+        {
+            forceDequeue(entry, _flowMode == MessageFlowMode.WINDOW);
+
+        }
+    }
+
+    void recordUnacknowledged(QueueEntry entry)
+    {
+        _unacknowledgedCount.incrementAndGet();
+        _unacknowledgedBytes.addAndGet(entry.getSize());
+    }
+
+    private void deferredAddCredit(final int deferredMessageCredit, final long deferredSizeCredit)
+    {
+        _deferredMessageCredit += deferredMessageCredit;
+        _deferredSizeCredit += deferredSizeCredit;
+
+    }
+
+    public void flushCreditState(boolean strict)
+    {
+        if(strict || !isSuspended() || _deferredMessageCredit >= 200
+          || !(_creditManager instanceof WindowCreditManager)
+          || ((WindowCreditManager)_creditManager).getMessageCreditLimit() < 400 )
+        {
+            _creditManager.restoreCredit(_deferredMessageCredit, _deferredSizeCredit);
+            _deferredMessageCredit = 0;
+            _deferredSizeCredit = 0l;
+        }
+    }
+
+    private void forceDequeue(final QueueEntry entry, final boolean restoreCredit)
+    {
+        AutoCommitTransaction dequeueTxn = new AutoCommitTransaction(getQueue().getVirtualHost().getMessageStore());
+        dequeueTxn.dequeue(entry.getQueue(), entry.getMessage(),
+                           new ServerTransaction.Action()
+                           {
+                               public void postCommit()
+                               {
+                                   if (restoreCredit)
+                                   {
+                                       restoreCredit(entry);
+                                   }
+                                   entry.discard();
+                               }
+
+                               public void onRollback()
+                               {
+
+                               }
+                           });
+   }
+
+    void reject(final QueueEntry entry)
+    {
+        entry.setRedelivered();
+        entry.routeToAlternate();
+        if(entry.isAcquiredBy(this))
+        {
+            entry.discard();
+        }
+    }
+
+    void release(final QueueEntry entry, final boolean setRedelivered)
+    {
+        if (setRedelivered)
+        {
+            entry.setRedelivered();
+        }
+
+        if (getSessionModel().isClosing() || !setRedelivered)
+        {
+            entry.decrementDeliveryCount();
+        }
+
+        if (isMaxDeliveryLimitReached(entry))
+        {
+            sendToDLQOrDiscard(entry);
+        }
+        else
+        {
+            entry.release();
+        }
+    }
+
+    protected void sendToDLQOrDiscard(QueueEntry entry)
+    {
+        final Exchange alternateExchange = entry.getQueue().getAlternateExchange();
+        final LogActor logActor = CurrentActor.get();
+        final ServerMessage msg = entry.getMessage();
+        if (alternateExchange != null)
+        {
+            final InboundMessage m = new InboundMessageAdapter(entry);
+
+            final List<? extends BaseQueue> destinationQueues = alternateExchange.route(m);
+
+            if (destinationQueues == null || destinationQueues.isEmpty())
+            {
+                entry.discard();
+
+                logActor.message( ChannelMessages.DISCARDMSG_NOROUTE(msg.getMessageNumber(), alternateExchange.getName()));
+            }
+            else
+            {
+                entry.routeToAlternate();
+
+                //output operational logging for each delivery post commit
+                for (final BaseQueue destinationQueue : destinationQueues)
+                {
+                    logActor.message( ChannelMessages.DEADLETTERMSG(msg.getMessageNumber(), destinationQueue.getNameShortString().asString()));
+                }
+            }
+        }
+        else
+        {
+            entry.discard();
+            logActor.message(ChannelMessages.DISCARDMSG_NOALTEXCH(msg.getMessageNumber(), entry.getQueue().getName(), msg.getRoutingKey()));
+        }
+    }
+
+    private boolean isMaxDeliveryLimitReached(QueueEntry entry)
+    {
+        final int maxDeliveryLimit = entry.getQueue().getMaximumDeliveryCount();
+        return (maxDeliveryLimit > 0 && entry.getDeliveryCount() >= maxDeliveryLimit);
+    }
+
+    public void queueDeleted(AMQQueue queue)
+    {
+        _deleted.set(true);
+    }
+
+    public boolean wouldSuspend(QueueEntry entry)
+    {
+        return !_creditManager.useCreditForMessage(entry.getMessage().getSize());
+    }
+
+    public boolean trySendLock()
+    {
+        return _stateChangeLock.tryLock();
+    }
+
+
+    public void getSendLock()
+    {
+        _stateChangeLock.lock();
+    }
+
+    public void releaseSendLock()
+    {
+        _stateChangeLock.unlock();
+    }
+
+    public void restoreCredit(QueueEntry queueEntry)
+    {
+        _creditManager.restoreCredit(1, queueEntry.getSize());
+    }
+
+    public void onDequeue(QueueEntry queueEntry)
+    {
+        // no-op for 0-10, credit restored by completing command.
+    }
+
+    public void releaseQueueEntry(QueueEntry queueEntry)
+    {
+        // no-op for 0-10, credit restored by completing command.
+    }
+
+    public void setStateListener(StateListener listener)
+    {
+        _stateListener = listener;
+    }
+
+    public State getState()
+    {
+        return _state.get();
+    }
+
+    public AMQQueue.Context getQueueContext()
+    {
+        return _queueContext;
+    }
+
+    public void setQueueContext(AMQQueue.Context queueContext)
+    {
+        _queueContext = queueContext;
+    }
+
+    public boolean isActive()
+    {
+        return getState() == State.ACTIVE;
+    }
+
+    public void set(String key, Object value)
+    {
+        _properties.put(key, value);
+    }
+
+    public Object get(String key)
+    {
+        return _properties.get(key);
+    }
+
+
+    public FlowCreditManager_0_10 getCreditManager()
+    {
+        return _creditManager;
+    }
+
+
+    public void stop()
+    {
+        try
+        {
+            getSendLock();
+
+            if(_state.compareAndSet(State.ACTIVE, State.SUSPENDED))
+            {
+                _stateListener.stateChange(this, State.ACTIVE, State.SUSPENDED);
+            }
+            _stopped.set(true);
+            FlowCreditManager_0_10 creditManager = getCreditManager();
+            creditManager.clearCredit();
+        }
+        finally
+        {
+            releaseSendLock();
+        }
+    }
+
+    public void addCredit(MessageCreditUnit unit, long value)
+    {
+        FlowCreditManager_0_10 creditManager = getCreditManager();
+
+        switch (unit)
+        {
+            case MESSAGE:
+
+                creditManager.addCredit(value, 0L);
+                break;
+            case BYTE:
+                creditManager.addCredit(0l, value);
+                break;
+        }
+
+        _stopped.set(false);
+
+        if(creditManager.hasCredit())
+        {
+            if(_state.compareAndSet(State.SUSPENDED, State.ACTIVE))
+            {
+                _stateListener.stateChange(this, State.SUSPENDED, State.ACTIVE);
+            }
+        }
+
+    }
+
+    public void setFlowMode(MessageFlowMode flowMode)
+    {
+
+
+        _creditManager.removeListener(this);
+
+        switch(flowMode)
+        {
+            case CREDIT:
+                _creditManager = new CreditCreditManager(0l,0l);
+                break;
+            case WINDOW:
+                _creditManager = new WindowCreditManager(0l,0l);
+                break;
+            default:
+                throw new RuntimeException("Unknown message flow mode: " + flowMode);
+        }
+        _flowMode = flowMode;
+        if(_state.compareAndSet(State.ACTIVE, State.SUSPENDED))
+        {
+            _stateListener.stateChange(this, State.ACTIVE, State.SUSPENDED);
+        }
+
+        _creditManager.addStateListener(this);
+
+    }
+
+    public boolean isStopped()
+    {
+        return _stopped.get();
+    }
+
+    public boolean acquires()
+    {
+        return _acquireMode == MessageAcquireMode.PRE_ACQUIRED;
+    }
+
+    public void acknowledge(QueueEntry entry)
+    {
+        // TODO Fix Store Context / cleanup
+        if(entry.isAcquiredBy(this))
+        {
+            _unacknowledgedBytes.addAndGet(-entry.getSize());
+            _unacknowledgedCount.decrementAndGet();
+            entry.discard();
+        }
+    }
+
+    public void flush() throws AMQException
+    {
+        flushCreditState(true);
+        _queue.flushSubscription(this);
+        stop();
+    }
+
+    public long getSubscriptionID()
+    {
+        return _subscriptionID;
+    }
+
+    public LogActor getLogActor()
+    {
+        return _logActor;
+    }
+
+    public boolean isTransient()
+    {
+        return false;
+    }
+
+    public ServerSession getSessionModel()
+    {
+        return _session;
+    }
+
+    public boolean isBrowsing()
+    {
+        return _acquireMode == MessageAcquireMode.NOT_ACQUIRED;
+    }
+
+    public boolean isExclusive()
+    {
+        return getQueue().hasExclusiveSubscriber();
+    }
+
+    public boolean isDurable()
+    {
+        return false;
+    }
+
+
+    public boolean isExplicitAcknowledge()
+    {
+        return _acceptMode == MessageAcceptMode.EXPLICIT;
+    }
+
+    public String getCreditMode()
+    {
+        return _flowMode.toString();
+    }
+
+    public String getName()
+    {
+        return _destination;
+    }
+
+    public Map<String, Object> getArguments()
+    {
+        return _arguments;
+    }
+
+    public boolean isSessionTransactional()
+    {
+        return _session.isTransactional();
+    }
+
+    public void queueEmpty()
+    {
+    }
+
+    public long getCreateTime()
+    {
+        return _createTime;
+    }
+
+    public String toLogString()
+    {
+        String queueInfo = MessageFormat.format(QUEUE_FORMAT, _queue.getVirtualHost().getName(),
+                  _queue.getNameShortString());
+        String result = "[" + MessageFormat.format(SUBSCRIPTION_FORMAT, getSubscriptionID()) + "("
+                // queueString is "vh(/{0})/qu({1}) " so need to trim
+                + queueInfo.substring(0, queueInfo.length() - 1) + ")" + "] ";
+        return result;
+    }
+
+    private String getFilterLogString()
+    {
+        StringBuilder filterLogString = new StringBuilder();
+        String delimiter = ", ";
+        boolean hasEntries = false;
+        if (_filters != null && _filters.hasFilters())
+        {
+            filterLogString.append(_filters.toString());
+            hasEntries = true;
+        }
+
+        if (isBrowser())
+        {
+            if (hasEntries)
+            {
+                filterLogString.append(delimiter);
+            }
+            filterLogString.append("Browser");
+            hasEntries = true;
+        }
+
+        if (isDurable())
+        {
+            if (hasEntries)
+            {
+                filterLogString.append(delimiter);
+            }
+            filterLogString.append("Durable");
+            hasEntries = true;
+        }
+
+        return filterLogString.toString();
+    }
+
+    public LogSubject getLogSubject()
+    {
+        return (LogSubject) this;
+    }
+
+
+    public void flushBatched()
+    {
+        _session.getConnection().flush();
+    }
+
+    public long getBytesOut()
+    {
+        return _deliveredBytes.longValue();
+    }
+
+    public long getMessagesOut()
+    {
+        return _deliveredCount.longValue();
+    }
+
+    public long getUnacknowledgedBytes()
+    {
+        return _unacknowledgedBytes.longValue();
+    }
+
+    public long getUnacknowledgedMessages()
+    {
+        return _unacknowledgedCount.longValue();
+    }
+}
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/TransferMessageReference.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/TransferMessageReference.java
new file mode 100644
index 0000000000..0c04f22232
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/TransferMessageReference.java
@@ -0,0 +1,41 @@
+/*
+ *
+ * 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.v0_10;
+
+import org.apache.qpid.server.message.MessageReference;
+
+public class TransferMessageReference extends MessageReference<MessageTransferMessage>
+{
+    public TransferMessageReference(MessageTransferMessage message)
+    {
+        super(message);
+    }
+
+    protected void onReference(MessageTransferMessage message)
+    {
+        message.incrementReference();
+    }
+
+    protected void onRelease(MessageTransferMessage message)
+    {
+        message.decrementReference();
+    }
+}
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/WindowCreditManager.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/WindowCreditManager.java
new file mode 100644
index 0000000000..8e48741b91
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/WindowCreditManager.java
@@ -0,0 +1,207 @@
+/*
+ *
+ * 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.v0_10;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.server.flow.AbstractFlowCreditManager;
+
+public class WindowCreditManager extends AbstractFlowCreditManager implements FlowCreditManager_0_10
+{
+    private static final Logger LOGGER = Logger.getLogger(WindowCreditManager.class);
+
+    private volatile long _bytesCreditLimit;
+    private volatile long _messageCreditLimit;
+
+    private volatile long _bytesUsed;
+    private volatile long _messageUsed;
+
+     public WindowCreditManager()
+     {
+         this(0L, 0L);
+     }
+
+    public WindowCreditManager(long bytesCreditLimit, long messageCreditLimit)
+    {
+        _bytesCreditLimit = bytesCreditLimit;
+        _messageCreditLimit = messageCreditLimit;
+        setSuspended(!hasCredit());
+
+    }
+
+    public long getBytesCreditLimit()
+    {
+        return _bytesCreditLimit;
+    }
+
+    public long getMessageCreditLimit()
+    {
+        return _messageCreditLimit;
+    }
+
+    public synchronized void setCreditLimits(final long bytesCreditLimit, final long messageCreditLimit)
+    {
+        _bytesCreditLimit = bytesCreditLimit;
+        _messageCreditLimit = messageCreditLimit;
+
+        setSuspended(!hasCredit());
+
+    }
+
+
+    public long getMessageCredit()
+    {
+         return _messageCreditLimit == -1L
+                    ? Long.MAX_VALUE
+                    : _messageUsed < _messageCreditLimit ? _messageCreditLimit - _messageUsed : 0L;
+    }
+
+    public long getBytesCredit()
+    {
+        return _bytesCreditLimit == -1L
+                    ? Long.MAX_VALUE
+                    : _bytesUsed < _bytesCreditLimit ? _bytesCreditLimit - _bytesUsed : 0L;
+    }
+
+    public synchronized void restoreCredit(final long messageCredit, final long bytesCredit)
+    {
+        _messageUsed -= messageCredit;
+        if(_messageUsed < 0L)
+        {
+            LOGGER.error("Message credit used value was negative: "+ _messageUsed);
+            _messageUsed = 0;
+        }
+
+        boolean notifyIncrease = true;
+
+        if(_messageCreditLimit > 0L)
+        {
+            notifyIncrease = (_messageUsed != _messageCreditLimit);
+        }
+
+        _bytesUsed -= bytesCredit;
+        if(_bytesUsed < 0L)
+        {
+            LOGGER.error("Bytes credit used value was negative: "+ _messageUsed);
+            _bytesUsed = 0;
+        }
+
+        if(_bytesCreditLimit > 0L)
+        {
+            notifyIncrease = notifyIncrease && bytesCredit>0;
+
+            if(notifyIncrease)
+            {
+                notifyIncreaseBytesCredit();
+            }
+        }
+
+        setSuspended(!hasCredit());
+    }
+
+
+
+    public synchronized boolean hasCredit()
+    {
+        return (_bytesCreditLimit < 0L || _bytesCreditLimit > _bytesUsed)
+                && (_messageCreditLimit < 0L || _messageCreditLimit > _messageUsed);
+    }
+
+    public synchronized boolean useCreditForMessage(final long msgSize)
+    {
+        if(_messageCreditLimit >= 0L)
+        {
+            if(_messageUsed < _messageCreditLimit)
+            {
+                if(_bytesCreditLimit < 0L)
+                {
+                    _messageUsed++;
+
+                    return true;
+                }
+                else if(_bytesUsed + msgSize <= _bytesCreditLimit)
+                {
+                    _messageUsed++;
+                    _bytesUsed += msgSize;
+
+                    return true;
+                }
+                else
+                {
+                    return false;
+                }
+            }
+            else
+            {
+                setSuspended(true);
+                return false;
+            }
+        }
+        else if(_bytesCreditLimit >= 0L)
+        {
+            if(_bytesUsed + msgSize <= _bytesCreditLimit)
+            {
+                 _bytesUsed += msgSize;
+
+                return true;
+            }
+            else
+            {
+                return false;
+            }
+
+        }
+        else
+        {
+            return true;
+        }
+
+    }
+
+
+    public synchronized void addCredit(long count, long bytes)
+    {
+        if(bytes > 0)
+        {
+            _bytesCreditLimit += bytes;
+        }
+        else if(bytes == -1)
+        {
+            _bytesCreditLimit = -1;
+        }
+
+
+        if(count > 0)
+        {
+            _messageCreditLimit += count;
+        }
+        else if(count == -1)
+        {
+            _messageCreditLimit = -1;
+        }
+    }
+
+    public void clearCredit()
+    {
+        _bytesCreditLimit = 0l;
+        _messageCreditLimit = 0l;
+        setSuspended(true);
+    }
+}
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/resources/META-INF/services/org.apache.qpid.server.plugin.MessageConverter b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/resources/META-INF/services/org.apache.qpid.server.plugin.MessageConverter
new file mode 100644
index 0000000000..995b0fabdc
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/resources/META-INF/services/org.apache.qpid.server.plugin.MessageConverter
@@ -0,0 +1,19 @@
+#
+# 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.
+#
+org.apache.qpid.server.protocol.v0_10.MessageConverter_v0_10
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/resources/META-INF/services/org.apache.qpid.server.plugin.MessageMetaDataType b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/resources/META-INF/services/org.apache.qpid.server.plugin.MessageMetaDataType
new file mode 100644
index 0000000000..36118fe053
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/resources/META-INF/services/org.apache.qpid.server.plugin.MessageMetaDataType
@@ -0,0 +1,19 @@
+#
+# 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.
+#
+org.apache.qpid.server.protocol.v0_10.MessageMetaDataType_0_10
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/resources/META-INF/services/org.apache.qpid.server.plugin.ProtocolEngineCreator b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/resources/META-INF/services/org.apache.qpid.server.plugin.ProtocolEngineCreator
new file mode 100644
index 0000000000..3d94e26671
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/resources/META-INF/services/org.apache.qpid.server.plugin.ProtocolEngineCreator
@@ -0,0 +1,19 @@
+#
+# 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.
+#
+org.apache.qpid.server.protocol.v0_10.ProtocolEngineCreator_0_10
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/ServerSessionTest.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/ServerSessionTest.java
new file mode 100644
index 0000000000..421adb33a8
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/ServerSessionTest.java
@@ -0,0 +1,84 @@
+/*
+ * 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.v0_10;
+
+import org.apache.qpid.server.logging.RootMessageLogger;
+import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.logging.actors.GenericActor;
+import org.apache.qpid.server.model.Broker;
+import org.apache.qpid.server.util.BrokerTestHelper;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.test.utils.QpidTestCase;
+import org.apache.qpid.transport.Binary;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class ServerSessionTest extends QpidTestCase
+{
+
+    private VirtualHost _virtualHost;
+
+    @Override
+    public void setUp() throws Exception
+    {
+        super.setUp();
+        BrokerTestHelper.setUp();
+        _virtualHost = BrokerTestHelper.createVirtualHost(getName());
+        GenericActor.setDefaultMessageLogger(CurrentActor.get().getRootMessageLogger());
+    }
+
+    @Override
+    public void tearDown() throws Exception
+    {
+        try
+        {
+            if (_virtualHost != null)
+            {
+                _virtualHost.close();
+            }
+        }
+        finally
+        {
+            BrokerTestHelper.tearDown();
+            super.tearDown();
+        }
+    }
+
+    public void testCompareTo() throws Exception
+    {
+        final Broker broker = mock(Broker.class);
+        when(broker.getRootMessageLogger()).thenReturn(mock(RootMessageLogger.class));
+        ServerConnection connection = new ServerConnection(1, broker);
+        connection.setVirtualHost(_virtualHost);
+        ServerSession session1 = new ServerSession(connection, new ServerSessionDelegate(),
+                new Binary(getName().getBytes()), 0);
+
+        // create a session with the same name but on a different connection
+        ServerConnection connection2 = new ServerConnection(2, broker);
+        connection2.setVirtualHost(_virtualHost);
+        ServerSession session2 = new ServerSession(connection2, new ServerSessionDelegate(),
+                new Binary(getName().getBytes()), 0);
+
+        assertFalse("Unexpected compare result", session1.compareTo(session2) == 0);
+        assertEquals("Unexpected compare result", 0, session1.compareTo(session1));
+    }
+
+
+}
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/WindowCreditManagerTest.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/WindowCreditManagerTest.java
new file mode 100644
index 0000000000..1c4a694be6
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/WindowCreditManagerTest.java
@@ -0,0 +1,84 @@
+/*
+ *
+ * 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.v0_10;
+
+import org.apache.qpid.server.protocol.v0_10.WindowCreditManager;
+import org.apache.qpid.test.utils.QpidTestCase;
+
+public class WindowCreditManagerTest extends QpidTestCase
+{
+    private WindowCreditManager _creditManager;
+
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+        _creditManager = new WindowCreditManager();
+    }
+
+    /**
+     * Tests that after the credit limit is cleared (e.g. from a message.stop command), credit is
+     * restored (e.g. from completed MessageTransfer) without increasing the available credit, and
+     * more credit is added, that the 'used' count is correct and the proper values for bytes
+     * and message credit are returned along with appropriate 'hasCredit' results (QPID-3592).
+     */
+    public void testRestoreCreditDecrementsUsedCountAfterCreditClear()
+    {
+        assertEquals("unexpected credit value", 0, _creditManager.getMessageCredit());
+        assertEquals("unexpected credit value", 0, _creditManager.getBytesCredit());
+
+        //give some message credit
+        _creditManager.addCredit(1, 0);
+        assertFalse("Manager should not 'haveCredit' due to having 0 bytes credit", _creditManager.hasCredit());
+        assertEquals("unexpected credit value", 1, _creditManager.getMessageCredit());
+        assertEquals("unexpected credit value", 0, _creditManager.getBytesCredit());
+
+        //give some bytes credit
+        _creditManager.addCredit(0, 1);
+        assertTrue("Manager should 'haveCredit'", _creditManager.hasCredit());
+        assertEquals("unexpected credit value", 1, _creditManager.getMessageCredit());
+        assertEquals("unexpected credit value", 1, _creditManager.getBytesCredit());
+
+        //use all the credit
+        _creditManager.useCreditForMessage(1);
+        assertEquals("unexpected credit value", 0, _creditManager.getBytesCredit());
+        assertEquals("unexpected credit value", 0, _creditManager.getMessageCredit());
+        assertFalse("Manager should not 'haveCredit'", _creditManager.hasCredit());
+
+        //clear credit out (eg from a message.stop command)
+        _creditManager.clearCredit();
+        assertEquals("unexpected credit value", 0, _creditManager.getBytesCredit());
+        assertEquals("unexpected credit value", 0, _creditManager.getMessageCredit());
+        assertFalse("Manager should not 'haveCredit'", _creditManager.hasCredit());
+
+        //restore credit (e.g the original message transfer command got completed)
+        //this should not increase credit, because it is now limited to 0
+        _creditManager.restoreCredit(1, 1);
+        assertEquals("unexpected credit value", 0, _creditManager.getBytesCredit());
+        assertEquals("unexpected credit value", 0, _creditManager.getMessageCredit());
+        assertFalse("Manager should not 'haveCredit'", _creditManager.hasCredit());
+
+        //give more credit to open the window again
+        _creditManager.addCredit(1, 1);
+        assertEquals("unexpected credit value", 1, _creditManager.getBytesCredit());
+        assertEquals("unexpected credit value", 1, _creditManager.getMessageCredit());
+        assertTrue("Manager should 'haveCredit'", _creditManager.hasCredit());
+    }
+}
diff --git a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Subscription_1_0.java b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Subscription_1_0.java
index 28d212c93d..b9695ba87a 100644
--- a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Subscription_1_0.java
+++ b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Subscription_1_0.java
@@ -20,10 +20,7 @@
  */
 package org.apache.qpid.server.protocol.v1_0;
 
-import java.io.EOFException;
 import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -38,47 +35,27 @@ import org.apache.qpid.amqp_1_0.type.AmqpErrorException;
 import org.apache.qpid.amqp_1_0.type.Binary;
 import org.apache.qpid.amqp_1_0.type.DeliveryState;
 import org.apache.qpid.amqp_1_0.type.Outcome;
-import org.apache.qpid.amqp_1_0.type.Section;
-import org.apache.qpid.amqp_1_0.type.Symbol;
-import org.apache.qpid.amqp_1_0.type.UnsignedByte;
 import org.apache.qpid.amqp_1_0.type.UnsignedInteger;
 import org.apache.qpid.amqp_1_0.type.codec.AMQPDescribedTypeRegistry;
 import org.apache.qpid.amqp_1_0.type.messaging.Accepted;
-import org.apache.qpid.amqp_1_0.type.messaging.AmqpValue;
-import org.apache.qpid.amqp_1_0.type.messaging.ApplicationProperties;
-import org.apache.qpid.amqp_1_0.type.messaging.Data;
 import org.apache.qpid.amqp_1_0.type.messaging.Header;
 import org.apache.qpid.amqp_1_0.type.messaging.Modified;
-import org.apache.qpid.amqp_1_0.type.messaging.Properties;
 import org.apache.qpid.amqp_1_0.type.messaging.Released;
 import org.apache.qpid.amqp_1_0.type.messaging.Source;
 import org.apache.qpid.amqp_1_0.type.messaging.StdDistMode;
 import org.apache.qpid.amqp_1_0.type.transaction.TransactionalState;
 import org.apache.qpid.amqp_1_0.type.transport.SenderSettleMode;
 import org.apache.qpid.amqp_1_0.type.transport.Transfer;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.BasicContentHeaderProperties;
-import org.apache.qpid.framing.FieldTable;
 import org.apache.qpid.server.plugin.MessageConverter;
 import org.apache.qpid.server.protocol.MessageConverterRegistry;
-import org.apache.qpid.typedmessage.TypedBytesContentReader;
-import org.apache.qpid.typedmessage.TypedBytesFormatException;
 import org.apache.qpid.server.filter.FilterManager;
 import org.apache.qpid.server.logging.LogActor;
-import org.apache.qpid.server.protocol.v0_8.AMQMessage;
-import org.apache.qpid.server.protocol.v0_10.MessageTransferMessage;
 import org.apache.qpid.server.message.ServerMessage;
 import org.apache.qpid.server.protocol.AMQSessionModel;
 import org.apache.qpid.server.queue.AMQQueue;
 import org.apache.qpid.server.queue.QueueEntry;
-import org.apache.qpid.server.store.StoreFuture;
-import org.apache.qpid.server.store.StoredMessage;
 import org.apache.qpid.server.subscription.Subscription;
 import org.apache.qpid.server.txn.ServerTransaction;
-import org.apache.qpid.transport.DeliveryProperties;
-import org.apache.qpid.transport.MessageDeliveryMode;
-import org.apache.qpid.transport.MessageProperties;
-import org.apache.qpid.transport.codec.BBDecoder;
 
 class Subscription_1_0 implements Subscription
 {
diff --git a/qpid/java/broker-plugins/amqp-msg-conv-0-10-to-1-0/build.xml b/qpid/java/broker-plugins/amqp-msg-conv-0-10-to-1-0/build.xml
index 80d832fcce..b3f5214b5e 100644
--- a/qpid/java/broker-plugins/amqp-msg-conv-0-10-to-1-0/build.xml
+++ b/qpid/java/broker-plugins/amqp-msg-conv-0-10-to-1-0/build.xml
@@ -17,11 +17,11 @@
  - under the License.
  -->
 <project name="Qpid Broker-Plugins AMQP 0.8 to 1.0 MessageConversion" default="build">
-    <property name="module.depends" value="common broker amqp-1-0-common broker-plugins/amqp-1-0-protocol" />
+    <property name="module.depends" value="common broker amqp-1-0-common broker-plugins/amqp-1-0-protocol broker-plugins/amqp-0-10-protocol" />
     <property name="module.test.depends" value="common/tests broker/tests" />
 
     <property name="module.genpom" value="true"/>
-    <property name="module.genpom.args" value="-Sqpid-common=provided -Sqpid-broker=provided -Sqpid-amqp-1-0-common=provided -Sqpid-broker-plugins-amqp-1-0-protocol=provided"/>
+    <property name="module.genpom.args" value="-Sqpid-common=provided -Sqpid-broker=provided -Sqpid-amqp-1-0-common=provided -Sqpid-broker-plugins-amqp-0-10-protocol=provided -Sqpid-broker-plugins-amqp-1-0-protocol=provided"/>
     <property name="broker-plugins-amqp-msg-conv-0-8-to-1-0-libs" value="" />
 
     <property name="broker.plugin" value="true"/>
diff --git a/qpid/java/broker-plugins/amqp-msg-conv-0-8-to-0-10/build.xml b/qpid/java/broker-plugins/amqp-msg-conv-0-8-to-0-10/build.xml
new file mode 100644
index 0000000000..abbfaa644e
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-msg-conv-0-8-to-0-10/build.xml
@@ -0,0 +1,33 @@
+<!--
+ - 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.
+ -->
+<project name="Qpid Broker-Plugins AMQP 0.8 to 0.10 MessageConversion" default="build">
+    <property name="module.depends" value="common broker broker-plugins/amqp-0-10-protocol" />
+    <property name="module.test.depends" value="common/tests broker/tests" />
+
+    <property name="module.genpom" value="true"/>
+    <property name="module.genpom.args" value="-Sqpid-common=provided -Sqpid-broker=provided -Sqpid-broker-plugins-amqp-0-10-protocol=provided"/>
+    <property name="broker-plugins-amqp-msg-conv-0-8-to-0-10-libs" value="" />
+
+    <property name="broker.plugin" value="true"/>
+
+    <import file="../../module.xml" />
+
+    <target name="bundle" depends="bundle-tasks"/>
+
+</project>
diff --git a/qpid/java/broker-plugins/amqp-msg-conv-0-8-to-0-10/src/main/java/org/apache/qpid/server/protocol/converter/v0_8_v0_10/MessageConverter_0_10_to_0_8.java b/qpid/java/broker-plugins/amqp-msg-conv-0-8-to-0-10/src/main/java/org/apache/qpid/server/protocol/converter/v0_8_v0_10/MessageConverter_0_10_to_0_8.java
new file mode 100644
index 0000000000..c14896079f
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-msg-conv-0-8-to-0-10/src/main/java/org/apache/qpid/server/protocol/converter/v0_8_v0_10/MessageConverter_0_10_to_0_8.java
@@ -0,0 +1,271 @@
+package org.apache.qpid.server.protocol.converter.v0_8_v0_10;
+
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.qpid.AMQPInvalidClassException;
+import org.apache.qpid.exchange.ExchangeDefaults;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.BasicContentHeaderProperties;
+import org.apache.qpid.framing.ContentHeaderBody;
+import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.framing.abstraction.MessagePublishInfo;
+import org.apache.qpid.server.exchange.Exchange;
+import org.apache.qpid.server.plugin.MessageConverter;
+import org.apache.qpid.server.protocol.v0_10.MessageTransferMessage;
+import org.apache.qpid.server.protocol.v0_8.AMQMessage;
+import org.apache.qpid.server.protocol.v0_8.MessageMetaData;
+import org.apache.qpid.server.store.StoreFuture;
+import org.apache.qpid.server.store.StoredMessage;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.transport.DeliveryProperties;
+import org.apache.qpid.transport.Header;
+import org.apache.qpid.transport.MessageDeliveryMode;
+import org.apache.qpid.transport.MessageProperties;
+import org.apache.qpid.transport.ReplyTo;
+
+public class MessageConverter_0_10_to_0_8 implements MessageConverter<MessageTransferMessage, AMQMessage>
+{
+    private static final int BASIC_CLASS_ID = 60;
+
+    public static BasicContentHeaderProperties convertContentHeaderProperties(MessageTransferMessage messageTransferMessage,
+                                                                              VirtualHost vhost)
+    {
+        BasicContentHeaderProperties props = new BasicContentHeaderProperties();
+
+        Header header = messageTransferMessage.getHeader();
+        DeliveryProperties deliveryProps = header.getDeliveryProperties();
+        MessageProperties messageProps = header.getMessageProperties();
+
+        if(deliveryProps != null)
+        {
+            if(deliveryProps.hasDeliveryMode())
+            {
+                props.setDeliveryMode((byte) (deliveryProps.getDeliveryMode() == MessageDeliveryMode.PERSISTENT
+                                              ? BasicContentHeaderProperties.PERSISTENT
+                                              : BasicContentHeaderProperties.NON_PERSISTENT));
+            }
+            if(deliveryProps.hasExpiration())
+            {
+                props.setExpiration(deliveryProps.getExpiration());
+            }
+            if(deliveryProps.hasPriority())
+            {
+                props.setPriority((byte) deliveryProps.getPriority().getValue());
+            }
+            if(deliveryProps.hasTimestamp())
+            {
+                props.setTimestamp(deliveryProps.getTimestamp());
+            }
+        }
+        if(messageProps != null)
+        {
+            if(messageProps.hasAppId())
+            {
+                props.setAppId(new AMQShortString(messageProps.getAppId()));
+            }
+            if(messageProps.hasContentType())
+            {
+                props.setContentType(messageProps.getContentType());
+            }
+            if(messageProps.hasCorrelationId())
+            {
+                props.setCorrelationId(new AMQShortString(messageProps.getCorrelationId()));
+            }
+            if(messageProps.hasContentEncoding())
+            {
+                props.setEncoding(messageProps.getContentEncoding());
+            }
+            if(messageProps.hasMessageId())
+            {
+                props.setMessageId("ID:" + messageProps.getMessageId().toString());
+            }
+            if(messageProps.hasReplyTo())
+            {
+                ReplyTo replyTo = messageProps.getReplyTo();
+                String exchangeName = replyTo.getExchange();
+                String routingKey = replyTo.getRoutingKey();
+                if(exchangeName == null)
+                {
+                    exchangeName = "";
+                }
+
+                Exchange exchange = vhost.getExchange(exchangeName);
+                String exchangeClass = exchange == null
+                                            ? ExchangeDefaults.DIRECT_EXCHANGE_CLASS.asString()
+                                            : exchange.getType().getName().asString();
+                props.setReplyTo(exchangeClass + "://" + exchangeName + "//?routingkey='" + (routingKey == null
+                                                                                             ? ""
+                                                                                             : routingKey + "'"));
+
+            }
+            if(messageProps.hasUserId())
+            {
+                props.setUserId(new AMQShortString(messageProps.getUserId()));
+            }
+
+            if(messageProps.hasApplicationHeaders())
+            {
+                Map<String, Object> appHeaders = new HashMap<String, Object>(messageProps.getApplicationHeaders());
+                if(messageProps.getApplicationHeaders().containsKey("x-jms-type"))
+                {
+                    props.setType(String.valueOf(appHeaders.remove("x-jms-type")));
+                }
+
+                FieldTable ft = new FieldTable();
+                for(Map.Entry<String, Object> entry : appHeaders.entrySet())
+                {
+                    try
+                    {
+                        ft.put(new AMQShortString(entry.getKey()), entry.getValue());
+                    }
+                    catch (AMQPInvalidClassException e)
+                    {
+                        // TODO
+                        // log here, but ignore - just can;t convert
+                    }
+                }
+                props.setHeaders(ft);
+
+            }
+        }
+
+        return props;
+    }
+
+    @Override
+    public Class<MessageTransferMessage> getInputClass()
+    {
+        return MessageTransferMessage.class;
+    }
+
+    @Override
+    public Class<AMQMessage> getOutputClass()
+    {
+        return AMQMessage.class;
+    }
+
+    @Override
+    public AMQMessage convert(MessageTransferMessage message, VirtualHost vhost)
+    {
+        return new AMQMessage(convertToStoredMessage(message, vhost));
+    }
+
+    private StoredMessage<MessageMetaData> convertToStoredMessage(final MessageTransferMessage message,
+                                                                  VirtualHost vhost)
+    {
+        final MessageMetaData metaData = convertMetaData(message, vhost);
+        return new StoredMessage<org.apache.qpid.server.protocol.v0_8.MessageMetaData>()
+        {
+            @Override
+            public MessageMetaData getMetaData()
+            {
+                return metaData;
+            }
+
+            @Override
+            public long getMessageNumber()
+            {
+                return message.getMessageNumber();
+            }
+
+            @Override
+            public void addContent(int offsetInMessage, ByteBuffer src)
+            {
+                throw new UnsupportedOperationException();
+            }
+
+            @Override
+            public int getContent(int offsetInMessage, ByteBuffer dst)
+            {
+                return message.getContent(dst, offsetInMessage);
+            }
+
+            @Override
+            public ByteBuffer getContent(int offsetInMessage, int size)
+            {
+                return message.getContent(offsetInMessage, size);
+            }
+
+            @Override
+            public StoreFuture flushToStore()
+            {
+                return StoreFuture.IMMEDIATE_FUTURE;
+            }
+
+            @Override
+            public void remove()
+            {
+                throw new UnsupportedOperationException();
+            }
+        };
+    }
+
+    private MessageMetaData convertMetaData(MessageTransferMessage message, VirtualHost vhost)
+    {
+        return new MessageMetaData(convertPublishBody(message),
+                convertContentHeaderBody(message, vhost),
+                1,
+                message.getArrivalTime());
+    }
+
+    private ContentHeaderBody convertContentHeaderBody(MessageTransferMessage message, VirtualHost vhost)
+    {
+        BasicContentHeaderProperties props = convertContentHeaderProperties(message, vhost);
+        ContentHeaderBody chb = new ContentHeaderBody(props, BASIC_CLASS_ID);
+        chb.setBodySize(message.getSize());
+        return chb;
+    }
+
+    private MessagePublishInfo convertPublishBody(MessageTransferMessage message)
+    {
+        DeliveryProperties delvProps = message.getHeader().getDeliveryProperties();
+        final AMQShortString exchangeName = (delvProps == null || delvProps.getExchange() == null)
+                                            ? null
+                                            : new AMQShortString(delvProps.getExchange());
+        final AMQShortString routingKey = (delvProps == null || delvProps.getRoutingKey() == null)
+                                          ? null
+                                          : new AMQShortString(delvProps.getRoutingKey());
+        final boolean immediate = delvProps != null && delvProps.getImmediate();
+        final boolean mandatory = delvProps != null && !delvProps.getDiscardUnroutable();
+
+        return new MessagePublishInfo()
+        {
+            @Override
+            public AMQShortString getExchange()
+            {
+                return exchangeName;
+            }
+
+            @Override
+            public void setExchange(AMQShortString exchange)
+            {
+                throw new UnsupportedOperationException();
+            }
+
+            @Override
+            public boolean isImmediate()
+            {
+                return immediate;
+            }
+
+            @Override
+            public boolean isMandatory()
+            {
+                return mandatory;
+            }
+
+            @Override
+            public AMQShortString getRoutingKey()
+            {
+                return routingKey;
+            }
+        };
+    }
+
+    @Override
+    public String getType()
+    {
+        return "v0-10 to v0-8";
+    }
+}
diff --git a/qpid/java/broker-plugins/amqp-msg-conv-0-8-to-0-10/src/main/java/org/apache/qpid/server/protocol/converter/v0_8_v0_10/MessageConverter_0_8_to_0_10.java b/qpid/java/broker-plugins/amqp-msg-conv-0-8-to-0-10/src/main/java/org/apache/qpid/server/protocol/converter/v0_8_v0_10/MessageConverter_0_8_to_0_10.java
new file mode 100644
index 0000000000..e1e8fbd9d3
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-msg-conv-0-8-to-0-10/src/main/java/org/apache/qpid/server/protocol/converter/v0_8_v0_10/MessageConverter_0_8_to_0_10.java
@@ -0,0 +1,226 @@
+/*
+ *
+ * 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.converter.v0_8_v0_10;
+
+import java.net.URISyntaxException;
+import java.nio.ByteBuffer;
+import java.util.Map;
+import java.util.UUID;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.BasicContentHeaderProperties;
+import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.server.plugin.MessageConverter;
+import org.apache.qpid.server.protocol.v0_10.MessageMetaData_0_10;
+import org.apache.qpid.server.protocol.v0_10.MessageTransferMessage;
+import org.apache.qpid.server.protocol.v0_8.AMQMessage;
+import org.apache.qpid.server.store.StoreFuture;
+import org.apache.qpid.server.store.StoredMessage;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.transport.DeliveryProperties;
+import org.apache.qpid.transport.Header;
+import org.apache.qpid.transport.MessageDeliveryPriority;
+import org.apache.qpid.transport.MessageProperties;
+import org.apache.qpid.transport.ReplyTo;
+import org.apache.qpid.url.AMQBindingURL;
+
+public class MessageConverter_0_8_to_0_10  implements MessageConverter<AMQMessage, MessageTransferMessage>
+{
+    @Override
+    public Class<AMQMessage> getInputClass()
+    {
+        return AMQMessage.class;
+    }
+
+    @Override
+    public Class<MessageTransferMessage> getOutputClass()
+    {
+        return MessageTransferMessage.class;
+    }
+
+    @Override
+    public MessageTransferMessage convert(AMQMessage message_0_8, VirtualHost vhost)
+    {
+        return new MessageTransferMessage(convertToStoredMessage(message_0_8), null);
+    }
+
+    private StoredMessage<MessageMetaData_0_10> convertToStoredMessage(final AMQMessage message_0_8)
+    {
+        final MessageMetaData_0_10 messageMetaData_0_10 = convertMetaData(message_0_8);
+        return new StoredMessage<MessageMetaData_0_10>()
+        {
+            @Override
+            public MessageMetaData_0_10 getMetaData()
+            {
+                return messageMetaData_0_10;
+            }
+
+            @Override
+            public long getMessageNumber()
+            {
+                return message_0_8.getMessageNumber();
+            }
+
+            @Override
+            public void addContent(int offsetInMessage, ByteBuffer src)
+            {
+                throw new UnsupportedOperationException();
+            }
+
+            @Override
+            public int getContent(int offsetInMessage, ByteBuffer dst)
+            {
+                return message_0_8.getContent(dst, offsetInMessage);
+            }
+
+            @Override
+            public ByteBuffer getContent(int offsetInMessage, int size)
+            {
+                return message_0_8.getContent(offsetInMessage, size);
+            }
+
+            @Override
+            public StoreFuture flushToStore()
+            {
+                return StoreFuture.IMMEDIATE_FUTURE;
+            }
+
+            @Override
+            public void remove()
+            {
+                throw new UnsupportedOperationException();
+            }
+        };
+    }
+
+    private MessageMetaData_0_10 convertMetaData(AMQMessage message_0_8)
+    {
+        DeliveryProperties deliveryProps = new DeliveryProperties();
+        MessageProperties messageProps = new MessageProperties();
+
+        int size = (int) message_0_8.getSize();
+        ByteBuffer body = ByteBuffer.allocate(size);
+        message_0_8.getContent(body, 0);
+        body.flip();
+
+        BasicContentHeaderProperties properties =
+                (BasicContentHeaderProperties) message_0_8.getContentHeaderBody().getProperties();
+
+        final AMQShortString exchange = message_0_8.getMessagePublishInfo().getExchange();
+        if(exchange != null)
+        {
+            deliveryProps.setExchange(exchange.toString());
+        }
+
+        deliveryProps.setExpiration(message_0_8.getExpiration());
+        deliveryProps.setImmediate(message_0_8.isImmediate());
+        deliveryProps.setPriority(MessageDeliveryPriority.get(properties.getPriority()));
+        deliveryProps.setRoutingKey(message_0_8.getRoutingKey());
+        deliveryProps.setTimestamp(properties.getTimestamp());
+
+        messageProps.setContentEncoding(properties.getEncodingAsString());
+        messageProps.setContentLength(size);
+        if(properties.getAppId() != null)
+        {
+            messageProps.setAppId(properties.getAppId().getBytes());
+        }
+        messageProps.setContentType(properties.getContentTypeAsString());
+        if(properties.getCorrelationId() != null)
+        {
+            messageProps.setCorrelationId(properties.getCorrelationId().getBytes());
+        }
+
+        if(properties.getReplyTo() != null && properties.getReplyTo().length() != 0)
+        {
+            String origReplyToString = properties.getReplyTo().asString();
+            ReplyTo replyTo = new ReplyTo();
+            // if the string looks like a binding URL, then attempt to parse it...
+            try
+            {
+                AMQBindingURL burl = new AMQBindingURL(origReplyToString);
+                AMQShortString routingKey = burl.getRoutingKey();
+                if(routingKey != null)
+                {
+                    replyTo.setRoutingKey(routingKey.asString());
+                }
+
+                AMQShortString exchangeName = burl.getExchangeName();
+                if(exchangeName != null)
+                {
+                    replyTo.setExchange(exchangeName.asString());
+                }
+            }
+            catch (URISyntaxException e)
+            {
+                replyTo.setRoutingKey(origReplyToString);
+            }
+            messageProps.setReplyTo(replyTo);
+
+        }
+
+        if(properties.getMessageId() != null)
+        {
+            try
+            {
+                String messageIdAsString = properties.getMessageIdAsString();
+                if(messageIdAsString.startsWith("ID:"))
+                {
+                    messageIdAsString = messageIdAsString.substring(3);
+                }
+                UUID uuid = UUID.fromString(messageIdAsString);
+                messageProps.setMessageId(uuid);
+            }
+            catch(IllegalArgumentException e)
+            {
+                // ignore - can't parse
+            }
+        }
+
+
+
+        if(properties.getUserId() != null)
+        {
+            messageProps.setUserId(properties.getUserId().getBytes());
+        }
+
+        FieldTable fieldTable = properties.getHeaders();
+
+        Map<String, Object> appHeaders = FieldTable.convertToMap(fieldTable);
+
+        if(properties.getType() != null)
+        {
+            appHeaders.put("x-jms-type", properties.getTypeAsString());
+        }
+
+
+        messageProps.setApplicationHeaders(appHeaders);
+
+        Header header = new Header(deliveryProps, messageProps, null);
+
+
+        return new MessageMetaData_0_10(header, size, message_0_8.getArrivalTime());
+    }
+
+    @Override
+    public String getType()
+    {
+        return "v0-8 to v0-10";
+    }
+}
diff --git a/qpid/java/broker-plugins/amqp-msg-conv-0-8-to-0-10/src/main/resources/META-INF/services/org.apache.qpid.server.plugin.MessageConverter b/qpid/java/broker-plugins/amqp-msg-conv-0-8-to-0-10/src/main/resources/META-INF/services/org.apache.qpid.server.plugin.MessageConverter
new file mode 100644
index 0000000000..5aeef786ae
--- /dev/null
+++ b/qpid/java/broker-plugins/amqp-msg-conv-0-8-to-0-10/src/main/resources/META-INF/services/org.apache.qpid.server.plugin.MessageConverter
@@ -0,0 +1,20 @@
+#
+# 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.
+#
+org.apache.qpid.server.protocol.converter.v0_8_v0_10.MessageConverter_0_8_to_0_10
+org.apache.qpid.server.protocol.converter.v0_8_v0_10.MessageConverter_0_10_to_0_8