https://issues.apache.org/activemq/browse/AMQ-1547 - selector for rest consumer

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

activemq-web-demo/src/test/java/org/apache/activemq/web/JettyServer.java

@@ -0,0 +1,80 @@
+/**
+ * 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.web;
+
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.demo.DefaultQueueSender;
+import org.eclipse.jetty.server.Connector;
+import org.eclipse.jetty.server.Server;
+import org.eclipse.jetty.server.nio.SelectChannelConnector;
+import org.eclipse.jetty.webapp.WebAppContext;
+
+/**
+ * A simple bootstrap class for starting Jetty in your IDE using the local web
+ * application.
+ * 
+ * @version $Revision$
+ */
+public final class JettyServer {
+
+    public static final int PORT = 8080;
+
+    public static final String WEBAPP_DIR = "src/main/webapp";
+
+    public static final String WEBAPP_CTX = "/";
+
+    private JettyServer() {
+    }
+
+    public static void main(String[] args) throws Exception {
+        // lets create a broker
+        BrokerService broker = new BrokerService();
+        broker.setPersistent(false);
+        broker.setUseJmx(true);
+        broker.addConnector("tcp://localhost:61616");
+        broker.addConnector("stomp://localhost:61613");
+        broker.start();
+
+        // lets publish some messages so that there is some stuff to browse
+        DefaultQueueSender.main(new String[] {
+            "FOO.BAR"
+        });
+
+        // now lets start the web server
+        int port = PORT;
+        if (args.length > 0) {
+            String text = args[0];
+            port = Integer.parseInt(text);
+        }
+        System.out.println("Starting Web Server on port: " + port);
+        Server server = new Server();
+        SelectChannelConnector connector = new SelectChannelConnector();
+        connector.setPort(port);
+        connector.setServer(server);
+        WebAppContext context = new WebAppContext();
+
+        context.setResourceBase(WEBAPP_DIR);
+        context.setContextPath(WEBAPP_CTX);
+        context.setServer(server);
+        server.setHandler(context);
+        server.setConnectors(new Connector[] {
+            connector
+        });
+        server.start();
+    }
+}

activemq-web-demo/src/test/java/org/apache/activemq/web/JettyTestSupport.java

@@ -0,0 +1,65 @@
+package org.apache.activemq.web;
+
+import javax.jms.Connection;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.eclipse.jetty.server.Connector;
+import org.eclipse.jetty.server.Server;
+import org.eclipse.jetty.server.nio.SelectChannelConnector;
+import org.eclipse.jetty.webapp.WebAppContext;
+
+import junit.framework.TestCase;
+
+public class JettyTestSupport extends TestCase {
+
+    BrokerService broker;
+    Server server;
+    ActiveMQConnectionFactory factory;
+    Connection connection;
+    Session session;
+    MessageProducer producer;
+    
+    protected void setUp() throws Exception {
+        broker = new BrokerService();
+        broker.setPersistent(false);
+        broker.setUseJmx(true);
+        broker.addConnector("tcp://localhost:61616");
+        broker.start();
+        broker.waitUntilStarted();
+        
+        server = new Server();
+        SelectChannelConnector connector = new SelectChannelConnector();
+        connector.setPort(8080);
+        connector.setServer(server);
+        WebAppContext context = new WebAppContext();
+
+        context.setResourceBase("src/main/webapp");
+        context.setContextPath("/");
+        context.setServer(server);
+        server.setHandler(context);
+        server.setConnectors(new Connector[] {
+            connector
+        });
+        server.start();   
+        
+        factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
+        connection = factory.createConnection();
+        connection.start();
+        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        producer = session.createProducer(session.createQueue("test"));
+    }
+
+    protected void tearDown() throws Exception {
+        server.stop();
+        broker.stop();
+        broker.waitUntilStopped();
+        session.close();
+        connection.close();
+    }
+
+    
+    
+}

activemq-web-demo/src/test/java/org/apache/activemq/web/RestTest.java

