Added AMQP 0-9-1 support

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

8 files changed, 657 insertions, 45 deletions

diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/AccessRequestHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/AccessRequestHandler.java
index e64eaeae76..d587ef0c16 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/handler/AccessRequestHandler.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/handler/AccessRequestHandler.java
@@ -1,31 +1,34 @@
 package org.apache.qpid.server.handler;

-/*
- * 
- * 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.
- * 
- */
-
+/*

+ * 

+ * 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.

+ * 

+ */

+

 

 import org.apache.qpid.framing.*;

+import org.apache.qpid.framing.amqp_8_0.MethodRegistry_8_0;

+import org.apache.qpid.framing.amqp_0_9.MethodRegistry_0_9;

 import org.apache.qpid.server.state.StateAwareMethodListener;

 import org.apache.qpid.server.state.AMQStateManager;

 import org.apache.qpid.server.protocol.AMQProtocolSession;

 import org.apache.qpid.AMQException;

+import org.apache.qpid.protocol.AMQConstant;

 

 /**

  * @author Apache Software Foundation

@@ -54,7 +57,20 @@ public class AccessRequestHandler implements StateAwareMethodListener<AccessRequ
 

         // We don't implement access control class, but to keep clients happy that expect it

         // always use the "0" ticket.

-        AccessRequestOkBody response = methodRegistry.createAccessRequestOkBody(0);

+        AccessRequestOkBody response;

+        if(methodRegistry instanceof MethodRegistry_0_9)

+        {

+            response = ((MethodRegistry_0_9)methodRegistry).createAccessRequestOkBody(0);

+        }

+        else if(methodRegistry instanceof MethodRegistry_8_0)

+        {

+            response = ((MethodRegistry_8_0)methodRegistry).createAccessRequestOkBody(0);

+        }

+        else

+        {

+            throw new AMQException(AMQConstant.COMMAND_INVALID, "AccessRequest not present in AMQP versions other than 0-8, 0-9");

+        }

+

 

         session.writeFrame(response.generateFrame(channelId));

     }

diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRecoverSyncMethodHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRecoverSyncMethodHandler.java
index 2c264c3d45..f9feada6fe 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRecoverSyncMethodHandler.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRecoverSyncMethodHandler.java
@@ -1,25 +1,25 @@
 package org.apache.qpid.server.handler;

-/*
- * 
- * 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.
- * 
- */
-
+/*

+ * 

+ * 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.

+ * 

+ */

+

 

 import org.apache.log4j.Logger;

 

@@ -27,6 +27,7 @@ import org.apache.qpid.framing.BasicRecoverBody;
 import org.apache.qpid.framing.ProtocolVersion;

 import org.apache.qpid.framing.AMQMethodBody;

 import org.apache.qpid.framing.BasicRecoverSyncBody;

+import org.apache.qpid.framing.amqp_0_91.MethodRegistry_0_91;

 import org.apache.qpid.framing.amqp_0_9.MethodRegistry_0_9;

 import org.apache.qpid.framing.amqp_8_0.MethodRegistry_8_0;

 import org.apache.qpid.server.state.StateAwareMethodListener;

@@ -70,6 +71,13 @@ public class BasicRecoverSyncMethodHandler implements StateAwareMethodListener<B
             session.writeFrame(recoverOk.generateFrame(channelId));

 

         }

+        else if(session.getProtocolVersion().equals(ProtocolVersion.v0_91))

+        {

+            MethodRegistry_0_91 methodRegistry = (MethodRegistry_0_91) session.getMethodRegistry();

+            AMQMethodBody recoverOk = methodRegistry.createBasicRecoverSyncOkBody();

+            session.writeFrame(recoverOk.generateFrame(channelId));

+

+        }

 

     }

 }

diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelOpenHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelOpenHandler.java
index 9918013888..f5c2b93f46 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelOpenHandler.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelOpenHandler.java
@@ -27,6 +27,7 @@ import java.io.IOException;
 
 import org.apache.qpid.AMQException;
 import org.apache.qpid.framing.*;
+import org.apache.qpid.framing.amqp_0_91.MethodRegistry_0_91;
 import org.apache.qpid.framing.amqp_0_9.MethodRegistry_0_9;
 import org.apache.qpid.framing.amqp_8_0.MethodRegistry_8_0;
 import org.apache.qpid.protocol.AMQConstant;
@@ -102,6 +103,30 @@ public class ChannelOpenHandler implements StateAwareMethodListener<ChannelOpenB
             
             response = methodRegistry.createChannelOpenOkBody(channelName);
         }
