activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From tab...@apache.org
Subject activemq git commit: https://issues.apache.org/jira/browse/AMQ-5666
Date Tue, 17 Mar 2015 19:29:06 GMT
Repository: activemq
Updated Branches:
  refs/heads/master 934ad44ad -> 1a0f73ed1


https://issues.apache.org/jira/browse/AMQ-5666

Add some tests and cleanup of the testing client

Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/1a0f73ed
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/1a0f73ed
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/1a0f73ed

Branch: refs/heads/master
Commit: 1a0f73ed193fce9152bad866b4116d5f0f437337
Parents: 934ad44
Author: Timothy Bish <tabish121@gmail.com>
Authored: Tue Mar 17 15:28:48 2015 -0400
Committer: Timothy Bish <tabish121@gmail.com>
Committed: Tue Mar 17 15:28:48 2015 -0400

----------------------------------------------------------------------
 .../amqp/client/AmqpAbstractResource.java       |   8 +-
 .../transport/amqp/client/AmqpClient.java       |   8 +-
 .../transport/amqp/client/AmqpConnection.java   |   3 +-
 .../transport/amqp/client/AmqpSession.java      |  94 ++++++++++++++++-
 .../amqp/client/AmqpStateInspector.java         |  88 ----------------
 .../amqp/client/AmqpUnknownFilterType.java      |  44 ++++++++
 .../transport/amqp/client/AmqpValidator.java    | 101 +++++++++++++++++++
 .../amqp/client/util/UnmodifiableLink.java      |  16 +--
 .../amqp/interop/AmqpConnectionsTest.java       |   9 +-
 .../amqp/interop/AmqpReceiverTest.java          |  82 +++++++++++++++
 10 files changed, 340 insertions(+), 113 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/activemq/blob/1a0f73ed/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpAbstractResource.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpAbstractResource.java
b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpAbstractResource.java
index b5a6324..fbc4dda 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpAbstractResource.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpAbstractResource.java
@@ -41,7 +41,7 @@ public abstract class AmqpAbstractResource<E extends Endpoint> implements
AmqpRe
     protected AsyncResult openRequest;
     protected AsyncResult closeRequest;
 
-    private AmqpStateInspector amqpStateInspector = new AmqpStateInspector();
+    private AmqpValidator amqpStateInspector;
 
     private E endpoint;
 
@@ -184,13 +184,13 @@ public abstract class AmqpAbstractResource<E extends Endpoint>
implements AmqpRe
         this.endpoint = endpoint;
     }
 