@@ -0,0 +1,44 @@
+package org.apache.activemq.web;
+
+import javax.jms.TextMessage;
+
+import org.eclipse.jetty.client.ContentExchange;
+import org.eclipse.jetty.client.HttpClient;
+
+public class RestTest extends JettyTestSupport {
+	
+	public void testConsume() throws Exception {
+	    producer.send(session.createTextMessage("test"));
+	    
+	    HttpClient httpClient = new HttpClient();
+        httpClient.start();
+        ContentExchange contentExchange = new ContentExchange();
+        httpClient.setConnectorType(HttpClient.CONNECTOR_SELECT_CHANNEL);
+        contentExchange.setURL("http://localhost:8080/message/test?timeout=1000&type=queue");
+        httpClient.send(contentExchange);
+        contentExchange.waitForDone();
+        assertEquals("test", contentExchange.getResponseContent());
+	    
+	}
+	
+	public void testSelector() throws Exception {
+	    TextMessage msg1 = session.createTextMessage("test1");
+	    msg1.setIntProperty("test", 1);
+	    producer.send(msg1);
+	    
+	    TextMessage msg2 = session.createTextMessage("test2");
+	    msg2.setIntProperty("test", 2);
+	    producer.send(msg2);
+	    
+        HttpClient httpClient = new HttpClient();
+        httpClient.start();
+        ContentExchange contentExchange = new ContentExchange();
+        httpClient.setConnectorType(HttpClient.CONNECTOR_SELECT_CHANNEL);
+        contentExchange.setURL("http://localhost:8080/message/test?timeout=1000&type=queue");
+        contentExchange.setRequestHeader(WebClient.SELECTOR_NAME, "test=2");
+        httpClient.send(contentExchange);
+        contentExchange.waitForDone();
+        assertEquals("test2", contentExchange.getResponseContent());
+	}
+
+}

activemq-web-demo/src/test/resources/log4j.properties

@@ -0,0 +1,36 @@
+## ---------------------------------------------------------------------------
+## 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.
+## ---------------------------------------------------------------------------
+
+#  
+# The logging properties used during tests..
+#
+log4j.rootLogger=INFO, out, stdout
+
+log4j.logger.org.apache.activemq.spring=WARN
+log4j.logger.org.apache.activemq.web=DEBUG
+
+# CONSOLE appender not used by default
+log4j.appender.stdout=org.apache.log4j.ConsoleAppender
+log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern=%d [%-15.15t] %-5p %-30.30c{1} - %m%n
+
+# File appender
+log4j.appender.out=org.apache.log4j.FileAppender
+log4j.appender.out.layout=org.apache.log4j.PatternLayout
+log4j.appender.out.layout.ConversionPattern=%d [%-15.15t] %-5p %-30.30c{1} - %m%n
+log4j.appender.out.file=target/test-reports/activemq-test.log
+log4j.appender.out.append=true

activemq-web/src/main/java/org/apache/activemq/web/MessageListenerServlet.java

