Author: bill.burke(a)jboss.com Date: 2010-10-25 15:35:10 -0400 (Mon, 25 Oct 2010) New Revision: 9812 Added: trunk/hornetq-rest/hornetq-rest/src/test/java/org/hornetq/rest/test/SelectorTest.java Modified: trunk/docs/user-manual/en/configuration-index.xml trunk/docs/user-manual/en/embedding-hornetq.xml trunk/hornetq-rest/docbook/reference/en/master.xml trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/AcknowledgedQueueConsumer.java trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/ConsumersResource.java trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/PostMessage.java trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/QueueConsumer.java trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/push/PushConsumer.java trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/push/xml/PushRegistration.java trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/topic/AcknowledgedSubscriptionResource.java trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/topic/FileTopicPushStore.java trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/topic/SubscriptionResource.java trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/topic/SubscriptionsResource.java trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/util/HttpMessageHelper.java Log: fix docs, merge REST from beta2 release Modified: trunk/docs/user-manual/en/configuration-index.xml =================================================================== --- trunk/docs/user-manual/en/configuration-index.xml 2010-10-25 15:49:13 UTC (rev 9811) +++ trunk/docs/user-manual/en/configuration-index.xml 2010-10-25 19:35:10 UTC (rev 9812) @@ -1026,7 +1026,7 @@ <entry>generic</entry> </row> <row> - <entry id="configuration.connection-factory.signature"> + <entry id="configuration.connection-factory.signature.xa"> <link linkend="using-jms.configure.factory.types">connection-factory.xa</link> </entry> <entry>Boolean</entry> Modified: trunk/docs/user-manual/en/embedding-hornetq.xml =================================================================== --- trunk/docs/user-manual/en/embedding-hornetq.xml 2010-10-25 15:49:13 UTC (rev 9811) +++ trunk/docs/user-manual/en/embedding-hornetq.xml 2010-10-25 19:35:10 UTC (rev 9812) @@ -46,19 +46,8 @@ two different helper classes for this depending on whether your using the HornetQ Core API or JMS.</para> -<<<<<<< .mine <section> <title>Core API Only</title> -======= -config.setAcceptorConfigurations(transports);</programlisting> - <para>You need to instantiate and start HornetQ server. The class <literal - >org.hornetq.api.core.server.HornetQ</literal> has a few static methods for creating - servers with common configurations.</para> - <programlisting>import org.hornetq.core.server.HornetQServer; -import org.hornetq.core.server.HornetQServers; - ->>>>>>> .r9629 - <para>For instantiating a core HornetQ Server only, the steps are pretty simple. The example requires that you have defined a configuration file <literal>hornetq-configuration.xml</literal> in your @@ -66,38 +55,9 @@ ... -<<<<<<< .mine EmbeddedHornetQ embedded = new EmbeddedHornetQ(); embedded.start(); -======= -HornetQServer server = HornetQServers.newHornetQServer(config); ->>>>>>> .r9629 -<<<<<<< .mine -// Assuming you defined an "in-vm" acceptor within your hornetq-configuration.xml file -======= -server.start();</programlisting> - <para>You also have the option of instantiating <literal>HornetQServerImpl</literal> - directly:</para> - <programlisting>HornetQServer server = new HornetQServerImpl(config); -server.start();</programlisting> - </section> - <section> - <title>Dependency Frameworks</title> - <para>You may also choose to use a dependency injection framework such as <trademark>JBoss - Micro Container</trademark> or <trademark>Spring Framework</trademark>.</para> - <para>HornetQ standalone uses JBoss Micro Container as the injection framework. <literal - >HornetQBootstrapServer</literal> and <literal>hornetq-beans.xml</literal> which are - part of the HornetQ distribution provide a very complete implementation of what's needed - to bootstrap the server using JBoss Micro Container. </para> - <para>When using JBoss Micro Container, you need to provide an XML file declaring the - <literal>HornetQServer</literal> and <literal>Configuration</literal> object, you - can also inject a security manager and a MBean server if you want, but those are - optional.</para> - <para>A very basic XML Bean declaration for the JBoss Micro Container would be:</para> - <programlisting>&lt;?xml version="1.0" encoding="UTF-8"?> ->>>>>>> .r9629 - ClientSessionFactory nettyFactory = HornetQClient.createClientSessionFactory( new TransportConfiguration( InVMConnectorFactory.class.getName())); Modified: trunk/hornetq-rest/docbook/reference/en/master.xml =================================================================== --- trunk/hornetq-rest/docbook/reference/en/master.xml 2010-10-25 15:49:13 UTC (rev 9811) +++ trunk/hornetq-rest/docbook/reference/en/master.xml 2010-10-25 19:35:10 UTC (rev 9812) @@ -870,6 +870,16 @@ the server. Only usable on topics.</para> </listitem> </varlistentry> + <varlistentry> + <term>selector</term> + + <listitem> + <para>This is an optional JMS selector string. The HornetQ REST + interface adds HTTP headers to the JMS message for REST produced + messages. HTTP headers are prefixed with "http_" and every '-' + charactor is converted to a '$'.</para> + </listitem> + </varlistentry> </variablelist> <sect1> @@ -1480,6 +1490,10 @@ <programlisting>&lt;push-registration&gt; &lt;durable&gt;false&lt;/durable&gt; + &lt;selector&gt;&lt;![CDATA[ + SomeAttribute &gt; 1 + ]]&gt; + &lt;/selector&gt; &lt;link rel="push" href="http://somewhere.com" type="application/json" method="PUT"/&gt; &lt;/push-registration&gt; </programlisting> @@ -1493,6 +1507,10 @@ <literal>queue-push-store-dir</literal> config variable defined in Chapter 2. (<literal>topic-push-store-dir</literal> for topics).</para> + <para>The <literal>selector</literal> element is optional and defines a + JMS message selector. You should enclose it within CDATA blocks as some + of the selector characters are illegal XML.</para> + <para>The <literal>link</literal> element specifies the basis of the interaction. The <literal>href</literal> attribute contains the URL you want to interact with. It is the only required attribute. The @@ -1562,11 +1580,16 @@ <title>The Topic Push Subscription XML</title> <para>The push XML for a topic is the same except the root element is - push-topic-registration. The rest of the document is the same. Here's an + push-topic-registration. (Also remember the <literal>selector</literal> + element is optional). The rest of the document is the same. Here's an example of a template registration:</para> <programlisting>&lt;push-topic-registration&gt; &lt;durable&gt;true&lt;/durable&gt; + &lt;selector&gt;&lt;![CDATA[ + SomeAttribute &gt; 1 + ]]&gt; + &lt;/selector&gt; &lt;link rel="template" href="http://somewhere.com/resources/{id}/messages" method="POST"/&gt; &lt;/push-topic registration&gt;</programlisting> </sect1> Modified: trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/AcknowledgedQueueConsumer.java =================================================================== --- trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/AcknowledgedQueueConsumer.java 2010-10-25 15:49:13 UTC (rev 9811) +++ trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/AcknowledgedQueueConsumer.java 2010-10-25 19:35:10 UTC (rev 9812) @@ -31,10 +31,10 @@ protected String startup = Long.toString(System.currentTimeMillis()); protected volatile Acknowledgement ack; - public AcknowledgedQueueConsumer(ClientSessionFactory factory, String destination, String id, DestinationServiceManager serviceManager) + public AcknowledgedQueueConsumer(ClientSessionFactory factory, String destination, String id, DestinationServiceManager serviceManager, String selector) throws HornetQException { - super(factory, destination, id, serviceManager); + super(factory, destination, id, serviceManager, selector); autoAck = false; } @@ -187,9 +187,7 @@ try { - session = factory.createSession(); - consumer = session.createConsumer(destination); - session.start(); + createSession(); } catch (Exception e) { Modified: trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/ConsumersResource.java =================================================================== --- trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/ConsumersResource.java 2010-10-25 15:49:13 UTC (rev 9811) +++ trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/ConsumersResource.java 2010-10-25 19:35:10 UTC (rev 9812) @@ -36,6 +36,9 @@ protected int consumerTimeoutSeconds; protected DestinationServiceManager serviceManager; + protected static final int ACKNOWLEDGED = 0x01; + protected static final int SELECTOR_SET = 0x02; + public DestinationServiceManager getServiceManager() { return serviceManager; @@ -78,6 +81,7 @@ private Object timeoutLock = new Object(); + @Override public void testTimeout(String target) { synchronized (timeoutLock) @@ -107,32 +111,41 @@ @POST public Response createSubscription(@FormParam("autoAck") @DefaultValue("true") boolean autoAck, + @FormParam("selector") String selector, @Context UriInfo uriInfo) { try { QueueConsumer consumer = null; + int attributes = 0; + if (selector != null) + { + attributes = attributes | SELECTOR_SET; + } + if (autoAck) { - consumer = createConsumer(); + consumer = createConsumer(selector); } else { - consumer = createAcknowledgedConsumer(); + attributes |= ACKNOWLEDGED; + consumer = createAcknowledgedConsumer(selector); } + String attributesSegment = "attributes-" + attributes; UriBuilder location = uriInfo.getAbsolutePathBuilder(); - if (autoAck) location.path("auto-ack"); - else location.path("acknowledged"); + location.path(attributesSegment); location.path(consumer.getId()); Response.ResponseBuilder builder = Response.created(location.build()); + if (autoAck) { - QueueConsumer.setConsumeNextLink(serviceManager.getLinkStrategy(), builder, uriInfo, uriInfo.getMatchedURIs().get(1) + "/auto-ack/" + consumer.getId(), "-1"); + QueueConsumer.setConsumeNextLink(serviceManager.getLinkStrategy(), builder, uriInfo, uriInfo.getMatchedURIs().get(1) + "/" + attributesSegment +"/" + consumer.getId(), "-1"); } else { - AcknowledgedQueueConsumer.setAcknowledgeNextLink(serviceManager.getLinkStrategy(), builder, uriInfo, uriInfo.getMatchedURIs().get(1) + "/acknowledged/" + consumer.getId(), "-1"); + AcknowledgedQueueConsumer.setAcknowledgeNextLink(serviceManager.getLinkStrategy(), builder, uriInfo, uriInfo.getMatchedURIs().get(1) + "/" + attributesSegment +"/" + consumer.getId(), "-1"); } return builder.build(); @@ -147,11 +160,11 @@ } } - public QueueConsumer createConsumer() + public QueueConsumer createConsumer(String selector) throws HornetQException { String genId = sessionCounter.getAndIncrement() + "-queue-" + destination + "-" + startup; - QueueConsumer consumer = new QueueConsumer(sessionFactory, destination, genId, serviceManager); + QueueConsumer consumer = new QueueConsumer(sessionFactory, destination, genId, serviceManager, selector); synchronized (timeoutLock) { queueConsumers.put(genId, consumer); @@ -160,11 +173,11 @@ return consumer; } - public QueueConsumer createAcknowledgedConsumer() + public QueueConsumer createAcknowledgedConsumer(String selector) throws HornetQException { String genId = sessionCounter.getAndIncrement() + "-queue-" + destination + "-" + startup; - QueueConsumer consumer = new AcknowledgedQueueConsumer(sessionFactory, destination, genId, serviceManager); + QueueConsumer consumer = new AcknowledgedQueueConsumer(sessionFactory, destination, genId, serviceManager, selector); synchronized (timeoutLock) { queueConsumers.put(genId, consumer); @@ -173,85 +186,81 @@ return consumer; } - @Path("auto-ack/{consumer-id}") + @Path("attributes-{attributes}/{consumer-id}") @GET - public Response getConsumer(@PathParam("consumer-id") String consumerId, + public Response getConsumer(@PathParam("attributes") int attributes, + @PathParam("consumer-id") String consumerId, @Context UriInfo uriInfo) throws Exception { - return headConsumer(consumerId, uriInfo); + return headConsumer(attributes, consumerId, uriInfo); } - @Path("auto-ack/{consumer-id}") + @Path("attributes-{attributes}/{consumer-id}") @HEAD - public Response headConsumer(@PathParam("consumer-id") String consumerId, + public Response headConsumer(@PathParam("attributes") int attributes, + @PathParam("consumer-id") String consumerId, @Context UriInfo uriInfo) throws Exception { - QueueConsumer consumer = findConsumer(consumerId); + QueueConsumer consumer = findConsumer(attributes, consumerId, uriInfo); Response.ResponseBuilder builder = Response.noContent(); // we synchronize just in case a failed request is still processing synchronized (consumer) { - QueueConsumer.setConsumeNextLink(serviceManager.getLinkStrategy(), builder, uriInfo, uriInfo.getMatchedURIs().get(1) + "/acknowledged/" + consumer.getId(), Long.toString(consumer.getConsumeIndex())); + if ( (attributes & ACKNOWLEDGED) > 0) + { + AcknowledgedQueueConsumer ackedConsumer = (AcknowledgedQueueConsumer)consumer; + Acknowledgement ack = ackedConsumer.getAck(); + if (ack == null || ack.wasSet()) + { + AcknowledgedQueueConsumer.setAcknowledgeNextLink(serviceManager.getLinkStrategy(), builder, uriInfo, uriInfo.getMatchedURIs().get(1) + "/attributes-" + attributes + "/" + consumer.getId(), Long.toString(consumer.getConsumeIndex())); + } + else + { + ackedConsumer.setAcknowledgementLink(builder, uriInfo, uriInfo.getMatchedURIs().get(1) + "/attributes-" + attributes + "/" + consumer.getId()); + } + + } + else + { + QueueConsumer.setConsumeNextLink(serviceManager.getLinkStrategy(), builder, uriInfo, uriInfo.getMatchedURIs().get(1) + "/attributes-" + attributes + "/" + consumer.getId(), Long.toString(consumer.getConsumeIndex())); + } } return builder.build(); } - @Path("auto-ack/{consumer-id}") + @Path("attributes-{attributes}/{consumer-id}") public QueueConsumer findConsumer( - @PathParam("consumer-id") String consumerId) throws Exception + @PathParam("attributes") int attributes, + @PathParam("consumer-id") String consumerId, + @Context UriInfo uriInfo) throws Exception { QueueConsumer consumer = queueConsumers.get(consumerId); if (consumer == null) { - QueueConsumer tmp = new QueueConsumer(sessionFactory, destination, consumerId, serviceManager); - consumer = addConsumerToMap(consumerId, tmp); - } - return consumer; - } + if ( (attributes & SELECTOR_SET) > 0) + { - @Path("acknowledged/{consumer-id}") - @GET - public Response getAcknowledgedConsumer(@PathParam("consumer-id") String consumerId, - @Context UriInfo uriInfo) throws Exception - { - return headAcknowledgedConsumer(consumerId, uriInfo); - } + Response.ResponseBuilder builder = Response.status(Response.Status.GONE) + .entity("Cannot reconnect to selector-based consumer. You must recreate the consumer session.") + .type("text/plain"); + UriBuilder uriBuilder = uriInfo.getBaseUriBuilder(); + uriBuilder.path(uriInfo.getMatchedURIs().get(1)); + serviceManager.getLinkStrategy().setLinkHeader(builder, "pull-consumers", "pull-consumers", uriBuilder.build().toString(), null); + throw new WebApplicationException(builder.build()); + + } + if ( (attributes & ACKNOWLEDGED) > 0) + { + QueueConsumer tmp = new AcknowledgedQueueConsumer(sessionFactory, destination, consumerId, serviceManager, null); + consumer = addConsumerToMap(consumerId, tmp); - @Path("acknowledged/{consumer-id}") - @HEAD - public Response headAcknowledgedConsumer(@PathParam("consumer-id") String consumerId, - @Context UriInfo uriInfo) throws Exception - { - AcknowledgedQueueConsumer consumer = (AcknowledgedQueueConsumer) findAcknowledgedConsumer(consumerId); - Response.ResponseBuilder builder = Response.ok(); - // we synchronize just in case a failed request is still processing - synchronized (consumer) - { - Acknowledgement ack = consumer.getAck(); - if (ack == null || ack.wasSet()) - { - AcknowledgedQueueConsumer.setAcknowledgeNextLink(serviceManager.getLinkStrategy(), builder, uriInfo, uriInfo.getMatchedURIs().get(1) + "/acknowledged/" + consumer.getId(), Long.toString(consumer.getConsumeIndex())); } else { - consumer.setAcknowledgementLink(builder, uriInfo, uriInfo.getMatchedURIs().get(1) + "/acknowledged/" + consumer.getId()); + QueueConsumer tmp = new QueueConsumer(sessionFactory, destination, consumerId, serviceManager, null); + consumer = addConsumerToMap(consumerId, tmp); } } - return builder.build(); - } - - - @Path("acknowledged/{consumer-id}") - public QueueConsumer findAcknowledgedConsumer( - @PathParam("consumer-id") String consumerId) throws Exception - { - QueueConsumer consumer = queueConsumers.get(consumerId); - if (consumer == null) - { - QueueConsumer tmp = new AcknowledgedQueueConsumer(sessionFactory, destination, consumerId, serviceManager); - ; - consumer = addConsumerToMap(consumerId, tmp); - } return consumer; } @@ -275,16 +284,8 @@ } - @Path("acknowledged/{consumer-id}") + @Path("attributes-{attributes}/{consumer-id}") @DELETE - public void closeAcknowledgedSession( - @PathParam("consumer-id") String consumerId) - { - closeSession(consumerId); - } - - @Path("auto-ack/{consumer-id}") - @DELETE public void closeSession( @PathParam("consumer-id") String consumerId) { Modified: trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/PostMessage.java =================================================================== --- trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/PostMessage.java 2010-10-25 15:49:13 UTC (rev 9811) +++ trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/PostMessage.java 2010-10-25 19:35:10 UTC (rev 9812) @@ -1,12 +1,12 @@ package org.hornetq.rest.queue; import org.hornetq.api.core.HornetQException; -import org.hornetq.api.core.Message; import org.hornetq.api.core.client.ClientMessage; import org.hornetq.api.core.client.ClientProducer; import org.hornetq.api.core.client.ClientSession; import org.hornetq.api.core.client.ClientSessionFactory; import org.hornetq.rest.util.HttpMessageHelper; +import org.hornetq.api.core.Message; import javax.ws.rs.POST; import javax.ws.rs.PUT; Modified: trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/QueueConsumer.java =================================================================== --- trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/QueueConsumer.java 2010-10-25 15:49:13 UTC (rev 9811) +++ trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/QueueConsumer.java 2010-10-25 19:35:10 UTC (rev 9812) @@ -5,6 +5,7 @@ import org.hornetq.api.core.client.ClientMessage; import org.hornetq.api.core.client.ClientSession; import org.hornetq.api.core.client.ClientSessionFactory; +import org.hornetq.jms.client.SelectorTranslator; import org.hornetq.rest.util.Constants; import org.hornetq.rest.util.LinkStrategy; @@ -37,6 +38,7 @@ protected long lastPing = System.currentTimeMillis(); protected DestinationServiceManager serviceManager; protected boolean autoAck = true; + protected String selector; /** * token used to create consume-next links @@ -70,14 +72,15 @@ lastPing = System.currentTimeMillis(); } - public QueueConsumer(ClientSessionFactory factory, String destination, String id, DestinationServiceManager serviceManager) throws HornetQException + public QueueConsumer(ClientSessionFactory factory, String destination, String id, DestinationServiceManager serviceManager, String selector) throws HornetQException { this.factory = factory; this.destination = destination; this.id = id; this.serviceManager = serviceManager; + this.selector = selector; - createSession(factory, destination); + createSession(); } public String getId() @@ -191,11 +194,18 @@ } } - protected void createSession(ClientSessionFactory factory, String destination) + protected void createSession() throws HornetQException { session = factory.createSession(true, true); - consumer = session.createConsumer(destination); + if (selector == null) + { + consumer = session.createConsumer(destination); + } + else + { + consumer = session.createConsumer(destination, SelectorTranslator.convertToHornetQFilterString(selector)); + } session.start(); } Modified: trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/push/PushConsumer.java =================================================================== --- trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/push/PushConsumer.java 2010-10-25 15:49:13 UTC (rev 9811) +++ trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/push/PushConsumer.java 2010-10-25 19:35:10 UTC (rev 9812) @@ -7,6 +7,7 @@ import org.hornetq.api.core.client.ClientSessionFactory; import org.hornetq.api.core.client.MessageHandler; import org.hornetq.core.logging.Logger; +import org.hornetq.jms.client.SelectorTranslator; import org.hornetq.rest.queue.push.xml.PushRegistration; /** @@ -68,7 +69,14 @@ strategy.start(); session = factory.createSession(false, false); - consumer = session.createConsumer(destination); + if (registration.getSelector() != null) + { + consumer = session.createConsumer(destination, SelectorTranslator.convertToHornetQFilterString(registration.getSelector())); + } + else + { + consumer = session.createConsumer(destination); + } consumer.setMessageHandler(this); session.start(); log.info("Push consumer started for: " + registration.getTarget()); @@ -100,6 +108,7 @@ } } + @Override public void onMessage(ClientMessage clientMessage) { if (strategy.push(clientMessage) == false) Modified: trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/push/xml/PushRegistration.java =================================================================== --- trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/push/xml/PushRegistration.java 2010-10-25 15:49:13 UTC (rev 9811) +++ trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/push/xml/PushRegistration.java 2010-10-25 19:35:10 UTC (rev 9812) @@ -18,7 +18,7 @@ */ @XmlRootElement(name = "push-registration") @XmlAccessorType(XmlAccessType.PROPERTY) -@XmlType(propOrder = {"destination", "durable", "target", "authenticationMechanism", "headers"}) +@XmlType(propOrder = {"destination", "durable", "selector", "target", "authenticationMechanism", "headers"}) public class PushRegistration implements Serializable { private String id; @@ -28,6 +28,7 @@ private List<XmlHttpHeader> headers = new ArrayList<XmlHttpHeader>(); private String destination; private Object loadedFrom; + private String selector; @XmlTransient public Object getLoadedFrom() @@ -73,6 +74,16 @@ this.durable = durable; } + public String getSelector() + { + return selector; + } + + public void setSelector(String selector) + { + this.selector = selector; + } + @XmlElementRef public XmlLink getTarget() { Modified: trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/topic/AcknowledgedSubscriptionResource.java =================================================================== --- trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/topic/AcknowledgedSubscriptionResource.java 2010-10-25 15:49:13 UTC (rev 9811) +++ trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/topic/AcknowledgedSubscriptionResource.java 2010-10-25 19:35:10 UTC (rev 9812) @@ -13,10 +13,10 @@ { private boolean durable; - public AcknowledgedSubscriptionResource(ClientSessionFactory factory, String destination, String id, DestinationServiceManager serviceManager) + public AcknowledgedSubscriptionResource(ClientSessionFactory factory, String destination, String id, DestinationServiceManager serviceManager, String selector) throws HornetQException { - super(factory, destination, id, serviceManager); + super(factory, destination, id, serviceManager, selector); } public boolean isDurable() Modified: trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/topic/FileTopicPushStore.java =================================================================== --- trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/topic/FileTopicPushStore.java 2010-10-25 15:49:13 UTC (rev 9811) +++ trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/topic/FileTopicPushStore.java 2010-10-25 19:35:10 UTC (rev 9812) @@ -18,6 +18,7 @@ super(dirname); } + @Override public synchronized List<PushTopicRegistration> getByTopic(String topic) { List<PushTopicRegistration> list = new ArrayList<PushTopicRegistration>(); Modified: trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/topic/SubscriptionResource.java =================================================================== --- trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/topic/SubscriptionResource.java 2010-10-25 15:49:13 UTC (rev 9811) +++ trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/topic/SubscriptionResource.java 2010-10-25 19:35:10 UTC (rev 9812) @@ -13,10 +13,10 @@ { boolean durable; - public SubscriptionResource(ClientSessionFactory factory, String destination, String id, DestinationServiceManager serviceManager) + public SubscriptionResource(ClientSessionFactory factory, String destination, String id, DestinationServiceManager serviceManager, String selector) throws HornetQException { - super(factory, destination, id, serviceManager); + super(factory, destination, id, serviceManager, selector); } public boolean isDurable() Modified: trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/topic/SubscriptionsResource.java =================================================================== --- trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/topic/SubscriptionsResource.java 2010-10-25 15:49:13 UTC (rev 9811) +++ trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/topic/SubscriptionsResource.java 2010-10-25 19:35:10 UTC (rev 9812) @@ -84,6 +84,7 @@ private Object timeoutLock = new Object(); + @Override public void testTimeout(String target) { synchronized (timeoutLock) @@ -127,6 +128,7 @@ public Response createSubscription(@FormParam("durable") @DefaultValue("false") boolean durable, @FormParam("autoAck") @DefaultValue("true") boolean autoAck, @FormParam("name") String subscriptionName, + @FormParam("selector") String selector, @Context UriInfo uriInfo) { if (subscriptionName != null) @@ -185,7 +187,7 @@ session.createTemporaryQueue(destination, subscriptionName); } } - QueueConsumer consumer = createConsumer(durable, autoAck, subscriptionName); + QueueConsumer consumer = createConsumer(durable, autoAck, subscriptionName, selector); queueConsumers.put(consumer.getId(), consumer); serviceManager.getTimeoutTask().add(this, consumer.getId()); @@ -225,19 +227,19 @@ } } - protected QueueConsumer createConsumer(boolean durable, boolean autoAck, String subscriptionName) + protected QueueConsumer createConsumer(boolean durable, boolean autoAck, String subscriptionName, String selector) throws HornetQException { QueueConsumer consumer; if (autoAck) { - SubscriptionResource subscription = new SubscriptionResource(sessionFactory, subscriptionName, subscriptionName, serviceManager); + SubscriptionResource subscription = new SubscriptionResource(sessionFactory, subscriptionName, subscriptionName, serviceManager, selector); subscription.setDurable(durable); consumer = subscription; } else { - AcknowledgedSubscriptionResource subscription = new AcknowledgedSubscriptionResource(sessionFactory, subscriptionName, subscriptionName, serviceManager); + AcknowledgedSubscriptionResource subscription = new AcknowledgedSubscriptionResource(sessionFactory, subscriptionName, subscriptionName, serviceManager, selector); subscription.setDurable(durable); consumer = subscription; } @@ -375,7 +377,7 @@ QueueConsumer tmp = null; try { - tmp = createConsumer(true, autoAck, subscriptionId); + tmp = createConsumer(true, autoAck, subscriptionId, null); } catch (HornetQException e) { Modified: trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/util/HttpMessageHelper.java =================================================================== --- trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/util/HttpMessageHelper.java 2010-10-25 15:49:13 UTC (rev 9811) +++ trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/util/HttpMessageHelper.java 2010-10-25 19:35:10 UTC (rev 9812) @@ -81,17 +81,20 @@ int size = message.getBodySize(); if (size > 0) { - byte[] body = new byte[size]; - message.getBodyBuffer().readBytes(body); Boolean aBoolean = message.getBooleanProperty(POSTED_AS_HTTP_MESSAGE); if (aBoolean != null && aBoolean.booleanValue()) { + byte[] body = new byte[size]; + message.getBodyBuffer().readBytes(body); //System.out.println("Building Message from HTTP message"); request.body(contentType, body); } else { // assume posted as a JMS or HornetQ object message + size = message.getBodyBuffer().readInt(); + byte[] body = new byte[size]; + message.getBodyBuffer().readBytes(body); ByteArrayInputStream bais = new ByteArrayInputStream(body); Object obj = null; try Added: trunk/hornetq-rest/hornetq-rest/src/test/java/org/hornetq/rest/test/SelectorTest.java =================================================================== --- trunk/hornetq-rest/hornetq-rest/src/test/java/org/hornetq/rest/test/SelectorTest.java (rev 0) +++ trunk/hornetq-rest/hornetq-rest/src/test/java/org/hornetq/rest/test/SelectorTest.java 2010-10-25 19:35:10 UTC (rev 9812) @@ -0,0 +1,304 @@ +package org.hornetq.rest.test; + +import org.hornetq.jms.client.HornetQConnectionFactory; +import org.hornetq.jms.client.HornetQDestination; +import org.hornetq.jms.client.HornetQJMSConnectionFactory; +import org.hornetq.rest.HttpHeaderProperty; +import org.hornetq.rest.queue.push.xml.XmlLink; +import org.hornetq.rest.topic.PushTopicRegistration; +import org.hornetq.rest.topic.TopicDeployment; +import org.jboss.resteasy.client.ClientRequest; +import org.jboss.resteasy.client.ClientResponse; +import org.jboss.resteasy.spi.Link; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; + +import javax.jms.Connection; +import javax.jms.ConnectionFactory; +import javax.jms.Destination; +import javax.jms.MessageProducer; +import javax.jms.ObjectMessage; +import javax.jms.Session; +import javax.ws.rs.POST; +import javax.ws.rs.Path; +import javax.xml.bind.annotation.XmlRootElement; +import java.io.Serializable; + +import static org.jboss.resteasy.test.TestPortProvider.*; + +/** + * @author <a href="mailto:bill@burkecentral.com">Bill Burke</a> + * @version $Revision: 1 $ + */ +public class SelectorTest extends MessageTestBase +{ + public static ConnectionFactory connectionFactory; + public static String topicName = HornetQDestination.createQueueAddressFromName("testTopic").toString(); + + @BeforeClass + public static void setup() throws Exception + { + connectionFactory = new HornetQJMSConnectionFactory(manager.getQueueManager().getSessionFactory()); + System.out.println("Queue name: " + topicName); + TopicDeployment deployment = new TopicDeployment(); + deployment.setDuplicatesAllowed(true); + deployment.setDurableSend(false); + deployment.setName(topicName); + manager.getTopicManager().deploy(deployment); + } + + @XmlRootElement + public static class Order implements Serializable + { + private String name; + private String amount; + + public String getName() + { + return name; + } + + public void setName(String name) + { + this.name = name; + } + + public String getAmount() + { + return amount; + } + + public void setAmount(String amount) + { + this.amount = amount; + } + + @Override + public boolean equals(Object o) + { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + Order order = (Order) o; + + if (!amount.equals(order.amount)) return false; + if (!name.equals(order.name)) return false; + + return true; + } + + @Override + public String toString() + { + return "Order{" + + "name='" + name + '\'' + + '}'; + } + + @Override + public int hashCode() + { + int result = name.hashCode(); + result = 31 * result + amount.hashCode(); + return result; + } + } + + public static Destination createDestination(String dest) + { + HornetQDestination destination = (HornetQDestination) HornetQDestination.fromAddress(dest); + System.out.println("SimpleAddress: " + destination.getSimpleAddress()); + return destination; + } + + public static void publish(String dest, Serializable object, String contentType, String tag) throws Exception + { + Connection conn = connectionFactory.createConnection(); + try + { + Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); + Destination destination = createDestination(dest); + MessageProducer producer = session.createProducer(destination); + ObjectMessage message = session.createObjectMessage(); + + if (contentType != null) + { + message.setStringProperty(HttpHeaderProperty.CONTENT_TYPE, contentType); + } + if (tag != null) + { + message.setStringProperty("MyTag", tag); + } + message.setObject(object); + + producer.send(message); + } + finally + { + conn.close(); + } + } + + @Path("/push") + public static class PushReceiver + { + public static Order oneOrder; + public static Order twoOrder; + + @POST + @Path("one") + public void one(Order order) + { + oneOrder = order; + } + + @POST + @Path("two") + public void two(Order order) + { + twoOrder = order; + } + + + } + + @Test + public void testPush() throws Exception + { + server.getJaxrsServer().getDeployment().getRegistry().addPerRequestResource(PushReceiver.class); + ClientRequest request = new ClientRequest(generateURL("/topics/" + topicName)); + + ClientResponse response = request.head(); + Assert.assertEquals(200, response.getStatus()); + Link consumers = MessageTestBase.getLinkByTitle(manager.getQueueManager().getLinkStrategy(), response, "push-subscriptions"); + System.out.println("push: " + consumers); + + PushTopicRegistration oneReg = new PushTopicRegistration(); + oneReg.setDurable(false); + XmlLink target = new XmlLink(); + target.setMethod("post"); + target.setHref(generateURL("/push/one")); + target.setType("application/xml"); + oneReg.setTarget(target); + oneReg.setSelector("MyTag = '1'"); + response = consumers.request().body("application/xml", oneReg).post(); + Link oneSubscription = response.getLocation(); + + PushTopicRegistration twoReg = new PushTopicRegistration(); + twoReg.setDurable(false); + target = new XmlLink(); + target.setMethod("post"); + target.setHref(generateURL("/push/two")); + target.setType("application/xml"); + twoReg.setTarget(target); + twoReg.setSelector("MyTag = '2'"); + response = consumers.request().body("application/xml", twoReg).post(); + Link twoSubscription = response.getLocation(); + + Order order = new Order(); + order.setName("1"); + order.setAmount("$5.00"); + publish(topicName, order, null, "1"); + Thread.sleep(200); + Assert.assertEquals(order, PushReceiver.oneOrder); + + order.setName("2"); + publish(topicName, order, null, "2"); + Thread.sleep(200); + Assert.assertEquals(order, PushReceiver.twoOrder); + + order.setName("3"); + publish(topicName, order, null, "2"); + Thread.sleep(200); + Assert.assertEquals(order, PushReceiver.twoOrder); + + order.setName("4"); + publish(topicName, order, null, "1"); + Thread.sleep(200); + Assert.assertEquals(order, PushReceiver.oneOrder); + + order.setName("5"); + publish(topicName, order, null, "1"); + Thread.sleep(200); + Assert.assertEquals(order, PushReceiver.oneOrder); + + order.setName("6"); + publish(topicName, order, null, "1"); + Thread.sleep(200); + Assert.assertEquals(order, PushReceiver.oneOrder); + + oneSubscription.request().delete(); + twoSubscription.request().delete(); + + + } + + + @Test + public void testPull() throws Exception + { + ClientRequest request = new ClientRequest(generateURL("/topics/" + topicName)); + + ClientResponse response = request.head(); + Assert.assertEquals(200, response.getStatus()); + Link consumers = MessageTestBase.getLinkByTitle(manager.getQueueManager().getLinkStrategy(), response, "pull-subscriptions"); + System.out.println("pull: " + consumers); + response = consumers.request().formParameter("autoAck", "true") + .formParameter("selector", "MyTag = '1'").post(); + + Link consumeOne = MessageTestBase.getLinkByTitle(manager.getQueueManager().getLinkStrategy(), response, "consume-next"); + System.out.println("consumeOne: " + consumeOne); + response = consumers.request().formParameter("autoAck", "true") + .formParameter("selector", "MyTag = '2'").post(); + Link consumeTwo = MessageTestBase.getLinkByTitle(manager.getQueueManager().getLinkStrategy(), response, "consume-next"); + System.out.println("consumeTwo: " + consumeTwo); + + + // test that Accept header is used to set content-type + { + Order order = new Order(); + order.setName("1"); + order.setAmount("$5.00"); + publish(topicName, order, null, "1"); + order.setName("2"); + publish(topicName, order, null, "2"); + order.setName("3"); + publish(topicName, order, null, "2"); + order.setName("4"); + publish(topicName, order, null, "1"); + order.setName("5"); + publish(topicName, order, null, "1"); + order.setName("6"); + publish(topicName, order, null, "1"); + + { + order.setName("1"); + consumeOne = consumeOrder(order, consumeOne); + order.setName("2"); + consumeTwo = consumeOrder(order, consumeTwo); + order.setName("3"); + consumeTwo = consumeOrder(order, consumeTwo); + order.setName("4"); + consumeOne = consumeOrder(order, consumeOne); + order.setName("5"); + consumeOne = consumeOrder(order, consumeOne); + order.setName("6"); + consumeOne = consumeOrder(order, consumeOne); + } + } + } + + private Link consumeOrder(Order order, Link consumeNext) + throws Exception + { + ClientResponse res = consumeNext.request().header("Accept-Wait", "4").accept("application/xml").post(String.class); + Assert.assertEquals(200, res.getStatus()); + Assert.assertEquals("application/xml", res.getHeaders().getFirst("Content-Type").toString().toLowerCase()); + Order order2 = (Order) res.getEntity(Order.class); + Assert.assertEquals(order, order2); + consumeNext = MessageTestBase.getLinkByTitle(manager.getQueueManager().getLinkStrategy(), res, "consume-next"); + Assert.assertNotNull(consumeNext); + return consumeNext; + } +} \ No newline at end of file