URI format for AMQP topic - help required

[gairey]

I'm using Apache Camel with Spring to handle consuming / publishing messages with the Apache Qpid Java message broker. My Spring configuration file contains:

<camel:camelContext id="camel">
  <camel:route>
    <camel:from uri="amqp:test-queue"/>
    <camel:pipeline>
      <camel:bean ref="myBean"/>
      <camel:to uri="amqp:amq.topic"/>
    </camel:pipeline>
  </camel:route>
</camel:camelContext>

<bean id="amqp" class="org.apache.camel.component.amqp.AMQPComponent">
  <property name="connectionFactory">
    <bean class="org.apache.qpid.client.AMQConnectionFactory">
      <constructor-arg index="0" type="java.lang.String" value="amqp://guest:guest@/localhost?brokerlist='tcp://localhost:5672'" />
    </bean>
  </property>
</bean>

My app receives messages from queue test-queue OK, and they're forwarded to myBean too. However, I don't know how to format the URI in the <camel:to ...> node above to post a message to the topic exchange amq.topic.

Using the RabbitMQ Java client, I can post a message with the code:

Channel channel = ...;
channel.exchangeDeclare("amq.topic", "topic");
channel.queueDeclare("test-other");
channel.queueBind("test-other", "amq.topic", "TEST");
channel.basicPublish("amq.topic", "TEST", null, messageBytes);

i.e. I post a message to a named exchange specifying a particular routing key ("TEST" in this case).

How do I specify this information within the URI of the <camel:to ...> node?

Any help would be very much appreciated - thank you.

[Ashwin Karpe]

Hi,

In order to post to a Topic you will need to do the following

        <camel:to uri="amqp:topic:amq.topic"/>

Similarly for a queue it is

        <camel:to uri="amqp:queue:amq.queue"/>

Cheers,

Ashwin...

gairey wrote:

I'm using Apache Camel with Spring to handle consuming / publishing messages with the Apache Qpid Java message broker. My Spring configuration file contains:

<camel:camelContext id="camel">
  <camel:route>
    <camel:from uri="amqp:test-queue"/>
    <camel:pipeline>
      <camel:bean ref="myBean"/>
      <camel:to uri="amqp:amq.topic"/>
    </camel:pipeline>
  </camel:route>
</camel:camelContext>

<bean id="amqp" class="org.apache.camel.component.amqp.AMQPComponent">
  <property name="connectionFactory">
    <bean class="org.apache.qpid.client.AMQConnectionFactory">
      <constructor-arg index="0" type="java.lang.String" value="amqp://guest:guest@/localhost?brokerlist='tcp://localhost:5672'" />
    </bean>
  </property>
</bean>

My app receives messages from queue test-queue OK, and they're forwarded to myBean too. However, I don't know how to format the URI in the <camel:to ...> node above to post a message to the topic exchange amq.topic.

Using the RabbitMQ Java client, I can post a message with the code:

Channel channel = ...;
channel.exchangeDeclare("amq.topic", "topic");
channel.queueDeclare("test-other");
channel.queueBind("test-other", "amq.topic", "TEST");
channel.basicPublish("amq.topic", "TEST", null, messageBytes);

i.e. I post a message to a named exchange specifying a particular routing key ("TEST" in this case).

How do I specify this information within the URI of the <camel:to ...> node?

Any help would be very much appreciated - thank you.

[willem.jiang]

Since AMQPComponent extends from JmsComponent, the JMS URI format[1]
should be reusable in AMQP component.

[1] http://camel.apache.org/jms.html

Willem

gairey wrote:

> I'm using Apache Camel with Spring to handle consuming / publishing messages
> with the Apache Qpid Java message broker. My Spring configuration file
> contains:
>
> <camel:camelContext id="camel">
>   <camel:route>
>     <camel:from uri="amqp:test-queue"/>
>     <camel:pipeline>
>       <camel:bean ref="myBean"/>
>       <camel:to uri="amqp:amq.topic"/>
>     </camel:pipeline>
>   </camel:route>
> </camel:camelContext>
>
> <bean id="amqp" class="org.apache.camel.component.amqp.AMQPComponent">
>   <property name="connectionFactory">
>     <bean class="org.apache.qpid.client.AMQConnectionFactory">
>       <constructor-arg index="0" type="java.lang.String"
> value="amqp://guest:guest@/localhost?brokerlist='tcp://localhost:5672'" />
>     </bean>
>   </property>
> </bean>
>
> My app receives messages from queue test-queue OK, and they're forwarded to
> myBean too. However, I don't know how to format the URI in the <camel:to
> ...> node above to post a message to the topic exchange amq.topic.
>
> Using the RabbitMQ Java client, I can post a message with the code:
>
> Channel channel = ...;
> channel.exchangeDeclare("amq.topic", "topic");
> channel.queueDeclare("test-other");
> channel.queueBind("test-other", "amq.topic", "TEST");
> channel.basicPublish("amq.topic", "TEST", null, messageBytes);
>
> i.e. I post a message to a named exchange specifying a particular routing
> key ("TEST" in this case).
>
> How do I specify this information within the URI of the <camel:to ...> node?
>
> Any help would be very much appreciated - thank you.

[gairey]

Ashwin, Willem,

Thanks for your prompt replies. I'm still having some trouble though. From my limited experience of using Qpid, it appears that when you consume messages, you specify a queue/topic name, whereas when you publish messages, you specify an exchange name and a routing key.

So, for the node:

  <camel:to uri="amqp:topic:destinationName"/>

I presume destinationName should be an exchange name? There doesn't seem to be a way of specifying the routing key though.

