diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/builders/AmqpConsumerBuilder.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/builders/AmqpConsumerBuilder.java index 60b590f89..6a2c2635b 100644 --- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/builders/AmqpConsumerBuilder.java +++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/builders/AmqpConsumerBuilder.java @@ -65,6 +65,7 @@ public class AmqpConsumerBuilder extends AmqpResourceBuilder filter = ((Source) remoteSource).getFilter(); + return filter != null && filter.containsKey(JMS_SELECTOR_SYMBOL); + } + + return false; + } + private void configureSource(Source source) { Map filters = new HashMap(); Symbol[] outcomes = new Symbol[]{ Accepted.DESCRIPTOR_SYMBOL, Rejected.DESCRIPTOR_SYMBOL, diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SessionIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SessionIntegrationTest.java index c4c06d41d..0e2279aee 100644 --- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SessionIntegrationTest.java +++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SessionIntegrationTest.java @@ -34,7 +34,9 @@ import java.io.IOException; import java.util.ArrayList; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.Random; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -55,6 +57,7 @@ import org.apache.qpid.jms.test.testpeer.basictypes.TerminusDurability; import org.apache.qpid.jms.test.testpeer.describedtypes.Accepted; import org.apache.qpid.jms.test.testpeer.describedtypes.Rejected; +import org.apache.qpid.jms.test.testpeer.describedtypes.Source; import org.apache.qpid.jms.test.testpeer.describedtypes.sections.AmqpValueDescribedType; import org.apache.qpid.jms.test.testpeer.describedtypes.sections.HeaderDescribedType; import org.apache.qpid.jms.test.testpeer.matchers.AcceptedMatcher; @@ -378,6 +381,60 @@ private void doCreateConsumerWithSelectorTestImpl(String messageSelector, boolea } } + @Test + @Timeout(20) + public void testCreateConsumerFailsWhenSelectorFilterNotEchoedAtAll() throws Exception { + // Peer answers with a source carrying no filter section. + doCreateConsumerFailsWhenSelectorFilterNotInPlaceTestImpl(null); + } + + @Test + @Timeout(20) + public void testCreateConsumerFailsWhenSelectorFilterOmittedFromEchoedFilterSet() throws Exception { + // Peer answers with an empty filter set, which is what a broker sends when it strips a + // filter it will not apply from the set it echoes back. + doCreateConsumerFailsWhenSelectorFilterNotInPlaceTestImpl(new HashMap()); + } + + private void doCreateConsumerFailsWhenSelectorFilterNotInPlaceTestImpl(Map responseFilter) throws Exception { + try (TestAmqpPeer testPeer = new TestAmqpPeer();) { + Connection connection = testFixture.establishConnecton(testPeer); + connection.start(); + + testPeer.expectBegin(); + + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + Queue queue = session.createQueue("myQueue"); + String selector = "myProp = 'foo'"; + + // "the sending endpoint sets the filter actually in place" (AMQP 1.0 section 3.5.3), + // so a response without the selector filter says the peer will not apply it. The + // consumer must fail rather than silently deliver messages the selector excludes. + Source responseSource = new Source(); + responseSource.setAddress("myQueue"); + responseSource.setFilter(responseFilter); + + Matcher filterMapMatcher = hasEntry(equalTo(Symbol.valueOf("jms-selector")), notNullValue()); + SourceMatcher sourceMatcher = new SourceMatcher(); + sourceMatcher.withFilter(filterMapMatcher); + + testPeer.expectReceiverAttachWithResponseSource(notNullValue(), sourceMatcher, responseSource); + testPeer.expectDetach(true, true, true); + + try { + session.createConsumer(queue, selector); + fail("Expected an exception to be thrown"); + } catch (JMSException jmse) { + // Expected + } + + testPeer.expectClose(); + connection.close(); + + testPeer.waitForAllHandlersToComplete(1000); + } + } + @Test @Timeout(20) public void testCreateConsumerFailsWhenLinkRefusedAndAttachResponseWriteIsNotDeferred() throws Exception { diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/test/testpeer/TestAmqpPeer.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/test/testpeer/TestAmqpPeer.java index c56341777..2bd4a0ef7 100644 --- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/test/testpeer/TestAmqpPeer.java +++ b/qpid-jms-client/src/test/java/org/apache/qpid/jms/test/testpeer/TestAmqpPeer.java @@ -1519,6 +1519,16 @@ public void expectReceiverAttach(final Matcher linkNameMatcher, final Matcher expectReceiverAttach(linkNameMatcher, sourceMatcher, settled, refuseLink, omitDetach, deferAttachResponseWrite, errorType, errorMessage, null, null, null); } + /** + * Accepts the receiver attach but answers with the given source instead of echoing back the + * source that was received, so that a test can control which filters the peer reports as + * actually being in place. + */ + public void expectReceiverAttachWithResponseSource(final Matcher linkNameMatcher, final Matcher sourceMatcher, final Source responseSource) + { + expectReceiverAttach(linkNameMatcher, sourceMatcher, false, false, false, false, null, null, responseSource, null, null); + } + private void expectReceiverAttach(final Matcher linkNameMatcher, final Matcher sourceMatcher, final boolean settled, final boolean refuseLink, boolean omitDetach, boolean deferAttachResponseWrite, Symbol errorType, String errorMessage, final Source responseSourceOverride, Matcher desiredCapabilitiesMatcher, Symbol[] offeredCapabilitiesResponse)