qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From oru...@apache.org
Subject [1/3] qpid-broker-j git commit: [Broker-J][System Tests] Add protocol system test suites for AMQP 0-8, 0-9 and 0-9-1
Date Sat, 18 Nov 2017 01:42:19 GMT
Repository: qpid-broker-j
Updated Branches:
  refs/heads/master a3c00bbfc -> c6d80d80e


[Broker-J][System Tests] Add protocol system test suites for AMQP 0-8,0-9 and 0-9-1


Project: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/commit/c6d80d80
Tree: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/tree/c6d80d80
Diff: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/diff/c6d80d80

Branch: refs/heads/master
Commit: c6d80d80e685f70653f609951868a9e8f0ffb5a6
Parents: 06e53d7
Author: Alex Rudyy <orudyy@apache.org>
Authored: Sat Nov 18 01:32:22 2017 +0000
Committer: Alex Rudyy <orudyy@apache.org>
Committed: Sat Nov 18 01:38:53 2017 +0000

----------------------------------------------------------------------
 .../qpid/server/protocol/v0_8/AMQDecoder.java   |   4 +-
 .../server/protocol/v0_8/ServerDecoder.java     |   2 +-
 .../server/protocol/v0_8/ClientDecoder.java     |   2 +-
 pom.xml                                         |   7 +
 systests/protocol-tests-amqp-0-8/pom.xml        | 109 +++++++
 .../tests/protocol/v0_8/BasicInteraction.java   | 214 ++++++++++++
 .../tests/protocol/v0_8/ChannelInteraction.java |  51 +++
 .../qpid/tests/protocol/v0_8/ClientDecoder.java | 323 +++++++++++++++++++
 .../protocol/v0_8/ConnectionInteraction.java    | 103 ++++++
 .../qpid/tests/protocol/v0_8/FrameDecoder.java  | 111 +++++++
 .../qpid/tests/protocol/v0_8/FrameEncoder.java  |  83 +++++
 .../tests/protocol/v0_8/FrameTransport.java     | 101 ++++++
 .../qpid/tests/protocol/v0_8/Interaction.java   | 109 +++++++
 .../protocol/v0_8/PerformativeResponse.java     |  54 ++++
 .../tests/protocol/v0_8/QueueInteraction.java   |  63 ++++
 .../resources/config-protocol-tests-0-8.json    |  78 +++++
 .../qpid/tests/protocol/v0_8/BasicTest.java     | 156 +++++++++
 .../qpid/tests/protocol/v0_8/ChannelTest.java   |  54 ++++
 .../tests/protocol/v0_8/ConnectionTest.java     | 183 +++++++++++
 .../qpid/tests/protocol/v0_8/QueueTest.java     |  67 ++++
 20 files changed, 1870 insertions(+), 4 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/c6d80d80/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQDecoder.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQDecoder.java b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQDecoder.java
index 706f1ae..add9fb1 100644
--- a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQDecoder.java
+++ b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQDecoder.java
@@ -217,11 +217,11 @@ public abstract class AMQDecoder<T extends MethodProcessor>
     }
 
 
