[hornetq-commits] JBoss hornetq SVN: r9812 - in trunk: hornetq-rest/docbook/reference/en and 6 other directories.
do-not-reply at jboss.org
do-not-reply at jboss.org
Mon Oct 25 15:35:11 EDT 2010
Author: bill.burke at 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><?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><push-registration>
<durable>false</durable>
+ <selector><![CDATA[
+ SomeAttribute > 1
+ ]]>
+ </selector>
<link rel="push" href="http://somewhere.com" type="application/json" method="PUT"/>
</push-registration>
</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><push-topic-registration>
<durable>true</durable>
+ <selector><![CDATA[
+ SomeAttribute > 1
+ ]]>
+ </selector>
<link rel="template" href="http://somewhere.com/resources/{id}/messages" method="POST"/>
</push-topic registration></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)
- at XmlType(propOrder = {"destination", "durable", "target", "authenticationMechanism", "headers"})
+ at 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 at 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
More information about the hornetq-commits
mailing list