Throw exception only for queues when selector is not null

Selectors are supported with topics.

[#159272967]

Fixes #52

(cherry picked from commit 920afc1)

Author: acogoluegnes

src/main/java/com/rabbitmq/jms/client/RMQSession.java

@@ -746,8 +746,10 @@ public MessageConsumer createConsumer(Destination destination, String messageSel
         illegalStateExceptionIfClosed();
         if (nullOrEmpty(messageSelector)) {
             return createConsumer(destination);
+        } else if (isTopic(destination)) {
+            return createConsumerInternal((RMQDestination) destination, null, false, messageSelector);
         } else {
-            // we are not implementing this method yet
+            // selectors are not supported for queues
             throw new UnsupportedOperationException();
         }
     }
@@ -756,6 +758,10 @@ private static boolean nullOrEmpty(String str) {
         return str==null || str.trim().isEmpty();
     }
 
+    private static boolean isTopic(Destination destination) {
+        return !((RMQDestination) destination).isQueue();
+    }
+
     /**
      * {@inheritDoc}
      * @throws UnsupportedOperationException - method not implemented until we support selectors
@@ -767,8 +773,12 @@ public MessageConsumer createConsumer(Destination destination, String messageSel
             RMQMessageConsumer consumer = (RMQMessageConsumer)createConsumer(destination);
             consumer.setNoLocal(noLocal);
             return consumer;
-        } else {
-            // we are not implementing this method yet
+        } else if (isTopic(destination)) {
+            RMQMessageConsumer consumer = createConsumerInternal((RMQDestination) destination, null, false, messageSelector);
+            consumer.setNoLocal(noLocal);
+            return consumer;
+        }  else {
+            // selectors are not supported for queues
             throw new UnsupportedOperationException();
         }
     }

src/test/java/com/rabbitmq/integration/tests/SelectorAppliedToTopicNotQueueIT.java

@@ -0,0 +1,103 @@
+/* Copyright (c) 2018 Pivotal Software, Inc. All rights reserved. */
+package com.rabbitmq.integration.tests;
+
+import com.rabbitmq.jms.admin.RMQConnectionFactory;
+import com.rabbitmq.jms.client.message.RMQTextMessage;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import javax.jms.DeliveryMode;
+import javax.jms.JMSException;
+import javax.jms.MessageConsumer;
+import javax.jms.Queue;
+import javax.jms.QueueConnection;
+import javax.jms.QueueSession;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import javax.jms.Topic;
+import javax.jms.TopicConnection;
+import javax.jms.TopicPublisher;
+import javax.jms.TopicSession;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/**
+ * To make sure Session#createConsumer works for at least topic when a
+ * selector is provided.
+ *
+ * See https://github.com/rabbitmq/rabbitmq-jms-client/issues/52
+ */
+public class SelectorAppliedToTopicNotQueueIT {
+
+    private static final String TOPIC_NAME = "test.topic." + SelectorAppliedToTopicNotQueueIT.class.getCanonicalName();
+    private static final String QUEUE_NAME = "test.queue." + SimpleQueueMessageIT.class.getCanonicalName();
+    private static final String MESSAGE1 = "Hello " + SelectorAppliedToTopicNotQueueIT.class.getName();
+
+    RMQConnectionFactory connFactory;
+    TopicConnection topicConn;
+    QueueConnection queueConn;
+
+    @BeforeEach
+    public void beforeTests() throws Exception {
+        this.connFactory =
+            (RMQConnectionFactory) AbstractTestConnectionFactory.getTestConnectionFactory()
+                .getConnectionFactory();
+        this.topicConn = connFactory.createTopicConnection();
+        this.queueConn = connFactory.createQueueConnection();
+    }
+
+    @AfterEach
+    public void afterTests() throws Exception {
+        if (this.topicConn != null)
+            this.topicConn.close();
+        if (queueConn != null)
+            queueConn.close();
+    }
+
+    @Test
+    public void sendAndReceiveSupportedOnTopic() throws Exception {
+        topicConn.start();
+        TopicSession topicSession = topicConn.createTopicSession(false, Session.DUPS_OK_ACKNOWLEDGE);
+        Topic topic = topicSession.createTopic(TOPIC_NAME);
+        MessageConsumer receiver = topicSession.createConsumer(topic, "boolProp");
+        sendAndReceive(topicSession, topic, receiver);
+    }
+
+    @Test
+    public void sendAndReceiveSupportedOnTopicNoLocal() throws Exception {
+        topicConn.start();
+        TopicSession topicSession = topicConn.createTopicSession(false, Session.DUPS_OK_ACKNOWLEDGE);
+        Topic topic = topicSession.createTopic(TOPIC_NAME);
+        MessageConsumer receiver = topicSession.createConsumer(topic, "boolProp", false);
+        sendAndReceive(topicSession, topic, receiver);
+    }
+
+    private void sendAndReceive(TopicSession topicSession, Topic topic, MessageConsumer receiver) throws JMSException {
+        TopicPublisher sender = topicSession.createPublisher(topic);
+        sender.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
+        TextMessage message = topicSession.createTextMessage(MESSAGE1);
+        message.setBooleanProperty("boolProp", true);
+        sender.send(message);
+        RMQTextMessage tmsg = (RMQTextMessage) receiver.receive();
+        String t = tmsg.getText();
+        assertEquals(MESSAGE1, t);
+    }
+
+    @Test
+    public void sendAndReceiveNotSupportedOnQueue() throws Exception {
+        queueConn.start();
+        QueueSession queueSession = queueConn.createQueueSession(false, Session.DUPS_OK_ACKNOWLEDGE);
+        Queue queue = queueSession.createQueue(QUEUE_NAME);
+        assertThrows(UnsupportedOperationException.class, () -> queueSession.createConsumer(queue, "boolProp"));
+    }
+
+    @Test
+    public void sendAndReceiveNotSupportedOnQueueNoLocal() throws Exception {
+        queueConn.start();
+        QueueSession queueSession = queueConn.createQueueSession(false, Session.DUPS_OK_ACKNOWLEDGE);
+        Queue queue = queueSession.createQueue(QUEUE_NAME);
+        assertThrows(UnsupportedOperationException.class, () -> queueSession.createConsumer(queue, "boolProp", false));
+    }
+}