I've tried running my app with the above setting (i.e. destinationName set to the exchange amq.topic), but no messages are received by a JMS listener in a separate app, configured as follows (again using Spring):

<bean id="connectionFactory" class="org.apache.qpid.client.AMQConnectionFactory">
  <constructor-arg index="0" type="java.lang.String" value="amqp://guest:guest@/localhost?brokerlist='tcp://localhost:5672'" />
</bean>

<bean id="destination" class="org.apache.qpid.client.AMQTopic">
  <constructor-arg index="0">
    <bean class="org.apache.qpid.framing.AMQShortString">
      <constructor-arg index="0" type="java.lang.String" value="amq.topic" />
    </bean>
  </constructor-arg>
  <constructor-arg index="1" type="java.lang.String" value="TEST" />
</bean>

<bean id="jmsConnection" class="org.springframework.jms.core.JmsTemplate" scope="prototype">
  <property name="connectionFactory" ref="blazeConnectionFactory"/>
  <property name="defaultDestination" ref="blazeDestination"/>
  <property name="receiveTimeout" value="60000"/>
</bean>

If I publish a message directly using the RabbitMQ client code in my previous post, the JMS listener does receive the message, so I'm pretty sure it's configured OK.

Any ideas? Thanks again for your help so far - I'm pretty new to all this stuff, so really appreciate your time.

[willem.jiang]

Hi,

I think you just need set the destinationName with topic queue's name.
Camel will create a exchage and set the correlation id (routing key) for
you.

Willem

gairey wrote:

> Ashwin, Willem,
>
> Thanks for your prompt replies. I'm still having some trouble though. From
> my limited experience of using Qpid, it appears that when you consume
> messages, you specify a queue/topic name, whereas when you publish messages,
> you specify an exchange name and a routing key.
>
> So, for the node:
>
>   <camel:to uri="amqp:topic:destinationName"/>
>
> I presume destinationName should be an exchange name? There doesn't seem to
> be a way of specifying the routing key though.
>
> I've tried running my app with the above setting (i.e. destinationName set
> to the exchange amq.topic), but no messages are received by a JMS listener
> in a separate app, configured as follows (again using Spring):
>
> <bean id="connectionFactory"
> class="org.apache.qpid.client.AMQConnectionFactory">
>   <constructor-arg index="0" type="java.lang.String"
> value="amqp://guest:guest@/localhost?brokerlist='tcp://localhost:5672'" />
> </bean>
>
> <bean id="destination" class="org.apache.qpid.client.AMQTopic">
>   <constructor-arg index="0">
>     <bean class="org.apache.qpid.framing.AMQShortString">
>       <constructor-arg index="0" type="java.lang.String" value="amq.topic"
> />
>     </bean>
>   </constructor-arg>
>   <constructor-arg index="1" type="java.lang.String" value="TEST" />
> </bean>
>
> <bean id="jmsConnection" class="org.springframework.jms.core.JmsTemplate"
> scope="prototype">
>   <property name="connectionFactory" ref="blazeConnectionFactory"/>
>   <property name="defaultDestination" ref="blazeDestination"/>
>   <property name="receiveTimeout" value="60000"/>
> </bean>
>
> If I publish a message directly using the RabbitMQ client code in my
> previous post, the JMS listener does receive the message, so I'm pretty sure
> it's configured OK.
>
> Any ideas? Thanks again for your help so far - I'm pretty new to all this
> stuff, so really appreciate your time.

[javaty]

Hi Willem,

You are right that camel automatically set the exchange and routingkey, but it also sets the flag to autodelete=true, kindly give some guidance how we can pass parameters to camel to send persistent messages to Qpid.

Kind Regards

willem.jiang wrote:

Hi,

I think you just need set the destinationName with topic queue's name.
Camel will create a exchage and set the correlation id (routing key) for
you.

Willem

gairey wrote:
> Ashwin, Willem,
>
> Thanks for your prompt replies. I'm still having some trouble though. From
> my limited experience of using Qpid, it appears that when you consume
> messages, you specify a queue/topic name, whereas when you publish messages,
> you specify an exchange name and a routing key.
>
> So, for the node:
>
>   <camel:to uri="amqp:topic:destinationName"/>
>
> I presume destinationName should be an exchange name? There doesn't seem to
> be a way of specifying the routing key though.
>
> I've tried running my app with the above setting (i.e. destinationName set
> to the exchange amq.topic), but no messages are received by a JMS listener
> in a separate app, configured as follows (again using Spring):
>
> <bean id="connectionFactory"
> class="org.apache.qpid.client.AMQConnectionFactory">
>   <constructor-arg index="0" type="java.lang.String"
> value="amqp://guest:guest@/localhost?brokerlist='tcp://localhost:5672'" />
> </bean>
>
> <bean id="destination" class="org.apache.qpid.client.AMQTopic">
>   <constructor-arg index="0">
>     <bean class="org.apache.qpid.framing.AMQShortString">
>       <constructor-arg index="0" type="java.lang.String" value="amq.topic"
> />
>     </bean>
>   </constructor-arg>
>   <constructor-arg index="1" type="java.lang.String" value="TEST" />
> </bean>
>
> <bean id="jmsConnection" class="org.springframework.jms.core.JmsTemplate"
> scope="prototype">
>   <property name="connectionFactory" ref="blazeConnectionFactory"/>
>   <property name="defaultDestination" ref="blazeDestination"/>
>   <property name="receiveTimeout" value="60000"/>
> </bean>
>
> If I publish a message directly using the RabbitMQ client code in my
> previous post, the JMS listener does receive the message, so I'm pretty sure
> it's configured OK.
>
> Any ideas? Thanks again for your help so far - I'm pretty new to all this
> stuff, so really appreciate your time.