+        else if(pv.equals(ProtocolVersion.v0_91))
+        {
+            MethodRegistry_0_91 methodRegistry = (MethodRegistry_0_91) MethodRegistry.getMethodRegistry(ProtocolVersion.v0_91);
+            UUID uuid = UUID.randomUUID();
+            ByteArrayOutputStream output = new ByteArrayOutputStream();
+            DataOutputStream dataOut = new DataOutputStream(output);
+            try
+            {
+                dataOut.writeLong(uuid.getMostSignificantBits());
+                dataOut.writeLong(uuid.getLeastSignificantBits());
+                dataOut.flush();
+                dataOut.close();
+            }
+            catch (IOException e)
+            {
+                // This *really* shouldn't happen as we're not doing any I/O
+                throw new RuntimeException("I/O exception when writing to byte array", e);
+            }
+
+            // should really associate this channelId to the session
+            byte[] channelName = output.toByteArray();
+
+            response = methodRegistry.createChannelOpenOkBody(channelName);
+        }
         else
         {
             throw new AMQException(AMQConstant.INTERNAL_ERROR, "Got channel open for protocol version not catered for: " + pv, null);
diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionOpenMethodHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionOpenMethodHandler.java
index 31bafac0a2..b5194084f2 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionOpenMethodHandler.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionOpenMethodHandler.java
@@ -22,6 +22,9 @@ package org.apache.qpid.server.handler;
 
 import org.apache.qpid.AMQException;
 import org.apache.qpid.framing.*;
+import org.apache.qpid.framing.amqp_0_91.MethodRegistry_0_91;
+import org.apache.qpid.framing.amqp_0_9.MethodRegistry_0_9;
+import org.apache.qpid.framing.amqp_8_0.MethodRegistry_8_0;
 import org.apache.qpid.protocol.AMQConstant;
 import org.apache.qpid.server.protocol.AMQProtocolSession;
 import org.apache.qpid.server.state.AMQState;
@@ -86,10 +89,10 @@ public class ConnectionOpenMethodHandler implements StateAwareMethodListener<Con
             if (session.getContextKey() == null)
             {
                 session.setContextKey(generateClientID());
-            }
+            }            
 
             MethodRegistry methodRegistry = session.getMethodRegistry();
-            AMQMethodBody responseBody = methodRegistry.createConnectionOpenOkBody(body.getVirtualHost());
+            AMQMethodBody responseBody =  methodRegistry.createConnectionOpenOkBody(body.getVirtualHost());            
 
             stateManager.changeState(AMQState.CONNECTION_OPEN);
 
diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ServerMethodDispatcherImpl.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ServerMethodDispatcherImpl.java
index 9475b83c8f..e290afcde3 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/handler/ServerMethodDispatcherImpl.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ServerMethodDispatcherImpl.java
@@ -59,6 +59,14 @@ public class ServerMethodDispatcherImpl implements MethodDispatcher
                                              return new ServerMethodDispatcherImpl_0_9(stateManager);

                                          }

                                      });

+            _dispatcherFactories.put(ProtocolVersion.v0_91,

+                         new DispatcherFactory()

+                         {

+                             public MethodDispatcher createMethodDispatcher(AMQStateManager stateManager)

+                             {

+                                 return new ServerMethodDispatcherImpl_0_91(stateManager);

+                             }

+                         });

 

         }

 

diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ServerMethodDispatcherImpl_0_91.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ServerMethodDispatcherImpl_0_91.java
new file mode 100644
index 0000000000..32cd4c4e9f
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ServerMethodDispatcherImpl_0_91.java
@@ -0,0 +1,168 @@
+/*

+ *

+ * 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.handler;

+

+

+import org.apache.qpid.framing.*;

+import org.apache.qpid.framing.amqp_0_91.MethodDispatcher_0_91;

+import org.apache.qpid.server.state.AMQStateManager;

+import org.apache.qpid.AMQException;

+

+

+public class ServerMethodDispatcherImpl_0_91

+        extends ServerMethodDispatcherImpl

+        implements MethodDispatcher_0_91

+

+{

+

+    private static final BasicRecoverSyncMethodHandler _basicRecoverSyncMethodHandler =

+            BasicRecoverSyncMethodHandler.getInstance();

+    private static final QueueUnbindHandler _queueUnbindHandler =

+            QueueUnbindHandler.getInstance();

+

+

+    public ServerMethodDispatcherImpl_0_91(AMQStateManager stateManager)

+    {

+        super(stateManager);

+    }

+

+    public boolean dispatchBasicRecoverSync(BasicRecoverSyncBody body, int channelId) throws AMQException

+    {

+        _basicRecoverSyncMethodHandler.methodReceived(getStateManager(), body, channelId);

+        return true;

+    }

+

+    public boolean dispatchBasicRecoverSyncOk(BasicRecoverSyncOkBody body, int channelId) throws AMQException

+    {

+        throw new UnexpectedMethodException(body);

+    }

+

+    public boolean dispatchChannelOk(ChannelOkBody body, int channelId) throws AMQException

+    {

+        return false;

+    }

+

+    public boolean dispatchChannelPing(ChannelPingBody body, int channelId) throws AMQException

+    {

+        return false;

+    }

+

+    public boolean dispatchChannelPong(ChannelPongBody body, int channelId) throws AMQException

+    {

+        return false;

+    }

+

+    public boolean dispatchChannelResume(ChannelResumeBody body, int channelId) throws AMQException

+    {

+        return false;

+    }

+

+    public boolean dispatchMessageAppend(MessageAppendBody body, int channelId) throws AMQException

+    {

+        return false;

+    }

+

+    public boolean dispatchMessageCancel(MessageCancelBody body, int channelId) throws AMQException

+    {

+        return false;

+    }

+

+    public boolean dispatchMessageCheckpoint(MessageCheckpointBody body, int channelId) throws AMQException

+    {

+        return false;

+    }

+

+    public boolean dispatchMessageClose(MessageCloseBody body, int channelId) throws AMQException

+    {

+        return false;

+    }

+

+    public boolean dispatchMessageConsume(MessageConsumeBody body, int channelId) throws AMQException

+    {

+        return false;

+    }

+

+    public boolean dispatchMessageEmpty(MessageEmptyBody body, int channelId) throws AMQException

+    {

+        return false;

+    }

+

+    public boolean dispatchMessageGet(MessageGetBody body, int channelId) throws AMQException

+    {

+        return false;

+    }

+

+    public boolean dispatchMessageOffset(MessageOffsetBody body, int channelId) throws AMQException

+    {

+        return false;

+    }

+

+    public boolean dispatchMessageOk(MessageOkBody body, int channelId) throws AMQException

+    {

+        return false;

+    }

+

+    public boolean dispatchMessageOpen(MessageOpenBody body, int channelId) throws AMQException

+    {

+        return false;

+    }

+

+    public boolean dispatchMessageQos(MessageQosBody body, int channelId) throws AMQException

+    {

+        return false;

+    }

+

+    public boolean dispatchMessageRecover(MessageRecoverBody body, int channelId) throws AMQException

+    {

+        return false;

+    }

+

+    public boolean dispatchMessageReject(MessageRejectBody body, int channelId) throws AMQException

+    {

+        return false;

+    }

+

+    public boolean dispatchMessageResume(MessageResumeBody body, int channelId) throws AMQException

+    {

+        return false;

+    }

+

+    public boolean dispatchMessageTransfer(MessageTransferBody body, int channelId) throws AMQException

+    {

+        return false;

+    }

+

+    public boolean dispatchBasicRecoverOk(BasicRecoverOkBody body, int channelId) throws AMQException

+    {

+        return false;  //To change body of implemented methods use File | Settings | File Templates.

+    }

+

+    public boolean dispatchQueueUnbindOk(QueueUnbindOkBody body, int channelId) throws AMQException

+    {

+        throw new UnexpectedMethodException(body);

+    }

+

+    public boolean dispatchQueueUnbind(QueueUnbindBody body, int channelId) throws AMQException

+    {

+        _queueUnbindHandler.methodReceived(getStateManager(),body,channelId);

+        return true;

+    }

+}
\ No newline at end of file
diff --git a/java/broker/src/main/java/org/apache/qpid/server/output/ProtocolOutputConverterRegistry.java b/java/broker/src/main/java/org/apache/qpid/server/output/ProtocolOutputConverterRegistry.java
index 36e7e88fd6..3a94160e22 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/output/ProtocolOutputConverterRegistry.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/output/ProtocolOutputConverterRegistry.java
@@ -44,6 +44,7 @@ public class ProtocolOutputConverterRegistry
     {

         register(ProtocolVersion.v8_0, org.apache.qpid.server.output.amqp0_8.ProtocolOutputConverterImpl.getInstanceFactory());

         register(ProtocolVersion.v0_9, org.apache.qpid.server.output.amqp0_9.ProtocolOutputConverterImpl.getInstanceFactory());

+        register(ProtocolVersion.v0_91, org.apache.qpid.server.output.amqp0_9_1.ProtocolOutputConverterImpl.getInstanceFactory());

 

     }

 

diff --git a/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_9_1/ProtocolOutputConverterImpl.java b/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_9_1/ProtocolOutputConverterImpl.java
new file mode 100644
index 0000000000..cffbe445ee
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_9_1/ProtocolOutputConverterImpl.java
@@ -0,0 +1,383 @@
+package org.apache.qpid.server.output.amqp0_9_1;

+/*

+ *

+ * 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.

+ *

+ */