-    public AmqpStateInspector getStateInspector() {
+    public AmqpValidator getStateInspector() {
         return amqpStateInspector;
     }
 
-    public void setStateInspector(AmqpStateInspector stateInspector) {
+    public void setStateInspector(AmqpValidator stateInspector) {
         if (stateInspector == null) {
-            stateInspector = new AmqpStateInspector();
+            stateInspector = new AmqpValidator();
         }
 
         this.amqpStateInspector = stateInspector;

http://git-wip-us.apache.org/repos/asf/activemq/blob/1a0f73ed/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpClient.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpClient.java
b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpClient.java
index 0b299e4..e7d3eaf 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpClient.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpClient.java
@@ -38,7 +38,7 @@ public class AmqpClient {
     private final String password;
     private final URI remoteURI;
 
-    private AmqpStateInspector stateInspector = new AmqpStateInspector();
+    private AmqpValidator stateInspector = new AmqpValidator();
     private List<Symbol> offeredCapabilities = Collections.emptyList();
     private Map<Symbol, Object> offeredProperties = Collections.emptyMap();
 
@@ -176,7 +176,7 @@ public class AmqpClient {
     /**
      * @return the currently set state inspector used to check state after various events.
      */
-    public AmqpStateInspector getStateInspector() {
+    public AmqpValidator getStateInspector() {
         return stateInspector;
     }
 
@@ -187,9 +187,9 @@ public class AmqpClient {
      * @param stateInspector
      *        the new state inspector to use.
      */
-    public void setStateInspector(AmqpStateInspector stateInspector) {
+    public void setStateInspector(AmqpValidator stateInspector) {
         if (stateInspector == null) {
-            stateInspector = new AmqpStateInspector();
+            stateInspector = new AmqpValidator();
         }
 
         this.stateInspector = stateInspector;

http://git-wip-us.apache.org/repos/asf/activemq/blob/1a0f73ed/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpConnection.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpConnection.java
b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpConnection.java
index a98f711..b4fd661 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpConnection.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpConnection.java
@@ -222,10 +222,9 @@ public class AmqpConnection extends AmqpAbstractResource<Connection>
implements
             @Override
             public void run() {
                 checkClosed();
-
                 session.setEndpoint(getEndpoint().session());
+                session.setStateInspector(getStateInspector());
                 session.open(request);
-
                 pumpToProtonTransport();
             }
         });

http://git-wip-us.apache.org/repos/asf/activemq/blob/1a0f73ed/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpSession.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpSession.java
b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpSession.java
index b2fc2f1..c747dc6 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpSession.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpSession.java
@@ -69,6 +69,7 @@ public class AmqpSession extends AmqpAbstractResource<Session> {
             @Override
             public void run() {
                 checkClosed();
+                sender.setStateInspector(getStateInspector());
                 sender.open(request);
                 pumpToProtonTransport();
             }
@@ -83,23 +84,63 @@ public class AmqpSession extends AmqpAbstractResource<Session> {
      * Create a receiver instance using the given address
      *
      * @param address
-     * 	      the address to which the receiver will subscribe for its messages.
+     *        the address to which the receiver will subscribe for its messages.
      *
      * @return a newly created receiver that is ready for use.
      *
      * @throws Exception if an error occurs while creating the receiver.
      */
     public AmqpReceiver createReceiver(String address) throws Exception {
+        return createReceiver(address, null, false);
+    }
+
+    /**
+     * Create a receiver instance using the given address
+     *
+     * @param address
+     *        the address to which the receiver will subscribe for its messages.
+     * @param selector
+     *        the JMS selector to use for the subscription
+     *
+     * @return a newly created receiver that is ready for use.
+     *
+     * @throws Exception if an error occurs while creating the receiver.
+     */
+    public AmqpReceiver createReceiver(String address, String selector) throws Exception
{
+        return createReceiver(address, selector, false);
+    }
+
+    /**
+     * Create a receiver instance using the given address
+     *
+     * @param address
+     * 	      the address to which the receiver will subscribe for its messages.
+     * @param selector
+     *        the JMS selector to use for the subscription
+     * @param noLocal
+     *        should the subscription have messages from its connection filtered.
+     *
+     * @return a newly created receiver that is ready for use.
+     *
+     * @throws Exception if an error occurs while creating the receiver.
+     */
+    public AmqpReceiver createReceiver(String address, String selector, boolean noLocal)
throws Exception {
         checkClosed();
 
-        final AmqpReceiver receiver = new AmqpReceiver(AmqpSession.this, address, getNextReceiverId());
         final ClientFuture request = new ClientFuture();
+        final AmqpReceiver receiver = new AmqpReceiver(AmqpSession.this, address, getNextReceiverId());
+
+        receiver.setNoLocal(noLocal);
+        if (selector != null && !selector.isEmpty()) {
+            receiver.setSelector(selector);
+        }
 
         connection.getScheduler().execute(new Runnable() {
 
             @Override
             public void run() {
                 checkClosed();
+                receiver.setStateInspector(getStateInspector());
                 receiver.open(request);
                 pumpToProtonTransport();
             }
@@ -123,17 +164,64 @@ public class AmqpSession extends AmqpAbstractResource<Session>
{
      * @throws Exception if an error occurs while creating the receiver.
      */
     public AmqpReceiver createDurableReceiver(String address, String subscriptionName) throws
Exception {
+        return createDurableReceiver(address, subscriptionName, null, false);
+    }
+
+    /**
+     * Create a receiver instance using the given address that creates a durable subscription.
+     *
+     * @param address
+     *        the address to which the receiver will subscribe for its messages.
+     * @param subscriptionName
+     *        the name of the subscription that is being created.
+     * @param selector
+     *        the JMS selector to use for the subscription
+     *
+     * @return a newly created receiver that is ready for use.
+     *
+     * @throws Exception if an error occurs while creating the receiver.
+     */
+    public AmqpReceiver createDurableReceiver(String address, String subscriptionName, String
selector) throws Exception {
+        return createDurableReceiver(address, subscriptionName, selector, false);
+    }
+
+    /**
+     * Create a receiver instance using the given address that creates a durable subscription.
+     *
+     * @param address
+     *        the address to which the receiver will subscribe for its messages.
+     * @param subscriptionName
+     *        the name of the subscription that is being created.
+     * @param selector
+     *        the JMS selector to use for the subscription
+     * @param noLocal
+     *        should the subscription have messages from its connection filtered.
+     *
+     * @return a newly created receiver that is ready for use.
+     *
+     * @throws Exception if an error occurs while creating the receiver.
+     */
+    public AmqpReceiver createDurableReceiver(String address, String subscriptionName, String
selector, boolean noLocal) throws Exception {
         checkClosed();
 
+        if (subscriptionName == null || subscriptionName.isEmpty()) {
+            throw new IllegalArgumentException("subscription name must not be null or empty.");
+        }
+
+        final ClientFuture request = new ClientFuture();
         final AmqpReceiver receiver = new AmqpReceiver(AmqpSession.this, address, getNextReceiverId());
         receiver.setSubscriptionName(subscriptionName);
-        final ClientFuture request = new ClientFuture();
+        receiver.setNoLocal(noLocal);
+        if (selector != null && !selector.isEmpty()) {
+            receiver.setSelector(selector);
+        }
 
         connection.getScheduler().execute(new Runnable() {
 
             @Override
             public void run() {
                 checkClosed();
+                receiver.setStateInspector(getStateInspector());
                 receiver.open(request);
                 pumpToProtonTransport();
             }

http://git-wip-us.apache.org/repos/asf/activemq/blob/1a0f73ed/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpStateInspector.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpStateInspector.java
b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpStateInspector.java
deleted file mode 100644
index 5471876..0000000
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpStateInspector.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.activemq.transport.amqp.client;
-
-import org.apache.qpid.proton.engine.Connection;
-import org.apache.qpid.proton.engine.Link;
-import org.apache.qpid.proton.engine.Session;
-
-/**
- * Abstract base for a validation hook that is used in tests to check
- * the state of a remote resource after a variety of lifecycle events.
- */
-public class AmqpStateInspector {
-
-    private boolean valid = true;
-    private String errorMessage;
-
-    public void inspectOpenedResource(Connection connection) {
-
-    }
-
-    public void inspectOpenedResource(Session session) {
-
-    }
-
-    public void inspectOpenedResource(Link link) {
-
-    }
-
-    public void inspectClosedResource(Connection remoteConnection) {
-
-    }
-
-    public void inspectClosedResource(Session session) {
-
-    }
-
-    public void inspectClosedResource(Link link) {
-
-    }
-
-    public void inspectDetachedResource(Link link) {
-
-    }
-
-    public boolean isValid() {
-        return valid;
-    }
-
-    protected void setValid(boolean valid) {
-        this.valid = valid;
-    }
-
-    public String getErrorMessage() {
-        return errorMessage;
-    }
-
-    protected void setErrorMessage(String errorMessage) {
-        this.errorMessage = errorMessage;
-    }
-
-    protected void markAsInvalid(String errorMessage) {
-        if (valid) {
-            setValid(false);
-            setErrorMessage(errorMessage);
-        }
-    }
-
-    public void assertIfStateChecksFailed() {
-        if (!isValid()) {
-            throw new AssertionError(errorMessage);
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/activemq/blob/1a0f73ed/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpUnknownFilterType.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpUnknownFilterType.java
b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpUnknownFilterType.java
new file mode 100644
index 0000000..c86a2c9
--- /dev/null
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpUnknownFilterType.java
@@ -0,0 +1,44 @@
+/**
+ * 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.activemq.transport.amqp.client;
+
+import org.apache.qpid.proton.amqp.DescribedType;
+import org.apache.qpid.proton.amqp.UnsignedLong;
+
+/**
+ * A Described Type wrapper for an unsupported filter that the broker should ignore.
+ */
+public class AmqpUnknownFilterType implements DescribedType {
+
+    public static final AmqpUnknownFilterType UNKOWN_FILTER = new AmqpUnknownFilterType();
+
+    private final String payload;
+
+    public AmqpUnknownFilterType() {
+        this.payload = "UnknownFilter{}";
+    }
+
+    @Override
+    public Object getDescriptor() {
+        return UnsignedLong.valueOf(0x0000468C00000099L);
+    }
+
+    @Override
+    public Object getDescribed() {
+        return this.payload;
+    }
+}

http://git-wip-us.apache.org/repos/asf/activemq/blob/1a0f73ed/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpValidator.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpValidator.java
b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpValidator.java
new file mode 100644
index 0000000..fc088fc
--- /dev/null
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpValidator.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.activemq.transport.amqp.client;
+
+import org.apache.qpid.proton.engine.Connection;
+import org.apache.qpid.proton.engine.Receiver;
+import org.apache.qpid.proton.engine.Sender;
+import org.apache.qpid.proton.engine.Session;
+
+/**
+ * Abstract base for a validation hook that is used in tests to check
+ * the state of a remote resource after a variety of lifecycle events.
+ */
+public class AmqpValidator {
+
+    private boolean valid = true;
+    private String errorMessage;
+
+    public void inspectOpenedResource(Connection connection) {
+
+    }
+
+    public void inspectOpenedResource(Session session) {
+
+    }
+
+    public void inspectOpenedResource(Sender sender) {
+
+    }
+
+    public void inspectOpenedResource(Receiver receiver) {
+
+    }
+
+    public void inspectClosedResource(Connection remoteConnection) {
+
+    }
+
+    public void inspectClosedResource(Session session) {
+
+    }
+
+    public void inspectClosedResource(Sender sender) {
+
+    }
+
+    public void inspectClosedResource(Receiver receiver) {
+
+    }
+
+    public void inspectDetachedResource(Sender sender) {
+
+    }
+
+    public void inspectDetachedResource(Receiver receiver) {
+
+    }
+
+    public boolean isValid() {
+        return valid;
+    }
+
+    protected void setValid(boolean valid) {
+        this.valid = valid;
+    }
+
+    public String getErrorMessage() {
+        return errorMessage;
+    }
+
+    protected void setErrorMessage(String errorMessage) {
+        this.errorMessage = errorMessage;
+    }
+
+    protected void markAsInvalid(String errorMessage) {
+        if (valid) {
+            setValid(false);
+            setErrorMessage(errorMessage);
+        }
+    }
+
+    public void assertValid() {
+        if (!isValid()) {
+            throw new AssertionError(errorMessage);
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/activemq/blob/1a0f73ed/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableLink.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableLink.java
b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableLink.java
index 70665c0..c7a99d3 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableLink.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableLink.java
@@ -124,14 +124,14 @@ public class UnmodifiableLink implements Link {
 
     @Override
     public Source getSource() {
-        // TODO Auto-generated method stub
-        return null;
+        // TODO Figure out a simple way to wrap the odd Source types in Proton-J
+        return link.getSource();
     }
 
     @Override
     public Target getTarget() {
-        // TODO Auto-generated method stub
-        return null;
+        // TODO Figure out a simple way to wrap the odd Source types in Proton-J
+        return link.getTarget();
     }
 
     @Override
@@ -146,14 +146,14 @@ public class UnmodifiableLink implements Link {
 
     @Override
     public Source getRemoteSource() {
-        // TODO Auto-generated method stub
-        return null;
+        // TODO Figure out a simple way to wrap the odd Source types in Proton-J
+        return link.getSource();
     }
 
     @Override
     public Target getRemoteTarget() {
-        // TODO Auto-generated method stub
-        return null;
+        // TODO Figure out a simple way to wrap the odd Target types in Proton-J
+        return link.getTarget();
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/activemq/blob/1a0f73ed/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpConnectionsTest.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpConnectionsTest.java
b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpConnectionsTest.java
index a35709d..2f9935f 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpConnectionsTest.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpConnectionsTest.java
@@ -26,7 +26,7 @@ import java.util.Map;
 import org.apache.activemq.transport.amqp.client.AmqpClient;
 import org.apache.activemq.transport.amqp.client.AmqpClientTestSupport;
 import org.apache.activemq.transport.amqp.client.AmqpConnection;
-import org.apache.activemq.transport.amqp.client.AmqpStateInspector;
+import org.apache.activemq.transport.amqp.client.AmqpValidator;
 import org.apache.qpid.proton.amqp.Symbol;
 import org.apache.qpid.proton.amqp.transport.AmqpError;
 import org.apache.qpid.proton.amqp.transport.ErrorCondition;
@@ -62,7 +62,7 @@ public class AmqpConnectionsTest extends AmqpClientTestSupport {
         AmqpClient client = createAmqpClient();
         assertNotNull(client);
 
-        client.setStateInspector(new AmqpStateInspector() {
+        client.setStateInspector(new AmqpValidator() {
 
             @Override
             public void inspectOpenedResource(Connection connection) {
@@ -88,6 +88,7 @@ public class AmqpConnectionsTest extends AmqpClientTestSupport {
 
         assertEquals(1, getProxyToBroker().getCurrentConnectionsCount());
 
+        connection.getStateInspector().assertValid();
         connection.close();
 
         assertEquals(0, getProxyToBroker().getCurrentConnectionsCount());
@@ -131,7 +132,7 @@ public class AmqpConnectionsTest extends AmqpClientTestSupport {
         connection1.connect();
         assertEquals(1, getProxyToBroker().getCurrentConnectionsCount());
 
-        connection2.setStateInspector(new AmqpStateInspector() {
+        connection2.setStateInspector(new AmqpValidator() {
 
             @Override
             public void inspectOpenedResource(Connection connection) {
@@ -160,7 +161,7 @@ public class AmqpConnectionsTest extends AmqpClientTestSupport {
             LOG.info("Second connection with same container Id failed as expected.");
         }
 
-        connection2.getStateInspector().assertIfStateChecksFailed();
+        connection2.getStateInspector().assertValid();
 
         assertEquals(1, getProxyToBroker().getCurrentConnectionsCount());
 

http://git-wip-us.apache.org/repos/asf/activemq/blob/1a0f73ed/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpReceiverTest.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpReceiverTest.java
b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpReceiverTest.java
index 1bc3d66..cdecab0 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpReceiverTest.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpReceiverTest.java
@@ -16,10 +16,14 @@
  */
 package org.apache.activemq.transport.amqp.interop;
 
+import static org.apache.activemq.transport.amqp.AmqpSupport.JMS_SELECTOR_FILTER_IDS;
+import static org.apache.activemq.transport.amqp.AmqpSupport.NO_LOCAL_FILTER_IDS;
+import static org.apache.activemq.transport.amqp.AmqpSupport.findFilter;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.activemq.broker.jmx.QueueViewMBean;
@@ -29,7 +33,11 @@ import org.apache.activemq.transport.amqp.client.AmqpConnection;
 import org.apache.activemq.transport.amqp.client.AmqpMessage;
 import org.apache.activemq.transport.amqp.client.AmqpReceiver;
 import org.apache.activemq.transport.amqp.client.AmqpSession;
+import org.apache.activemq.transport.amqp.client.AmqpValidator;
 import org.apache.activemq.util.Wait;
+import org.apache.qpid.proton.amqp.Symbol;
+import org.apache.qpid.proton.amqp.messaging.Source;
+import org.apache.qpid.proton.engine.Receiver;
 import org.junit.Ignore;
 import org.junit.Test;
 
@@ -63,6 +71,80 @@ public class AmqpReceiverTest extends AmqpClientTestSupport {
     }
 
     @Test(timeout = 60000)
+    public void testCreateQueueReceiverWithJMSSelector() throws Exception {
+        AmqpClient client = createAmqpClient();
+
+        client.setStateInspector(new AmqpValidator() {
+
+            @SuppressWarnings("unchecked")
+            @Override
+            public void inspectOpenedResource(Receiver receiver) {
+                LOG.info("Receiver opened: {}", receiver);
+
+                if (receiver.getSource() == null) {
+                    markAsInvalid("Link opened with null source.");
+                }
+
+                Source source = (Source) receiver.getSource();
+                Map<Symbol, Object> filters = source.getFilter();
+
+                if (findFilter(filters, JMS_SELECTOR_FILTER_IDS) == null) {
+                    markAsInvalid("Broker did not return the JMS Filter on Attach");
+                }
+            }
+        });
+
+        AmqpConnection connection = client.connect();
+        AmqpSession session = connection.createSession();
+
+        assertEquals(0, brokerService.getAdminView().getQueues().length);
+
+        session.createReceiver("queue://" + getTestName(), "JMSPriority > 8");
+
+        assertEquals(1, brokerService.getAdminView().getQueueSubscribers().length);
+
+        connection.getStateInspector().assertValid();
+        connection.close();
+    }
+
+    @Test(timeout = 60000)
+    public void testCreateQueueReceiverWithNoLocalSet() throws Exception {
+        AmqpClient client = createAmqpClient();
+
+        client.setStateInspector(new AmqpValidator() {
+
+            @SuppressWarnings("unchecked")
+            @Override
+            public void inspectOpenedResource(Receiver receiver) {
+                LOG.info("Receiver opened: {}", receiver);
+
+                if (receiver.getSource() == null) {
+                    markAsInvalid("Link opened with null source.");
+                }
+
+                Source source = (Source) receiver.getSource();
+                Map<Symbol, Object> filters = source.getFilter();
+
+                if (findFilter(filters, NO_LOCAL_FILTER_IDS) == null) {
+                    markAsInvalid("Broker did not return the NoLocal Filter on Attach");
+                }
+            }
+        });
+
+        AmqpConnection connection = client.connect();
+        AmqpSession session = connection.createSession();
+
+        assertEquals(0, brokerService.getAdminView().getQueues().length);
+
+        session.createReceiver("queue://" + getTestName(), null, true);
+
+        assertEquals(1, brokerService.getAdminView().getQueueSubscribers().length);
+
+        connection.getStateInspector().assertValid();
+        connection.close();
+    }
+
+    @Test(timeout = 60000)
     public void testCreateTopicReceiver() throws Exception {
         AmqpClient client = createAmqpClient();
         AmqpConnection connection = client.connect();


Mime
View raw message