@@ -156,7 +156,7 @@ public class MessageListenerServlet extends MessageServletSupport {
                         Map<MessageAvailableConsumer, String> consumerDestinationMap = getConsumerDestinationNameMap(request);
                         client.closeConsumer(destination); // drop any existing
                         // consumer.
-                        MessageAvailableConsumer consumer = (MessageAvailableConsumer)client.getConsumer(destination);
+                        MessageAvailableConsumer consumer = (MessageAvailableConsumer)client.getConsumer(destination, request.getHeader(WebClient.SELECTOR_NAME));
 
                         consumer.setAvailableListener(listener);
                         consumerIdMap.put(consumer, message);
@@ -167,7 +167,7 @@ public class MessageListenerServlet extends MessageServletSupport {
                     } else if ("unlisten".equals(type)) {
                         Map<MessageAvailableConsumer, String> consumerIdMap = getConsumerIdMap(request);
                         Map consumerDestinationMap = getConsumerDestinationNameMap(request);
-                        MessageAvailableConsumer consumer = (MessageAvailableConsumer)client.getConsumer(destination);
+                        MessageAvailableConsumer consumer = (MessageAvailableConsumer)client.getConsumer(destination, request.getHeader(WebClient.SELECTOR_NAME));
 
                         consumer.setAvailableListener(null);
                         consumerIdMap.remove(consumer);

activemq-web/src/main/java/org/apache/activemq/web/MessageServlet.java

@@ -180,7 +180,7 @@ public class MessageServlet extends MessageServletSupport {
                 LOG.debug("Receiving message(s) from: " + destination + " with timeout: " + timeout);
             }
 
-            MessageAvailableConsumer consumer = (MessageAvailableConsumer)client.getConsumer(destination);
+            MessageAvailableConsumer consumer = (MessageAvailableConsumer)client.getConsumer(destination, request.getHeader(WebClient.SELECTOR_NAME));
             Continuation continuation = null;
             Listener listener = null;
             Message message = null;
@@ -297,7 +297,7 @@ public class MessageServlet extends MessageServletSupport {
                 LOG.debug("Receiving message(s) from: " + destination + " with timeout: " + timeout);
             }
 
-            MessageAvailableConsumer consumer = (MessageAvailableConsumer)client.getConsumer(destination);
+            MessageAvailableConsumer consumer = (MessageAvailableConsumer)client.getConsumer(destination, request.getHeader(WebClient.SELECTOR_NAME));
             Message message = null;
 
             // write a responds

activemq-web/src/main/java/org/apache/activemq/web/MessageServletSupport.java

@@ -353,4 +353,8 @@ public abstract class MessageServletSupport extends HttpServlet {
         }
         return answer;
     }
+    
+    protected String getSelector(HttpServletRequest request) throws IOException {
+    	return request.getHeader(WebClient.SELECTOR_NAME);
+    }
 }

activemq-web/src/main/java/org/apache/activemq/web/WebClient.java

@@ -70,6 +70,7 @@ public class WebClient implements HttpSessionActivationListener, HttpSessionBind
     public static final String CONNECTION_FACTORY_PREFETCH_PARAM = "org.apache.activemq.connectionFactory.prefetch";
     public static final String CONNECTION_FACTORY_OPTIMIZE_ACK_PARAM = "org.apache.activemq.connectionFactory.optimizeAck";
     public static final String BROKER_URL_INIT_PARAM = "org.apache.activemq.brokerURL";
+    public static final String SELECTOR_NAME = "org.apache.activemq.selector";
 
     private static final Log LOG = LogFactory.getLog(WebClient.class);
 
@@ -196,7 +197,7 @@ public class WebClient implements HttpSessionActivationListener, HttpSessionBind
 
                 try {
                     Destination destination = destinationName.startsWith("topic://") ? (Destination)getSession().createTopic(destinationName) : (Destination)getSession().createQueue(destinationName);
-                    consumers.put(destination, getConsumer(destination, true));
+                    consumers.put(destination, getConsumer(destination, null, true));
                 } catch (JMSException e) {
                     LOG.debug("Caought Exception ", e);
                     IOException ex = new IOException(e.getMessage());
@@ -304,14 +305,14 @@ public class WebClient implements HttpSessionActivationListener, HttpSessionBind
         this.producer = producer;
     }
 
-    public synchronized MessageConsumer getConsumer(Destination destination) throws JMSException {
+    public synchronized MessageConsumer getConsumer(Destination destination, String selector) throws JMSException {
-        return getConsumer(destination, true);
+        return getConsumer(destination, selector, true);
     }
 
-    public synchronized MessageConsumer getConsumer(Destination destination, boolean create) throws JMSException {
+    public synchronized MessageConsumer getConsumer(Destination destination, String selector, boolean create) throws JMSException {
         MessageConsumer consumer = consumers.get(destination);
         if (create && consumer == null) {
-            consumer = getSession().createConsumer(destination);
+            consumer = getSession().createConsumer(destination, selector);
             consumers.put(destination, consumer);
         }
         return consumer;