+

+

+import org.apache.mina.common.ByteBuffer;

+

+import org.apache.qpid.server.output.ProtocolOutputConverter;

+import org.apache.qpid.server.output.HeaderPropertiesConverter;

+import org.apache.qpid.server.protocol.AMQProtocolSession;

+import org.apache.qpid.server.message.AMQMessage;

+import org.apache.qpid.server.queue.QueueEntry;

+import org.apache.qpid.server.message.MessageContentSource;

+import org.apache.qpid.server.message.MessageTransferMessage;

+import org.apache.qpid.framing.*;

+import org.apache.qpid.framing.amqp_0_91.BasicGetBodyImpl;

+import org.apache.qpid.framing.abstraction.MessagePublishInfo;

+import org.apache.qpid.framing.abstraction.ProtocolVersionMethodConverter;

+import org.apache.qpid.AMQException;

+import org.apache.qpid.transport.DeliveryProperties;

+import org.apache.qpid.protocol.AMQVersionAwareProtocolSession;

+

+public class ProtocolOutputConverterImpl implements ProtocolOutputConverter

+{

+    private static final MethodRegistry METHOD_REGISTRY = MethodRegistry.getMethodRegistry(ProtocolVersion.v0_91);

+    private static final ProtocolVersionMethodConverter

+            PROTOCOL_CONVERTER = METHOD_REGISTRY.getProtocolVersionMethodConverter();

+

+

+    public static Factory getInstanceFactory()

+    {

+        return new Factory()

+        {

+

+            public ProtocolOutputConverter newInstance(AMQProtocolSession session)

+            {

+                return new ProtocolOutputConverterImpl(session);

+            }

+        };

+    }

+

+    private final AMQProtocolSession _protocolSession;

+

+    private ProtocolOutputConverterImpl(AMQProtocolSession session)

+    {

+        _protocolSession = session;

+    }

+

+

+    public AMQProtocolSession getProtocolSession()

+    {

+        return _protocolSession;

+    }

+

+    public void writeDeliver(QueueEntry entry, int channelId, long deliveryTag, AMQShortString consumerTag)

+            throws AMQException

+    {

+        AMQBody deliverBody = createEncodedDeliverBody(entry, deliveryTag, consumerTag);

+        writeMessageDelivery(entry, channelId, deliverBody);

+    }

+

+

+    private ContentHeaderBody getContentHeaderBody(QueueEntry entry)

+            throws AMQException

+    {

+        if(entry.getMessage() instanceof AMQMessage)

+        {

+            return ((AMQMessage)entry.getMessage()).getContentHeaderBody();

+        }

+        else

+        {

+            final MessageTransferMessage message = (MessageTransferMessage) entry.getMessage();

+            BasicContentHeaderProperties props = HeaderPropertiesConverter.convert(message);

+            ContentHeaderBody chb = new ContentHeaderBody(props, BasicGetBodyImpl.CLASS_ID);

+            chb.bodySize = message.getSize();

+            return chb;

+        }

+    }

+

+

+    private void writeMessageDelivery(QueueEntry entry, int channelId, AMQBody deliverBody)

+            throws AMQException

+    {

+        writeMessageDelivery(entry.getMessage(), getContentHeaderBody(entry), channelId, deliverBody);

+    }

+

+    private void writeMessageDelivery(MessageContentSource message, ContentHeaderBody contentHeaderBody, int channelId, AMQBody deliverBody)

+            throws AMQException

+    {

+

+

+        int bodySize = (int) message.getSize();

+

+        if(bodySize == 0)

+        {

+            SmallCompositeAMQBodyBlock compositeBlock = new SmallCompositeAMQBodyBlock(channelId, deliverBody,

+                                                                             contentHeaderBody);

+

+            writeFrame(compositeBlock);

+        }

+        else

+        {

+            int maxBodySize = (int) getProtocolSession().getMaxFrameSize() - AMQFrame.getFrameOverhead();

+

+

+            final int capacity = bodySize > maxBodySize ? maxBodySize : bodySize;

+            java.nio.ByteBuffer buf = java.nio.ByteBuffer.allocate(capacity);

+

+            int writtenSize = 0;

+

+

+            writtenSize += message.getContent(buf, writtenSize);

+            buf.flip();

+            AMQBody firstContentBody = PROTOCOL_CONVERTER.convertToBody(buf);

+

+            CompositeAMQBodyBlock

+                    compositeBlock = new CompositeAMQBodyBlock(channelId, deliverBody, contentHeaderBody, firstContentBody);

+            writeFrame(compositeBlock);

+

+            while(writtenSize < bodySize)

+            {

+                buf = java.nio.ByteBuffer.allocate(capacity);

+

+                writtenSize += message.getContent(buf, writtenSize);

+                buf.flip();

+                writeFrame(new AMQFrame(channelId, PROTOCOL_CONVERTER.convertToBody(buf)));

+            }

+        }

+    }

+

+    private AMQDataBlock createContentHeaderBlock(final int channelId, final ContentHeaderBody contentHeaderBody)

+    {

+

+        AMQDataBlock contentHeader = ContentHeaderBody.createAMQFrame(channelId,

+                                                                      contentHeaderBody);

+        return contentHeader;

+    }

+

+

+    public void writeGetOk(QueueEntry entry, int channelId, long deliveryTag, int queueSize) throws AMQException

+    {

+        AMQBody deliver = createEncodedGetOkBody(entry, deliveryTag, queueSize);

+        writeMessageDelivery(entry, channelId, deliver);

+    }

+

+

+    private AMQBody createEncodedDeliverBody(QueueEntry entry,

+                                              final long deliveryTag,

+                                              final AMQShortString consumerTag)

+            throws AMQException

+    {

+

+        final AMQShortString exchangeName;

+        final AMQShortString routingKey;

+

+        if(entry.getMessage() instanceof AMQMessage)

+        {

+            final AMQMessage message = (AMQMessage) entry.getMessage();

+            final MessagePublishInfo pb = message.getMessagePublishInfo();

+            exchangeName = pb.getExchange();

+            routingKey = pb.getRoutingKey();

+        }

+        else

+        {

+            MessageTransferMessage message = (MessageTransferMessage) entry.getMessage();

+            DeliveryProperties delvProps = message.getHeader().get(DeliveryProperties.class);

+            exchangeName = (delvProps == null || delvProps.getExchange() == null) ? null : new AMQShortString(delvProps.getExchange());

+            routingKey = (delvProps == null || delvProps.getRoutingKey() == null) ? null : new AMQShortString(delvProps.getRoutingKey());

+        }

+

+        final boolean isRedelivered = entry.isRedelivered();

+

+        final AMQBody returnBlock = new AMQBody()

+        {

+

+            public AMQBody _underlyingBody;

+

+            public AMQBody createAMQBody()

+            {

+                return METHOD_REGISTRY.createBasicDeliverBody(consumerTag,

+                                                              deliveryTag,

+                                                              isRedelivered,

+                                                              exchangeName,

+                                                              routingKey);

+

+

+

+

+

+            }

+

+            public byte getFrameType()

+            {

+                return AMQMethodBody.TYPE;

+            }

+

+            public int getSize()

+            {

+                if(_underlyingBody == null)

+                {

+                    _underlyingBody = createAMQBody();

+                }

+                return _underlyingBody.getSize();

+            }

+

+            public void writePayload(ByteBuffer buffer)

+            {

+                if(_underlyingBody == null)

+                {

+                    _underlyingBody = createAMQBody();

+                }

+                _underlyingBody.writePayload(buffer);

+            }

+

+            public void handle(final int channelId, final AMQVersionAwareProtocolSession amqMinaProtocolSession)

+                throws AMQException

+            {

+                throw new AMQException("This block should never be dispatched!");

+            }

+        };

+        return returnBlock;

+    }

+

+    private AMQBody createEncodedGetOkBody(QueueEntry entry, long deliveryTag, int queueSize)

+            throws AMQException

+    {

+        final AMQShortString exchangeName;

+        final AMQShortString routingKey;

+

+        if(entry.getMessage() instanceof AMQMessage)

+        {

+            final AMQMessage message = (AMQMessage) entry.getMessage();

+            final MessagePublishInfo pb = message.getMessagePublishInfo();

+            exchangeName = pb.getExchange();

+            routingKey = pb.getRoutingKey();

+        }

+        else

+        {

+            MessageTransferMessage message = (MessageTransferMessage) entry.getMessage();

+            DeliveryProperties delvProps = message.getHeader().get(DeliveryProperties.class);

+            exchangeName = (delvProps == null || delvProps.getExchange() == null) ? null : new AMQShortString(delvProps.getExchange());

+            routingKey = (delvProps == null || delvProps.getRoutingKey() == null) ? null : new AMQShortString(delvProps.getRoutingKey());

+        }

+

+        final boolean isRedelivered = entry.isRedelivered();

+

+        BasicGetOkBody getOkBody =

+                METHOD_REGISTRY.createBasicGetOkBody(deliveryTag,

+                                                    isRedelivered,

+                                                    exchangeName,

+                                                    routingKey,

+                                                    queueSize);

+

+        return getOkBody;

+    }

+

+    public byte getProtocolMinorVersion()

+    {

+        return getProtocolSession().getProtocolMinorVersion();

+    }

+

+    public byte getProtocolMajorVersion()

+    {

+        return getProtocolSession().getProtocolMajorVersion();

+    }

+

+    private AMQBody createEncodedReturnFrame(MessagePublishInfo messagePublishInfo,

+                                             int replyCode,

+                                             AMQShortString replyText) throws AMQException

+    {

+

+        BasicReturnBody basicReturnBody =

+                METHOD_REGISTRY.createBasicReturnBody(replyCode,

+                                                     replyText,

+                                                     messagePublishInfo.getExchange(),

+                                                     messagePublishInfo.getRoutingKey());

+

+

+        return basicReturnBody;

+    }

+

+    public void writeReturn(MessagePublishInfo messagePublishInfo, ContentHeaderBody header, MessageContentSource message, int channelId, int replyCode, AMQShortString replyText)

+            throws AMQException

+    {

+

+        AMQBody returnFrame = createEncodedReturnFrame(messagePublishInfo, replyCode, replyText);

+

+        writeMessageDelivery(message, header, channelId, returnFrame);

+    }

+

+

+    public void writeFrame(AMQDataBlock block)

+    {

+        getProtocolSession().writeFrame(block);

+    }

+

+

+    public void confirmConsumerAutoClose(int channelId, AMQShortString consumerTag)

+    {

+

+        BasicCancelOkBody basicCancelOkBody = METHOD_REGISTRY.createBasicCancelOkBody(consumerTag);

+        writeFrame(basicCancelOkBody.generateFrame(channelId));

+

+    }

+

+

+    public static final class CompositeAMQBodyBlock extends AMQDataBlock

+    {

+        public static final int OVERHEAD = 3 * AMQFrame.getFrameOverhead();

+

+        private final AMQBody _methodBody;

+        private final AMQBody _headerBody;

+        private final AMQBody _contentBody;

+        private final int _channel;

+

+

+        public CompositeAMQBodyBlock(int channel, AMQBody methodBody, AMQBody headerBody, AMQBody contentBody)

+        {

+            _channel = channel;

+            _methodBody = methodBody;

+            _headerBody = headerBody;

+            _contentBody = contentBody;

+

+        }

+

+        public long getSize()

+        {

+            return OVERHEAD + _methodBody.getSize() + _headerBody.getSize() + _contentBody.getSize();

+        }

+

+        public void writePayload(ByteBuffer buffer)

+        {

+            AMQFrame.writeFrames(buffer, _channel, _methodBody, _headerBody, _contentBody);

+        }

+    }

+

+    public static final class SmallCompositeAMQBodyBlock extends AMQDataBlock

+    {

+        public static final int OVERHEAD = 2 * AMQFrame.getFrameOverhead();

+

+        private final AMQBody _methodBody;

+        private final AMQBody _headerBody;

+        private final int _channel;

+

+

+        public SmallCompositeAMQBodyBlock(int channel, AMQBody methodBody, AMQBody headerBody)

+        {

+            _channel = channel;

+            _methodBody = methodBody;

+            _headerBody = headerBody;

+

+        }

+

+        public long getSize()

+        {

+            return OVERHEAD + _methodBody.getSize() + _headerBody.getSize() ;

+        }

+

+        public void writePayload(ByteBuffer buffer)

+        {

+            AMQFrame.writeFrames(buffer, _channel, _methodBody, _headerBody);

+        }

+    }

+

+}
\ No newline at end of file