-    abstract void processMethod(int channelId,
+    protected abstract void processMethod(int channelId,
                                QpidByteBuffer in)
             throws AMQFrameDecodingException;
 
-    AMQFrameDecodingException newUnknownMethodException(final int classId,
+    protected AMQFrameDecodingException newUnknownMethodException(final int classId,
                                                         final int methodId,
                                                         ProtocolVersion protocolVersion)
     {

http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/c6d80d80/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ServerDecoder.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ServerDecoder.java b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ServerDecoder.java
index 08ccda9..59d4985 100644
--- a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ServerDecoder.java
+++ b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ServerDecoder.java
@@ -46,7 +46,7 @@ public class ServerDecoder extends AMQDecoder<ServerMethodProcessor<? extends Se
 
 
     @Override
-    void processMethod(int channelId,
+    protected void processMethod(int channelId,
                        QpidByteBuffer in)
             throws AMQFrameDecodingException
     {

http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/c6d80d80/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/ClientDecoder.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/ClientDecoder.java b/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/ClientDecoder.java
index de8afb9..8e02bbb 100644
--- a/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/ClientDecoder.java
+++ b/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/ClientDecoder.java
@@ -86,7 +86,7 @@ public class ClientDecoder extends AMQDecoder<ClientMethodProcessor<? extends Cl
     }
 
     @Override
-    void processMethod(int channelId,
+    protected void processMethod(int channelId,
                        QpidByteBuffer in)
             throws AMQFrameDecodingException
     {

http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/c6d80d80/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 62e7033..e7d58bb 100644
--- a/pom.xml
+++ b/pom.xml
@@ -197,6 +197,7 @@
     <module>systests/systests-utils</module>
     <module>systests/qpid-systests-jms_2.0</module>
     <module>systests/protocol-tests-core</module>
+    <module>systests/protocol-tests-amqp-0-8</module>
     <module>systests/protocol-tests-amqp-1-0</module>
     <module>systests/end-to-end-conversion-tests</module>
     <module>perftests</module>
@@ -420,6 +421,12 @@
         <version>${project.version}</version>
       </dependency>
 
+      <dependency>
+        <groupId>org.apache.qpid</groupId>
+        <artifactId>protocol-tests-amqp-0-8</artifactId>
+        <version>${project.version}</version>
+      </dependency>
+
       <!-- External dependencies -->
       <dependency>
         <groupId>org.apache.qpid</groupId>

http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/c6d80d80/systests/protocol-tests-amqp-0-8/pom.xml
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-0-8/pom.xml b/systests/protocol-tests-amqp-0-8/pom.xml
new file mode 100644
index 0000000..3f788db
--- /dev/null
+++ b/systests/protocol-tests-amqp-0-8/pom.xml
@@ -0,0 +1,109 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  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 xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <groupId>org.apache.qpid</groupId>
+        <artifactId>qpid-systests-parent</artifactId>
+        <version>7.1.0-SNAPSHOT</version>
+        <relativePath>../../qpid-systests-parent/pom.xml</relativePath>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>protocol-tests-amqp-0-8</artifactId>
+    <name>Apache Qpid Protocol Tests for AMQP 0-8,0-9,0-9-1</name>
+    <description>Tests for AMQP 0-8,0-9,0-9-1</description>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.qpid</groupId>
+            <artifactId>qpid-broker-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.qpid</groupId>
+            <artifactId>qpid-broker-plugins-amqp-0-8-protocol</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.qpid</groupId>
+            <artifactId>qpid-test-utils</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.qpid</groupId>
+            <artifactId>protocol-tests-core</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.qpid</groupId>
+            <artifactId>qpid-systests-utils</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.qpid</groupId>
+            <artifactId>qpid-broker-plugins-logging-logback</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.qpid</groupId>
+            <artifactId>qpid-broker-plugins-memory-store</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.qpid</groupId>
+            <artifactId>qpid-broker-plugins-derby-store</artifactId>
+            <optional>true</optional>
+            <scope>test</scope>
+        </dependency>
+
+       <dependency>
+            <groupId>org.apache.qpid</groupId>
+            <artifactId>qpid-bdbstore</artifactId>
+            <scope>test</scope>
+       </dependency>
+
+        <dependency>
+            <groupId>org.hamcrest</groupId>
+            <artifactId>hamcrest-library</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.hamcrest</groupId>
+            <artifactId>hamcrest-integration</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <configuration>
+                    <systemPropertyVariables>
+                        <qpid.initialConfigurationLocation>classpath:config-protocol-tests-0-8.json</qpid.initialConfigurationLocation>
+                    </systemPropertyVariables>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>

http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/c6d80d80/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/BasicInteraction.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/BasicInteraction.java b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/BasicInteraction.java
new file mode 100644
index 0000000..9ef66ec
--- /dev/null
+++ b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/BasicInteraction.java
@@ -0,0 +1,214 @@
+/*
+ *
+ * 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.tests.protocol.v0_8;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.qpid.server.protocol.v0_8.AMQShortString;
+import org.apache.qpid.server.protocol.v0_8.FieldTable;
+import org.apache.qpid.server.protocol.v0_8.transport.AMQFrame;
+import org.apache.qpid.server.protocol.v0_8.transport.BasicAckBody;
+import org.apache.qpid.server.protocol.v0_8.transport.BasicConsumeBody;
+import org.apache.qpid.server.protocol.v0_8.transport.BasicContentHeaderProperties;
+import org.apache.qpid.server.protocol.v0_8.transport.BasicPublishBody;
+import org.apache.qpid.server.protocol.v0_8.transport.BasicQosBody;
+import org.apache.qpid.server.protocol.v0_8.transport.CompositeAMQDataBlock;
+import org.apache.qpid.server.protocol.v0_8.transport.ContentBody;
+import org.apache.qpid.server.protocol.v0_8.transport.ContentHeaderBody;
+
+public class BasicInteraction
+{
+    private final Interaction _interaction;
+    private String _publishExchange;
+    private String _publishRoutingKey;
+    private boolean _publishMandatory;
+    private boolean _publishImmediate;
+    private byte[] _content;
+    private Map<String, Object> _contentHeaderPropertiesHeaders = new HashMap<>();
+    private String _contentHeaderPropertiesContentType;
+    private byte _contentHeaderPropertiesDeliveryMode;
+    private byte _contentHeaderPropertiesPriority;
+    private int _qosPrefetchCount;
+    private long _qosPrefetchSize;
+    private boolean _qosGlobal;
+    private String _consumeQueueName;
+    private String _consumeConsumerTag;
+    private boolean _consumeNoLocal;
+    private boolean _consumeNoAck;
+    private boolean _consumeExclusive;
+    private boolean _consumeNoWait;
+    private Map<String, Object> _consumeArguments = new HashMap<>();
+    private long _ackDeliveryTag;
+    private boolean _ackMultiple;
+
+    public BasicInteraction(final Interaction interaction)
+    {
+        _interaction = interaction;
+    }
+
+    public Interaction publish() throws Exception
+    {
+        return _interaction.sendPerformative(new BasicPublishBody(0,
+                                                                  AMQShortString.valueOf(_publishExchange),
+                                                                  AMQShortString.valueOf(_publishRoutingKey),
+                                                                  _publishMandatory,
+                                                                  _publishImmediate));
+    }
+
+    public BasicInteraction content(final String content)
+    {
+        _content = content.getBytes(StandardCharsets.UTF_8);
+        return this;
+    }
+
+    public BasicInteraction content(final byte[] content)
+    {
+        _content = content;
+        return this;
+    }
+
+    public BasicInteraction contentHeaderPropertiesHeaders(final Map<String, Object> messageHeaders)
+    {
+        _contentHeaderPropertiesHeaders = messageHeaders;
+        return this;
+    }
+
+    public BasicInteraction contentHeaderPropertiesContentType(final String messageContentType)
+    {
+        _contentHeaderPropertiesContentType = messageContentType;
+        return this;
+    }
+
+    public BasicInteraction contentHeaderPropertiesPriority(final byte priority)
+    {
+        _contentHeaderPropertiesPriority = priority;
+        return this;
+    }
+
+    public BasicInteraction contentHeaderPropertiesDeliveryMode(final byte deliveryMode)
+    {
+        _contentHeaderPropertiesDeliveryMode = deliveryMode;
+        return this;
+    }
+
+
+    public Interaction publishMessage() throws Exception
+    {
+        List<AMQFrame> frames = new ArrayList<>();
+        BasicPublishBody publishFrame = new BasicPublishBody(0,
+                                                             AMQShortString.valueOf(_publishExchange),
+                                                             AMQShortString.valueOf(_publishRoutingKey),
+                                                             _publishMandatory,
+                                                             _publishImmediate);
+        frames.add(new AMQFrame(_interaction.getChannelId(), publishFrame));
+        final BasicContentHeaderProperties basicContentHeaderProperties = new BasicContentHeaderProperties();
+        basicContentHeaderProperties.setHeaders(FieldTable.convertToFieldTable(_contentHeaderPropertiesHeaders));
+        basicContentHeaderProperties.setContentType(_contentHeaderPropertiesContentType);
+        basicContentHeaderProperties.setDeliveryMode(_contentHeaderPropertiesDeliveryMode);
+        basicContentHeaderProperties.setPriority(_contentHeaderPropertiesPriority);
+        final int contentSize = _content == null ? 0 : _content.length;
+        ContentHeaderBody contentHeaderBody = new ContentHeaderBody(basicContentHeaderProperties, contentSize);
+        frames.add(new AMQFrame(_interaction.getChannelId(), contentHeaderBody));
+        if (contentSize > 0)
+        {
+            final int framePayloadMax = _interaction.getMaximumFrameSize() - 8;
+            int offset = 0;
+            do
+            {
+                int contentToCopyLength = Math.min(framePayloadMax, contentSize - offset);
+                ContentBody contentBody = new ContentBody(ByteBuffer.wrap(_content, offset,
+                                                                          contentToCopyLength));
+                frames.add(new AMQFrame(_interaction.getChannelId(), contentBody));
+                offset += contentToCopyLength;
+            }
+            while (offset < contentSize);
+        }
+
+        CompositeAMQDataBlock frame = new CompositeAMQDataBlock(frames.toArray(new AMQFrame[frames.size()]));
+
+        return _interaction.sendPerformative(frame);
+    }
+
+    public BasicInteraction publishExchange(final String exchangeName)
+    {
+        _publishExchange = exchangeName;
+        return this;
+    }
+
+    public BasicInteraction publishRoutingKey(final String queueName)
+    {
+        _publishRoutingKey = queueName;
+        return this;
+    }
+
+    public BasicInteraction qosPrefetchCount(final int prefetchCount)
+    {
+        _qosPrefetchCount = prefetchCount;
+        return this;
+    }
+
+    public Interaction qos() throws Exception
+    {
+        return _interaction.sendPerformative(new BasicQosBody(_qosPrefetchSize,
+                                                              _qosPrefetchCount,
+                                                              _qosGlobal));
+    }
+
+    public Interaction consume() throws Exception
+    {
+        return _interaction.sendPerformative(new BasicConsumeBody(0,
+                                                                  AMQShortString.valueOf(_consumeQueueName),
+                                                                  AMQShortString.valueOf(_consumeConsumerTag),
+                                                                  _consumeNoLocal,
+                                                                  _consumeNoAck,
+                                                                  _consumeExclusive,
+                                                                  _consumeNoWait,
+                                                                  FieldTable.convertToFieldTable(_consumeArguments)));
+    }
+
+    public BasicInteraction consumeConsumerTag(final String consumerTag)
+    {
+        _consumeConsumerTag = consumerTag;
+        return this;
+    }
+
+    public BasicInteraction consumeQueue(final String queueName)
+    {
+        _consumeQueueName = queueName;
+        return this;
+    }
+
+    public Interaction ack() throws Exception
+    {
+        return _interaction.sendPerformative(new BasicAckBody(_ackDeliveryTag, _ackMultiple));
+    }
+
+    public BasicInteraction ackDeliveryTag(final long deliveryTag)
+    {
+        _ackDeliveryTag = deliveryTag;
+        return this;
+    }
+}

http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/c6d80d80/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/ChannelInteraction.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/ChannelInteraction.java b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/ChannelInteraction.java
new file mode 100644
index 0000000..51d4426
--- /dev/null
+++ b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/ChannelInteraction.java
@@ -0,0 +1,51 @@
+/*
+ *
+ * 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.tests.protocol.v0_8;
+
+import org.apache.qpid.server.protocol.v0_8.AMQShortString;
+import org.apache.qpid.server.protocol.v0_8.transport.ChannelCloseBody;
+import org.apache.qpid.server.protocol.v0_8.transport.ChannelFlowBody;
+import org.apache.qpid.server.protocol.v0_8.transport.ChannelOpenBody;
+
+public class ChannelInteraction
+{
+    private Interaction _interaction;
+
+    public ChannelInteraction(final Interaction interaction)
+    {
+        _interaction = interaction;
+    }
+
+    public Interaction open() throws Exception
+    {
+        return _interaction.sendPerformative(new ChannelOpenBody());
+    }
+
+    public Interaction close() throws Exception
+    {
+        return _interaction.sendPerformative(new ChannelCloseBody(200, AMQShortString.valueOf(""), 0, 0));
+    }
+
+    public Interaction flow(final boolean active) throws Exception
+    {
+        return _interaction.sendPerformative(new ChannelFlowBody(active));
+    }
+}

http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/c6d80d80/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/ClientDecoder.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/ClientDecoder.java b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/ClientDecoder.java
new file mode 100644
index 0000000..2bd9d27
--- /dev/null
+++ b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/ClientDecoder.java
@@ -0,0 +1,323 @@
+/*
+ *
+ * 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.tests.protocol.v0_8;
+
+import java.nio.ByteBuffer;
+
+import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.protocol.ProtocolVersion;
+import org.apache.qpid.server.protocol.v0_8.AMQDecoder;
+import org.apache.qpid.server.protocol.v0_8.AMQFrameDecodingException;
+import org.apache.qpid.server.protocol.v0_8.transport.*;
+
+public class ClientDecoder extends AMQDecoder<ClientMethodProcessor<? extends ClientChannelMethodProcessor>>
+{
+    private QpidByteBuffer _incompleteBuffer;
+
+    /**
+     * Creates a new AMQP decoder.
+     *
+     * @param methodProcessor          method processor
+     */
+    public ClientDecoder(final ClientMethodProcessor<? extends ClientChannelMethodProcessor> methodProcessor)
+    {
+        super(false, methodProcessor);
+    }
+
+    public void decodeBuffer(ByteBuffer incomingBuffer) throws AMQFrameDecodingException, AMQProtocolVersionException
+    {
+        if (_incompleteBuffer == null)
+        {
+            QpidByteBuffer qpidByteBuffer = QpidByteBuffer.wrap(incomingBuffer);
+            final int required = decode(qpidByteBuffer);
+            if (required != 0)
+            {
+                _incompleteBuffer = QpidByteBuffer.allocate(qpidByteBuffer.remaining() + required);
+                _incompleteBuffer.put(qpidByteBuffer);
+            }
+            qpidByteBuffer.dispose();
+        }
+        else
+        {
+            if (incomingBuffer.remaining() < _incompleteBuffer.remaining())
+            {
+                _incompleteBuffer.put(incomingBuffer);
+            }
+            else
+            {
+                _incompleteBuffer.flip();
+                final QpidByteBuffer aggregatedBuffer =
+                        QpidByteBuffer.allocate(_incompleteBuffer.remaining() + incomingBuffer.remaining());
+                aggregatedBuffer.put(_incompleteBuffer);
+                aggregatedBuffer.put(incomingBuffer);
+                aggregatedBuffer.flip();
+                final int required = decode(aggregatedBuffer);
+
+                _incompleteBuffer.dispose();
+                if (required != 0)
+                {
+                    _incompleteBuffer = QpidByteBuffer.allocate(aggregatedBuffer.remaining() + required);
+                    _incompleteBuffer.put(aggregatedBuffer);
+                }
+                else
+                {
+                    _incompleteBuffer = null;
+                }
+                aggregatedBuffer.dispose();
+            }
+        }
+        // post-condition: assert(!incomingBuffer.hasRemaining());
+    }
+
+    @Override
+    protected void processMethod(int channelId,
+                       QpidByteBuffer in)
+            throws AMQFrameDecodingException
+    {
+        ClientMethodProcessor<? extends ClientChannelMethodProcessor> methodProcessor = getMethodProcessor();
+        ClientChannelMethodProcessor channelMethodProcessor = methodProcessor.getChannelMethodProcessor(channelId);
+        final int classAndMethod = in.getInt();
+        int classId = classAndMethod >> 16;
+        int methodId = classAndMethod & 0xFFFF;
+        methodProcessor.setCurrentMethod(classId, methodId);
+        try
+        {
+            switch (classAndMethod)
+            {
+                //CONNECTION_CLASS:
+                case 0x000a000a:
+                    ConnectionStartBody.process(in, methodProcessor);
+                    break;
+                case 0x000a0014:
+                    ConnectionSecureBody.process(in, methodProcessor);
+                    break;
+                case 0x000a001e:
+                    ConnectionTuneBody.process(in, methodProcessor);
+                    break;
+                case 0x000a0029:
+                    ConnectionOpenOkBody.process(in, methodProcessor);
+                    break;
+                case 0x000a002a:
+                    ConnectionRedirectBody.process(in, methodProcessor);
+                    break;
+                case 0x000a0032:
+                    if (methodProcessor.getProtocolVersion().equals(ProtocolVersion.v0_8))
+                    {
+                        ConnectionRedirectBody.process(in, methodProcessor);
+                    }
+                    else
+                    {
+                        ConnectionCloseBody.process(in, methodProcessor);
+                    }
+                    break;
+                case 0x000a0033:
+                    if (methodProcessor.getProtocolVersion().equals(ProtocolVersion.v0_8))
+                    {
+                        throw newUnknownMethodException(classId, methodId,
+                                                        methodProcessor.getProtocolVersion());
+                    }
+                    else
+                    {
+                        methodProcessor.receiveConnectionCloseOk();
+                    }
+                    break;
+                case 0x000a003c:
+                    if (methodProcessor.getProtocolVersion().equals(ProtocolVersion.v0_8))
+                    {
+                        ConnectionCloseBody.process(in, methodProcessor);
+                    }
+                    else
+                    {
+                        throw newUnknownMethodException(classId, methodId,
+                                                        methodProcessor.getProtocolVersion());
+                    }
+                    break;
+                case 0x000a003d:
+                    if (methodProcessor.getProtocolVersion().equals(ProtocolVersion.v0_8))
+                    {
+                        methodProcessor.receiveConnectionCloseOk();
+                    }
+                    else
+                    {
+                        throw newUnknownMethodException(classId, methodId,
+                                                        methodProcessor.getProtocolVersion());
+                    }
+                    break;
+
+                // CHANNEL_CLASS:
+
+                case 0x0014000b:
+                    ChannelOpenOkBody.process(in, methodProcessor.getProtocolVersion(), channelMethodProcessor);
+                    break;
+                case 0x00140014:
+                    ChannelFlowBody.process(in, channelMethodProcessor);
+                    break;
+                case 0x00140015:
+                    ChannelFlowOkBody.process(in, channelMethodProcessor);
+                    break;
+                case 0x0014001e:
+                    ChannelAlertBody.process(in, channelMethodProcessor);
+                    break;
+                case 0x00140028:
+                    ChannelCloseBody.process(in, channelMethodProcessor);
+                    break;
+                case 0x00140029:
+                    channelMethodProcessor.receiveChannelCloseOk();
+                    break;
+
+                // ACCESS_CLASS:
+
+                case 0x001e000b:
+                    AccessRequestOkBody.process(in, channelMethodProcessor);
+                    break;
+
+                // EXCHANGE_CLASS:
+
+                case 0x0028000b:
+                    if(!channelMethodProcessor.ignoreAllButCloseOk())
+                    {
+                        channelMethodProcessor.receiveExchangeDeclareOk();
+                    }
+                    break;
+                case 0x00280015:
+                    if(!channelMethodProcessor.ignoreAllButCloseOk())
+                    {
+                        channelMethodProcessor.receiveExchangeDeleteOk();
+                    }
+                    break;
+                case 0x00280017:
+                    ExchangeBoundOkBody.process(in, channelMethodProcessor);
+                    break;
+
+
+                // QUEUE_CLASS:
+
+                case 0x0032000b:
+                    QueueDeclareOkBody.process(in, channelMethodProcessor);
+                    break;
+                case 0x00320015:
+                    if(!channelMethodProcessor.ignoreAllButCloseOk())
+                    {
+                        channelMethodProcessor.receiveQueueBindOk();
+                    }
+                    break;
+                case 0x0032001f:
+                    QueuePurgeOkBody.process(in, channelMethodProcessor);
+                    break;
+                case 0x00320029:
+                    QueueDeleteOkBody.process(in, channelMethodProcessor);
+                    break;
+                case 0x00320033:
+                    if(!channelMethodProcessor.ignoreAllButCloseOk())
+                    {
+                        channelMethodProcessor.receiveQueueUnbindOk();
+                    }
+                    break;
+
+
+                // BASIC_CLASS:
+
+                case 0x003c000b:
+                    if(!channelMethodProcessor.ignoreAllButCloseOk())
+                    {
+                        channelMethodProcessor.receiveBasicQosOk();
+                    }
+                    break;
+                case 0x003c0015:
+                    BasicConsumeOkBody.process(in, channelMethodProcessor);
+                    break;
+                case 0x003c001f:
+                    BasicCancelOkBody.process(in, channelMethodProcessor);
+                    break;
+                case 0x003c0032:
+                    BasicReturnBody.process(in, channelMethodProcessor);
+                    break;
+                case 0x003c003c:
+                    BasicDeliverBody.process(in, channelMethodProcessor);
+                    break;
+                case 0x003c0047:
+                    BasicGetOkBody.process(in, channelMethodProcessor);
+                    break;
+                case 0x003c0048:
+                    BasicGetEmptyBody.process(in, channelMethodProcessor);
+                    break;
+                case 0x003c0050:
+                    BasicAckBody.process(in, channelMethodProcessor);
+                    break;
+                case 0x003c0065:
+                    if(!channelMethodProcessor.ignoreAllButCloseOk())
+                    {
+                        channelMethodProcessor.receiveBasicRecoverSyncOk();
+                    }
+                    break;
+                case 0x003c006f:
+                    if(!channelMethodProcessor.ignoreAllButCloseOk())
+                    {
+                        channelMethodProcessor.receiveBasicRecoverSyncOk();
+                    }
+                    break;
+                case 0x003c0078:
+                    BasicNackBody.process(in, channelMethodProcessor);
+                    break;
+
+                // CONFIRM CLASS:
+
+                case 0x0055000b:
+                    if(!channelMethodProcessor.ignoreAllButCloseOk())
+                    {
+                        channelMethodProcessor.receiveConfirmSelectOk();
+                    }
+                    break;
+
+                // TX_CLASS:
+
+                case 0x005a000b:
+                    if(!channelMethodProcessor.ignoreAllButCloseOk())
+                    {
+                        channelMethodProcessor.receiveTxSelectOk();
+                    }
+                    break;
+                case 0x005a0015:
+                    if(!channelMethodProcessor.ignoreAllButCloseOk())
+                    {
+                        channelMethodProcessor.receiveTxCommitOk();
+                    }
+                    break;
+                case 0x005a001f:
+                    if(!channelMethodProcessor.ignoreAllButCloseOk())
+                    {
+                        channelMethodProcessor.receiveTxRollbackOk();
+                    }
+                    break;
+
+                default:
+                    throw newUnknownMethodException(classId, methodId,
+                                                    methodProcessor.getProtocolVersion());
+
+            }
+        }
+        finally
+        {
+            methodProcessor.setCurrentMethod(0, 0);
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/c6d80d80/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/ConnectionInteraction.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/ConnectionInteraction.java b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/ConnectionInteraction.java
new file mode 100644
index 0000000..023e7fc
--- /dev/null
+++ b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/ConnectionInteraction.java
@@ -0,0 +1,103 @@
+/*
+ *
+ * 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.tests.protocol.v0_8;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.qpid.server.protocol.v0_8.AMQShortString;
+import org.apache.qpid.server.protocol.v0_8.FieldTable;
+import org.apache.qpid.server.protocol.v0_8.transport.ConnectionOpenBody;
+import org.apache.qpid.server.protocol.v0_8.transport.ConnectionStartOkBody;
+import org.apache.qpid.server.protocol.v0_8.transport.ConnectionTuneOkBody;
+
+public class ConnectionInteraction
+{
+    private final Interaction _interaction;
+
+    private final Map<String, Object> _startOkClientProperties = new HashMap<>();
+    private String _startOkMechanism;
+    private byte[] _startOkResponse;
+    private String _startOkLocale;
+    private int _tuneOkChannelMax;
+    private long _tuneOkFrameMax;
+    private int _tuneOkHeartbeat;
+    private String _openVirtualHost;
+
+    public ConnectionInteraction(final Interaction interaction)
+    {
+        _interaction = interaction;
+    }
+
+
+    public ConnectionInteraction startOkMechanism(final String startOkMechanism)
+    {
+        _startOkMechanism = startOkMechanism;
+        return this;
+    }
+
+
+    public Interaction startOk() throws Exception
+    {
+        return _interaction.sendPerformative(new ConnectionStartOkBody(FieldTable.convertToFieldTable(_startOkClientProperties),
+                                                                       AMQShortString.valueOf(_startOkMechanism),
+                                                                       _startOkResponse,
+                                                                       AMQShortString.valueOf(_startOkLocale)));
+    }
+
+    public ConnectionInteraction tuneOkChannelMax(final int channelMax)
+    {
+        _tuneOkChannelMax = channelMax;
+        return this;
+    }
+
+    public ConnectionInteraction tuneOkFrameMax(final long frameMax)
+    {
+        _tuneOkFrameMax = frameMax;
+        return this;
+    }
+
+    public ConnectionInteraction tuneOkHeartbeat(final int heartbeat)
+    {
+        _tuneOkHeartbeat = heartbeat;
+        return this;
+    }
+
+    public Interaction tuneOk() throws Exception
+    {
+        return _interaction.sendPerformative(new ConnectionTuneOkBody(_tuneOkChannelMax,
+                                                                      _tuneOkFrameMax,
+                                                                      _tuneOkHeartbeat));
+    }
+
+    public ConnectionInteraction openVirtualHost(String virtualHost)
+    {
+        _openVirtualHost = virtualHost;
+        return this;
+    }
+
+    public Interaction open() throws Exception
+    {
+        return _interaction.sendPerformative(new ConnectionOpenBody(AMQShortString.valueOf(_openVirtualHost),
+                                                                    null,
+                                                                    false));
+    }
+}

http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/c6d80d80/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/FrameDecoder.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/FrameDecoder.java b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/FrameDecoder.java
new file mode 100644
index 0000000..499fe72
--- /dev/null
+++ b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/FrameDecoder.java
@@ -0,0 +1,111 @@
+/*
+ *
+ * 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.tests.protocol.v0_8;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.protocol.ProtocolVersion;
+import org.apache.qpid.server.protocol.v0_8.transport.AMQDataBlock;
+import org.apache.qpid.server.protocol.v0_8.transport.AMQFrame;
+import org.apache.qpid.server.protocol.v0_8.transport.FrameCreatingMethodProcessor;
+import org.apache.qpid.server.protocol.v0_8.transport.ProtocolInitiation;
+import org.apache.qpid.server.transport.ByteBufferSender;
+import org.apache.qpid.tests.protocol.HeaderResponse;
+import org.apache.qpid.tests.protocol.InputDecoder;
+import org.apache.qpid.tests.protocol.Response;
+
+public class FrameDecoder implements InputDecoder
+{
+    private final ClientDecoder _clientDecoder;
+    private final FrameCreatingMethodProcessor _methodProcessor;
+
+    FrameDecoder(ProtocolVersion protocolVersion)
+    {
+        _methodProcessor = new FrameCreatingMethodProcessor(protocolVersion);
+        _clientDecoder = new ClientDecoder(_methodProcessor);
+    }
+
+    @Override
+    public Collection<Response<?>> decode(final ByteBuffer inputBuffer) throws Exception
+    {
+        _clientDecoder.decodeBuffer(inputBuffer);
+
+        List<AMQDataBlock> receivedFrames = new ArrayList<>(_methodProcessor.getProcessedMethods());
+        List<Response<?>> result = new ArrayList<>();
+
+        for (AMQDataBlock frame : receivedFrames)
+        {
+            if (frame instanceof AMQFrame)
+            {
+                AMQFrame amqFrame = (AMQFrame) frame;
+                result.add(new PerformativeResponse(amqFrame.getChannel(), amqFrame.getSize(), amqFrame.getBodyFrame()));
+            }
+            else if (frame instanceof ProtocolInitiation)
+            {
+                byte[] data =  new byte[(int) frame.getSize()];
+                frame.writePayload(new ByteBufferSender()
+                {
+                    @Override
+                    public boolean isDirectBufferPreferred()
+                    {
+                        return false;
+                    }
+
+                    @Override
+                    public void send(final QpidByteBuffer msg)
+                    {
+                        msg.copyTo(data);
+                    }
+
+                    @Override
+                    public void flush()
+                    {
+
+                    }
+
+                    @Override
+                    public void close()
+                    {
+
+                    }
+                });
+
+                result.add(new HeaderResponse(data));
+            }
+            else
+            {
+                throw new IllegalArgumentException(String.format("Unexpected data block received %s", frame));
+            }
+        }
+        _methodProcessor.getProcessedMethods().removeAll(receivedFrames);
+        return result;
+    }
+
+    ProtocolVersion getVersion()
+    {
+        return _methodProcessor.getProtocolVersion();
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/c6d80d80/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/FrameEncoder.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/FrameEncoder.java b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/FrameEncoder.java
new file mode 100644
index 0000000..9b471fc
--- /dev/null
+++ b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/FrameEncoder.java
@@ -0,0 +1,83 @@
+/*
+ *
+ * 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.tests.protocol.v0_8;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.protocol.v0_8.transport.AMQDataBlock;
+import org.apache.qpid.server.transport.ByteBufferSender;
+import org.apache.qpid.tests.protocol.OutputEncoder;
+
+public class FrameEncoder implements OutputEncoder
+{
+
+    @Override
+    public ByteBuffer encode(final Object msg)
+    {
+        if (msg instanceof AMQDataBlock)
+        {
+            final List<ByteBuffer>buffers = new ArrayList<>();
+            ((AMQDataBlock)msg).writePayload(new ByteBufferSender()
+            {
+                @Override
+                public boolean isDirectBufferPreferred()
+                {
+                    return false;
+                }
+
+                @Override
+                public void send(final QpidByteBuffer msg)
+                {
+                    byte[] data = new byte[msg.remaining()];
+                    msg.get(data);
+                    buffers.add(ByteBuffer.wrap(data));
+                }
+
+                @Override
+                public void flush()
+                {
+                }
+
+                @Override
+                public void close()
+                {
+
+                }
+            });
+            int remaining = 0;
+            for (ByteBuffer byteBuffer: buffers)
+            {
+                remaining += byteBuffer.remaining();
+            }
+            ByteBuffer result = ByteBuffer.allocate(remaining);
+            for (ByteBuffer byteBuffer: buffers)
+            {
+                result.put(byteBuffer);
+            }
+            result.flip();
+            return result;
+        }
+        return null;
+    }
+}

http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/c6d80d80/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/FrameTransport.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/FrameTransport.java b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/FrameTransport.java
new file mode 100644
index 0000000..52cd7a0
--- /dev/null
+++ b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/FrameTransport.java
@@ -0,0 +1,101 @@
+/*
+ *
+ * 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.tests.protocol.v0_8;
+
+import java.net.InetSocketAddress;
+
+import org.apache.qpid.server.model.Protocol;
+import org.apache.qpid.server.plugin.ProtocolEngineCreator;
+import org.apache.qpid.server.plugin.QpidServiceLoader;
+import org.apache.qpid.server.protocol.ProtocolVersion;
+
+
+public class FrameTransport extends org.apache.qpid.tests.protocol.FrameTransport
+{
+    private final byte[] _protocolHeader;
+    private ProtocolVersion _protocolVersion;
+
+    public FrameTransport(final InetSocketAddress brokerAddress)
+    {
+        this(brokerAddress, Protocol.AMQP_0_9_1);
+    }
+    public FrameTransport(final InetSocketAddress brokerAddress, Protocol protocol)
+    {
+        super(brokerAddress, new FrameDecoder(getProtocolVersion(protocol)), new FrameEncoder());
+        _protocolVersion = getProtocolVersion(protocol);
+        byte[] protocolHeader = null;
+        for(ProtocolEngineCreator installedEngine : (new QpidServiceLoader()).instancesOf(ProtocolEngineCreator.class))
+        {
+            if (installedEngine.getVersion() == protocol)
+            {
+                protocolHeader = installedEngine.getHeaderIdentifier();
+            }
+        }
+
+        if (protocolHeader == null)
+        {
+            throw new IllegalArgumentException(String.format("Unsupported protocol %s", protocol));
+        }
+        _protocolHeader = protocolHeader;
+    }
+
+    @Override
+    public FrameTransport connect()
+    {
+        super.connect();
+        return this;
+    }
+
+    public Interaction newInteraction()
+    {
+        return new Interaction(this);
+    }
+
+    public byte[] getProtocolHeader()
+    {
+        return _protocolHeader;
+    }
+
+    public ProtocolVersion getProtocolVersion()
+    {
+        return _protocolVersion;
+    }
+
+    public static ProtocolVersion getProtocolVersion(Protocol protocol)
+    {
+        final ProtocolVersion protocolVersion;
+        switch (protocol)
+        {
+            case AMQP_0_8:
+                protocolVersion = ProtocolVersion.v0_8;
+                break;
+            case AMQP_0_9_1:
+                protocolVersion = ProtocolVersion.v0_91;
+                break;
+            case AMQP_0_9:
+                protocolVersion = ProtocolVersion.v0_9;
+                break;
+            default:
+                throw new IllegalArgumentException(String.format("Unsupported protocol %s", protocol));
+        }
+        return protocolVersion;
+    }
+}

http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/c6d80d80/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/Interaction.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/Interaction.java b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/Interaction.java
new file mode 100644
index 0000000..0b62770
--- /dev/null
+++ b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/Interaction.java
@@ -0,0 +1,109 @@
+/*
+ *
+ * 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.tests.protocol.v0_8;
+
+import org.apache.qpid.server.protocol.v0_8.transport.AMQBody;
+import org.apache.qpid.server.protocol.v0_8.transport.AMQDataBlock;
+import org.apache.qpid.server.protocol.v0_8.transport.AMQFrame;
+import org.apache.qpid.server.protocol.v0_8.transport.ConnectionOpenOkBody;
+import org.apache.qpid.server.protocol.v0_8.transport.ConnectionStartBody;
+import org.apache.qpid.server.protocol.v0_8.transport.ConnectionTuneBody;
+
+public class Interaction extends org.apache.qpid.tests.protocol.Interaction<Interaction>
+{
+
+    private int _channelId;
+    private int _maximumPayloadSize = 512;
+
+    Interaction(final FrameTransport transport)
+    {
+        super(transport);
+    }
+
+    @Override
+    protected byte[] getProtocolHeader()
+    {
+        return getTransport().getProtocolHeader();
+    }
+
+    @Override
+    protected Interaction getInteraction()
+    {
+        return this;
+    }
+
+    public Interaction sendPerformative(final AMQBody amqBody) throws Exception
+    {
+        return sendPerformative(getChannelId(), amqBody);
+    }
+
+    public Interaction sendPerformative(int channel, final AMQBody amqBody) throws Exception
+    {
+        final AMQFrame frameBody = new AMQFrame(channel, amqBody);
+        sendPerformativeAndChainFuture(frameBody, false);
+        return this;
+    }
+
+    public Interaction sendPerformative(final AMQDataBlock dataBlock) throws Exception
+    {
+        sendPerformativeAndChainFuture(dataBlock, false);
+        return this;
+    }
+
+    public Interaction openAnonymousConnection() throws Exception
+    {
+        return this.negotiateProtocol().consumeResponse(ConnectionStartBody.class)
+                   .connection().startOkMechanism("ANONYMOUS").startOk().consumeResponse(ConnectionTuneBody.class)
+                   .connection().tuneOk()
+                   .connection().open().consumeResponse(ConnectionOpenOkBody.class);
+
+    }
+
+    public ConnectionInteraction connection()
+    {
+        return new ConnectionInteraction(this);
+    }
+
+    public ChannelInteraction channel()
+    {
+        return new ChannelInteraction(this);
+    }
+
+    public QueueInteraction queue()
+    {
+        return new QueueInteraction(this);
+    }
+
+    public int getChannelId()
+    {
+        return _channelId;
+    }
+
+    public int getMaximumFrameSize()
+    {
+        return _maximumPayloadSize;
+    }
+
+    public BasicInteraction basic()
+    {
+        return new BasicInteraction(this);
+    }
+}

http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/c6d80d80/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/PerformativeResponse.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/PerformativeResponse.java b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/PerformativeResponse.java
new file mode 100644
index 0000000..66871e9
--- /dev/null
+++ b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/PerformativeResponse.java
@@ -0,0 +1,54 @@
+/*
+ * 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.tests.protocol.v0_8;
+
+import org.apache.qpid.server.protocol.v0_8.transport.AMQBody;
+import org.apache.qpid.server.protocol.v0_8.transport.AMQFrame;
+import org.apache.qpid.tests.protocol.Response;
+
+public class PerformativeResponse implements Response<AMQBody>
+{
+    private final int _channel;
+    private final long _size;
+    private final AMQBody _body;
+
+    public PerformativeResponse(int channel, long size, final AMQBody body)
+    {
+        _channel = channel;
+        _size = size;
+        _body = body;
+    }
+
+    @Override
+    public AMQBody getBody()
+    {
+        return _body;
+    }
+
+    @Override
+    public String toString()
+    {
+        return "PerformativeResponse{" +
+               "_channel=" + _channel +
+               ", _size=" + _size +
+               ", _body=" + _body +
+               '}';
+    }
+}

http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/c6d80d80/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/QueueInteraction.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/QueueInteraction.java b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/QueueInteraction.java
new file mode 100644
index 0000000..6e86385
--- /dev/null
+++ b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/QueueInteraction.java
@@ -0,0 +1,63 @@
+/*
+ *
+ * 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.tests.protocol.v0_8;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.qpid.server.protocol.v0_8.AMQShortString;
+import org.apache.qpid.server.protocol.v0_8.FieldTable;
+import org.apache.qpid.server.protocol.v0_8.transport.QueueDeclareBody;
+
+public class QueueInteraction
+{
+    private Interaction _interaction;
+    private String _declareName;
+    private boolean _declarePassive;
+    private boolean _declareDurable;
+    private boolean _declareExclusive;
+    private boolean _declareAutoDelete;
+    private boolean _declareNowait;
+    private Map<String, Object> _declareArguments = new HashMap<>();
+
+    public QueueInteraction(final Interaction interaction)
+    {
+        _interaction = interaction;
+    }
+
+    public QueueInteraction declareName(String name)
+    {
+        _declareName = name;
+        return this;
+    }
+
+    public Interaction declare() throws Exception
+    {
+        return _interaction.sendPerformative(new QueueDeclareBody(0,
+                                                                  AMQShortString.valueOf(_declareName),
+                                                                  _declarePassive,
+                                                                  _declareDurable,
+                                                                  _declareExclusive,
+                                                                  _declareAutoDelete,
+                                                                  _declareNowait,
+                                                                  FieldTable.convertToFieldTable(_declareArguments)));
+    }
+}

http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/c6d80d80/systests/protocol-tests-amqp-0-8/src/main/resources/config-protocol-tests-0-8.json
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-0-8/src/main/resources/config-protocol-tests-0-8.json b/systests/protocol-tests-amqp-0-8/src/main/resources/config-protocol-tests-0-8.json
new file mode 100644
index 0000000..d3738c9
--- /dev/null
+++ b/systests/protocol-tests-amqp-0-8/src/main/resources/config-protocol-tests-0-8.json
@@ -0,0 +1,78 @@
+/*
+ *
+ * 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.
+ *
+ */
+{
+  "name" : "${broker.name}",
+  "modelVersion" : "7.0",
+  "authenticationproviders" : [ {
+    "name" : "anon",
+    "type" : "Anonymous"
+  }, {
+    "name" : "plain",
+    "type" : "Plain",
+    "secureOnlyMechanisms" : [],
+    "users" : [ {
+      "name" : "admin",
+      "type" : "managed",
+      "password" : "admin"
+    }, {
+      "name" : "guest",
+      "type" : "managed",
+      "password" : "guest"
+    } ]
+  } ],
+  "ports" : [ {
+    "name" : "AMQP",
+    "type" : "AMQP",
+    "authenticationProvider" : "plain",
+    "port" : "0",
+    "protocols" : [ "AMQP_0_8", "AMQP_0_9", "AMQP_0_9_1" ],
+    "virtualhostaliases" : [ {
+      "name" : "defaultAlias",
+      "type" : "defaultAlias"
+    }, {
+      "name" : "hostnameAlias",
+      "type" : "hostnameAlias"
+    }, {
+      "name" : "nameAlias",
+      "type" : "nameAlias"
+    } ]
+  }, {
+    "name" : "ANONYMOUS_AMQP",
+    "type" : "AMQP",
+    "authenticationProvider" : "anon",
+    "port" : "0",
+    "protocols" : [ "AMQP_0_8", "AMQP_0_9", "AMQP_0_9_1" ],
+    "virtualhostaliases" : [ {
+      "name" : "defaultAlias",
+      "type" : "defaultAlias",
+      "durable" : true
+    }, {
+      "name" : "hostnameAlias",
+      "type" : "hostnameAlias",
+      "durable" : true
+    }, {
+      "name" : "nameAlias",
+      "type" : "nameAlias",
+      "durable" : true
+    } ]
+  } ],
+  "virtualhostnodes" : []
+}

http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/c6d80d80/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/BasicTest.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/BasicTest.java b/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/BasicTest.java
new file mode 100644
index 0000000..eb51925
--- /dev/null
+++ b/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/BasicTest.java
@@ -0,0 +1,156 @@
+/*
+ *
+ * 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.tests.protocol.v0_8;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.CoreMatchers.nullValue;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.protocol.v0_8.AMQShortString;
+import org.apache.qpid.server.protocol.v0_8.FieldTable;
+import org.apache.qpid.server.protocol.v0_8.transport.BasicConsumeOkBody;
+import org.apache.qpid.server.protocol.v0_8.transport.BasicContentHeaderProperties;
+import org.apache.qpid.server.protocol.v0_8.transport.BasicDeliverBody;
+import org.apache.qpid.server.protocol.v0_8.transport.BasicQosOkBody;
+import org.apache.qpid.server.protocol.v0_8.transport.ChannelCloseOkBody;
+import org.apache.qpid.server.protocol.v0_8.transport.ChannelFlowOkBody;
+import org.apache.qpid.server.protocol.v0_8.transport.ChannelOpenOkBody;
+import org.apache.qpid.server.protocol.v0_8.transport.ContentBody;
+import org.apache.qpid.server.protocol.v0_8.transport.ContentHeaderBody;
+import org.apache.qpid.tests.protocol.SpecificationTest;
+import org.apache.qpid.tests.utils.BrokerAdmin;
+import org.apache.qpid.tests.utils.BrokerAdminUsingTestBase;
+
+public class BasicTest extends BrokerAdminUsingTestBase
+{
+    private InetSocketAddress _brokerAddress;
+
+    @Before
+    public void setUp()
+    {
+        _brokerAddress = getBrokerAdmin().getBrokerAddress(BrokerAdmin.PortType.ANONYMOUS_AMQP);
+        getBrokerAdmin().createQueue(BrokerAdmin.TEST_QUEUE_NAME);
+    }
+
+    @Test
+    @SpecificationTest(section = "1.8.3.7", description = "publish a message")
+    public void publishMessage() throws Exception
+    {
+        try(FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+        {
+            final Interaction interaction = transport.newInteraction();
+            interaction.openAnonymousConnection()
+                       .channel().open().consumeResponse(ChannelOpenOkBody.class)
+                       .basic().contentHeaderPropertiesContentType("text/plain")
+                               .contentHeaderPropertiesHeaders(Collections.singletonMap("test", "testValue"))
+                               .contentHeaderPropertiesDeliveryMode((byte)1)
+                               .contentHeaderPropertiesPriority((byte)1)
+                               .publishExchange("")
+                               .publishRoutingKey(BrokerAdmin.TEST_QUEUE_NAME)
+                               .content("Test")
+                               .publishMessage()
+                       .channel().close()
+                       .consumeResponse(ChannelCloseOkBody.class);
+
+            assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(1)));
+        }
+    }
+
+
+    @Test
+    @SpecificationTest(section = "1.8.3.3", description = " start a queue consumer")
+    public void consumeMessage() throws Exception
+    {
+        try(FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+        {
+            final Interaction interaction = transport.newInteraction();
+            String messageContent = "Test";
+            String consumerTag = "A";
+            String queueName = BrokerAdmin.TEST_QUEUE_NAME;
+            Map<String, Object> messageHeaders = Collections.singletonMap("test", "testValue");
+            String messageContentType = "text/plain";
+            byte deliveryMode = (byte) 1;
+            byte priority = (byte) 2;
+            interaction.openAnonymousConnection()
+                       .channel().open().consumeResponse(ChannelOpenOkBody.class)
+                       .basic().qosPrefetchCount(1).qos().consumeResponse(BasicQosOkBody.class)
+                       .basic().consumeConsumerTag(consumerTag)
+                       .consumeQueue(queueName)
+                       .consume().consumeResponse(BasicConsumeOkBody.class)
+                       .channel().flow(true).consumeResponse(ChannelFlowOkBody.class)
+                       .basic().contentHeaderPropertiesContentType(messageContentType)
+                       .contentHeaderPropertiesHeaders(messageHeaders)
+                       .contentHeaderPropertiesDeliveryMode(deliveryMode)
+                       .contentHeaderPropertiesPriority(priority)
+                       .publishExchange("")
+                       .publishRoutingKey(queueName)
+                       .content(messageContent)
+                       .publishMessage()
+                       .consumeResponse(BasicDeliverBody.class);
+
+            BasicDeliverBody delivery = interaction.getLatestResponse(BasicDeliverBody.class);
+            assertThat(delivery.getConsumerTag(), is(equalTo(AMQShortString.valueOf(consumerTag))));
+            assertThat(delivery.getConsumerTag(), is(notNullValue()));
+            assertThat(delivery.getRedelivered(), is(equalTo(false)));
+            assertThat(delivery.getExchange(), is(nullValue()));
+            assertThat(delivery.getRoutingKey(), is(equalTo(AMQShortString.valueOf(queueName))));
+
+            ContentHeaderBody header =
+                    interaction.consumeResponse(ContentHeaderBody.class).getLatestResponse(ContentHeaderBody.class);
+
+            assertThat(header.getBodySize(), is(equalTo((long)messageContent.length())));
+            BasicContentHeaderProperties properties = header.getProperties();
+            Map<String, Object> receivedHeaders = new HashMap<>(FieldTable.convertToMap(properties.getHeaders()));
+            assertThat(receivedHeaders, is(equalTo(new HashMap<>(messageHeaders))));
+            assertThat(properties.getContentTypeAsString(), is(equalTo(messageContentType)));
+            assertThat(properties.getPriority(), is(equalTo(priority)));
+            assertThat(properties.getDeliveryMode(), is(equalTo(deliveryMode)));
+
+            ContentBody content = interaction.consumeResponse(ContentBody.class).getLatestResponse(ContentBody.class);
+
+            QpidByteBuffer payload = content.getPayload();
+            byte[] contentData = new byte[payload.remaining()];
+            payload.get(contentData);
+            payload.dispose();
+            String receivedContent = new String(contentData, StandardCharsets.UTF_8);
+
+            assertThat(receivedContent, is(equalTo(messageContent)));
+            assertThat(getBrokerAdmin().getQueueDepthMessages(queueName), is(equalTo(1)));
+
+            interaction.basic().ackDeliveryTag(delivery.getDeliveryTag())
+                              .ack()
+                       .channel().close().consumeResponse(ChannelCloseOkBody.class);
+            assertThat(getBrokerAdmin().getQueueDepthMessages(queueName), is(equalTo(0)));
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/c6d80d80/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/ChannelTest.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/ChannelTest.java b/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/ChannelTest.java
new file mode 100644
index 0000000..4ca4ae8
--- /dev/null
+++ b/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/ChannelTest.java
@@ -0,0 +1,54 @@
+/*
+ *
+ * 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.tests.protocol.v0_8;
+
+import java.net.InetSocketAddress;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.qpid.server.protocol.v0_8.transport.ChannelOpenOkBody;
+import org.apache.qpid.tests.protocol.SpecificationTest;
+import org.apache.qpid.tests.utils.BrokerAdmin;
+import org.apache.qpid.tests.utils.BrokerAdminUsingTestBase;
+
+public class ChannelTest extends BrokerAdminUsingTestBase
+{
+    private InetSocketAddress _brokerAddress;
+
+    @Before
+    public void setUp()
+    {
+        _brokerAddress = getBrokerAdmin().getBrokerAddress(BrokerAdmin.PortType.ANONYMOUS_AMQP);
+    }
+
+    @Test
+    @SpecificationTest(section = "1.4.2.1", description = "start connection negotiation")
+    public void channelOpen() throws Exception
+    {
+        try(FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+        {
+            final Interaction interaction = transport.newInteraction();
+            interaction.openAnonymousConnection()
+                       .channel().open().consumeResponse(ChannelOpenOkBody.class);
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/c6d80d80/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/ConnectionTest.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/ConnectionTest.java b/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/ConnectionTest.java
new file mode 100644
index 0000000..1fbbf2a
--- /dev/null
+++ b/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/ConnectionTest.java
@@ -0,0 +1,183 @@
+/*
+ *
+ * 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.tests.protocol.v0_8;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
+
+import java.net.InetSocketAddress;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.qpid.server.protocol.v0_8.transport.ConnectionCloseBody;
+import org.apache.qpid.server.protocol.v0_8.transport.ConnectionOpenOkBody;
+import org.apache.qpid.server.protocol.v0_8.transport.ConnectionStartBody;
+import org.apache.qpid.server.protocol.v0_8.transport.ConnectionTuneBody;
+import org.apache.qpid.tests.protocol.SpecificationTest;
+import org.apache.qpid.tests.utils.BrokerAdmin;
+import org.apache.qpid.tests.utils.BrokerAdminUsingTestBase;
+
+public class ConnectionTest extends BrokerAdminUsingTestBase
+{
+    private InetSocketAddress _brokerAddress;
+
+    @Before
+    public void setUp()
+    {
+        _brokerAddress = getBrokerAdmin().getBrokerAddress(BrokerAdmin.PortType.ANONYMOUS_AMQP);
+    }
+
+    @Test
+    @SpecificationTest(section = "1.4.2.1", description = "start connection negotiation")
+    public void connectionStart() throws Exception
+    {
+        try(FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+        {
+            final Interaction interaction = transport.newInteraction();
+            ConnectionStartBody response =
+                    interaction.negotiateProtocol().consumeResponse().getLatestResponse(ConnectionStartBody.class);
+
+            assertThat(response.getVersionMajor(), is(equalTo((short)transport.getProtocolVersion().getMajorVersion())));
+            assertThat(response.getVersionMinor(), is(equalTo((short)transport.getProtocolVersion().getActualMinorVersion())));
+        }
+    }
+
+    @Test
+    @SpecificationTest(section = "1.4.2.2", description = "select security mechanism and locale")
+    public void connectionStartOk() throws Exception
+    {
+        try (FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+        {
+            final Interaction interaction = transport.newInteraction();
+            interaction.negotiateProtocol()
+                       .consumeResponse(ConnectionStartBody.class)
+                       .connection().startOkMechanism("ANONYMOUS")
+                                    .startOk()
+                       .consumeResponse();
+
+            interaction.getLatestResponse(ConnectionTuneBody.class);
+        }
+    }
+
+    @Test
+    @SpecificationTest(section = "1.4.2.5", description = "select security mechanism and locale")
+    public void connectionTuneOkAndOpen() throws Exception
+    {
+        try (FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+        {
+            final Interaction interaction = transport.newInteraction();
+            ConnectionTuneBody response = interaction.negotiateProtocol()
+                                                     .consumeResponse(ConnectionStartBody.class)
+                                                     .connection().startOkMechanism("ANONYMOUS")
+                                                     .startOk()
+                                                     .consumeResponse().getLatestResponse(ConnectionTuneBody.class);
+
+            interaction.connection().tuneOkChannelMax(response.getChannelMax())
+                       .tuneOkFrameMax(response.getFrameMax())
+                       .tuneOkHeartbeat(response.getHeartbeat())
+                       .tuneOk()
+                       .connection().open()
+                       .consumeResponse().getLatestResponse(ConnectionOpenOkBody.class);
+        }
+    }
+
+    @Test
+    @SpecificationTest(section = "1.4.2.5", description = "[...] the minimum negotiated value for frame-max is also"
+                                                          + " frame-min-size [4096].")
+    public void tooSmallFrameSize() throws Exception
+    {
+        try (FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+        {
+            final Interaction interaction = transport.newInteraction();
+            ConnectionTuneBody response = interaction.negotiateProtocol()
+                                                     .consumeResponse(ConnectionStartBody.class)
+                                                     .connection().startOkMechanism("ANONYMOUS")
+                                                     .startOk()
+                                                     .consumeResponse().getLatestResponse(ConnectionTuneBody.class);
+
+            interaction.connection().tuneOkChannelMax(response.getChannelMax())
+                       .tuneOkFrameMax(1024)
+                       .tuneOkHeartbeat(response.getHeartbeat())
+                       .tuneOk()
+                       .consumeResponse().getLatestResponse(ConnectionCloseBody.class);
+        }
+    }
+
+    @Test
+    @SpecificationTest(section = "1.4.2.5.2.", description = "If the client specifies a frame max that is higher than"
+                                                             + " the value provided by the server, the server MUST"
+                                                             + " close the connection without attempting a negotiated"
+                                                             + " close. The server may report the error in some fashion"
+                                                             + " to assist implementors.")
+    public void tooLargeFrameSize() throws Exception
+    {
+        try (FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+        {
+            final Interaction interaction = transport.newInteraction();
+            ConnectionTuneBody response = interaction.negotiateProtocol()
+                                                     .consumeResponse(ConnectionStartBody.class)
+                                                     .connection().startOkMechanism("ANONYMOUS")
+                                                     .startOk()
+                                                     .consumeResponse().getLatestResponse(ConnectionTuneBody.class);
+
+            interaction.connection().tuneOkChannelMax(response.getChannelMax())
+                       .tuneOkFrameMax(Long.MAX_VALUE)
+                       .tuneOkHeartbeat(response.getHeartbeat())
+                       .tuneOk()
+                       .consumeResponse().getLatestResponse(ConnectionCloseBody.class);
+        }
+    }
+
+    @Test
+    @SpecificationTest(section = "1.4.", description = "open connection = C:protocolheader S:START C:START OK"
+                                                       + " *challenge S:TUNE C:TUNE OK C:OPEN S:OPEN OK")
+    public void authenticationBypassBySendingTuneOk() throws Exception
+    {
+        final InetSocketAddress brokerAddress = getBrokerAdmin().getBrokerAddress(BrokerAdmin.PortType.AMQP);
+        try (FrameTransport transport = new FrameTransport(brokerAddress).connect())
+        {
+            final Interaction interaction = transport.newInteraction();
+            interaction.negotiateProtocol().consumeResponse(ConnectionStartBody.class)
+                       .connection().tuneOk()
+                       .consumeResponse().getLatestResponse(ConnectionCloseBody.class);
+        }
+    }
+
+
+    @Test
+    @SpecificationTest(section = "1.4.", description = "open connection = C:protocolheader S:START C:START OK"
+                                                       + " *challenge S:TUNE C:TUNE OK C:OPEN S:OPEN OK")
+    public void authenticationBypassBySendingOpen() throws Exception
+    {
+        final InetSocketAddress brokerAddress = getBrokerAdmin().getBrokerAddress(BrokerAdmin.PortType.AMQP);
+        try (FrameTransport transport = new FrameTransport(brokerAddress).connect())
+        {
+            final Interaction interaction = transport.newInteraction();
+            interaction.negotiateProtocol().consumeResponse(ConnectionStartBody.class)
+                       .connection().open()
+                       .consumeResponse().getLatestResponse(ConnectionCloseBody.class);
+        }
+    }
+
+
+}

http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/c6d80d80/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/QueueTest.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/QueueTest.java b/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/QueueTest.java
new file mode 100644
index 0000000..920ea73
--- /dev/null
+++ b/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/QueueTest.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.tests.protocol.v0_8;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+import java.net.InetSocketAddress;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.qpid.server.protocol.v0_8.AMQShortString;
+import org.apache.qpid.server.protocol.v0_8.transport.ChannelOpenOkBody;
+import org.apache.qpid.server.protocol.v0_8.transport.QueueDeclareOkBody;
+import org.apache.qpid.tests.protocol.SpecificationTest;
+import org.apache.qpid.tests.utils.BrokerAdmin;
+import org.apache.qpid.tests.utils.BrokerAdminUsingTestBase;
+
+public class QueueTest extends BrokerAdminUsingTestBase
+{
+    private InetSocketAddress _brokerAddress;
+
+    @Before
+    public void setUp()
+    {
+        _brokerAddress = getBrokerAdmin().getBrokerAddress(BrokerAdmin.PortType.ANONYMOUS_AMQP);
+    }
+
+    @Test
+    @SpecificationTest(section = "1.4.2.1", description = "start connection negotiation")
+    public void queueDeclare() throws Exception
+    {
+        try(FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+        {
+            final Interaction interaction = transport.newInteraction();
+            final String queueName = "testQueue";
+            QueueDeclareOkBody response = interaction.openAnonymousConnection()
+                                                     .channel().open().consumeResponse(ChannelOpenOkBody.class)
+                                                     .queue().declareName(queueName).declare()
+                                                     .consumeResponse().getLatestResponse(QueueDeclareOkBody.class);
+
+            assertThat(response.getQueue(), is(equalTo(AMQShortString.valueOf(queueName))));
+            assertThat(response.getMessageCount(), is(equalTo(0L)));
+            assertThat(response.getConsumerCount(), is(equalTo(0L)));
+        }
+    }
+}


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org


Mime
View raw message