JMS Message - Spring integration

[curiousmind]

Hi,

I was trying to send messages from one queue to another queue. To achieve this functionality, I have done following configuration.


<jee:jndi-lookup jndi-name="ConnectionFactory" id="connectionFactory"
environment-ref="jndiProps" expected-type="javax.jms.ConnectionFactory" />

<jee:jndi-lookup jndi-name="queue/A" id="jBossQueue"
environment-ref="jndiProps" expected-type="javax.jms.Queue" />

<jee:jndi-lookup jndi-name="queue/B" id="jBossQueueTarget"
environment-ref="jndiProps" expected-type="javax.jms.Queue" />

<util:properties id="jndiProps" location="classpath:jndi.properties"/>

<jms:channel id="jmsIn" queue="jBossQueue"/>
<jms:channel id="jmsOut" queue="jBossQueueTarget"/>

<integration:bridge output-channel="jmsOut" input-channel="jmsIn" />


I am sending message using sample servlet code as


Queue queue = (Queue)ctx.lookup("queue/A");
QueueSession queueSession = queueCon.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
QueueSender queueSender = queueSession.createSender(queue);

TextMessage textMessage = queueSession.createTextMessage();
textMessage.setText("A critical Message.");
queueSender.send(textMessage);


I have created a listener to listen on "jBossQueueTarget". Following is the code snippet

try{
TextMessage textMessage = (TextMessage)message;
System.out.println("Message received: " + textMessage.getText());
}catch(Exception e)
{
System.out.println("Exception occurred while handling message.");
}

But I am getting classcastexception in above code.
Can someone tell me where i am making mistake or is there any way to achieve above functionality.

[chudak]

print the stack trace of the exception in your catch block and post it.

[Mark Fisher]

Indeed. It's necessary to see exactly where the ClassCastException occurs before we'd be able to address the problem.

[curiousmind]

I dont know what happenned suddenly. Using same code, I am getting different exception at this time.



23:31:06,399 WARN [DefaultMessageListenerContainer] Execution of JMS message listener failed, and no ErrorHandler has been set.
org.springframework.integration.core.MessagingExce ption: failed to handle incoming JMS Message
at org.springframework.integration.jms.JmsDestination BackedMessageChannel.onMessage(JmsDestinationBacke dMessageChannel.java:157)
at org.springframework.jms.listener.AbstractMessageLi stenerContainer.doInvokeListener(AbstractMessageLi stenerContainer.java:559)
at org.springframework.jms.listener.AbstractMessageLi stenerContainer.invokeListener(AbstractMessageList enerContainer.java:498)
at org.springframework.jms.listener.AbstractMessageLi stenerContainer.doExecuteListener(AbstractMessageL istenerContainer.java:467)
at org.springframework.jms.listener.AbstractPollingMe ssageListenerContainer.doReceiveAndExecute(Abstrac tPollingMessageListenerContainer.java:323)
at org.springframework.jms.listener.AbstractPollingMe ssageListenerContainer.receiveAndExecute(AbstractP ollingMessageListenerContainer.java:261)
at org.springframework.jms.listener.DefaultMessageLis tenerContainer$AsyncMessageListenerInvoker.invokeL istener(DefaultMessageListenerContainer.java:976)
at org.springframework.jms.listener.DefaultMessageLis tenerContainer$AsyncMessageListenerInvoker.execute OngoingLoop(DefaultMessageListenerContainer.java:9 68)
at org.springframework.jms.listener.DefaultMessageLis tenerContainer$AsyncMessageListenerInvoker.run(Def aultMessageListenerContainer.java:870)
at java.lang.Thread.run(Unknown Source)
Caused by: org.springframework.integration.message.MessageDel iveryException: Dispatcher has no subscribers.
at org.springframework.integration.dispatcher.Unicast ingDispatcher.doDispatch(UnicastingDispatcher.java :97)
at org.springframework.integration.dispatcher.Unicast ingDispatcher.dispatch(UnicastingDispatcher.java:9 0)
at org.springframework.integration.jms.JmsDestination BackedMessageChannel.onMessage(JmsDestinationBacke dMessageChannel.java:154)
... 9 more

[curiousmind]

I dont know whats going on, I restarted JBoss again and now i am getting following exception


23:57:57,696 ERROR [STDERR] java.lang.ClassCastException: org.jboss.mq.SpyObjectMessage cannot be cast to javax.jms.TextMessage
23:57:57,696 ERROR [STDERR] at com.spring.integration.mq.QueueListener.onMessage( QueueListener.java:21)
23:57:57,696 ERROR [STDERR] at org.springframework.jms.listener.AbstractMessageLi stenerContainer.doInvokeListener(AbstractMessageLi stenerContainer.java:559)
23:57:57,696 ERROR [STDERR] at org.springframework.jms.listener.AbstractMessageLi stenerContainer.invokeListener(AbstractMessageList enerContainer.java:498)
23:57:57,696 ERROR [STDERR] at org.springframework.jms.listener.AbstractMessageLi stenerContainer.doExecuteListener(AbstractMessageL istenerContainer.java:467)
23:57:57,696 ERROR [STDERR] at org.springframework.jms.listener.AbstractPollingMe ssageListenerContainer.doReceiveAndExecute(Abstrac tPollingMessageListenerContainer.java:323)
23:57:57,696 ERROR [STDERR] at org.springframework.jms.listener.AbstractPollingMe ssageListenerContainer.receiveAndExecute(AbstractP ollingMessageListenerContainer.java:261)
23:57:57,696 ERROR [STDERR] at org.springframework.jms.listener.DefaultMessageLis tenerContainer$AsyncMessageListenerInvoker.invokeL istener(DefaultMessageListenerContainer.java:976)
23:57:57,696 ERROR [STDERR] at org.springframework.jms.listener.DefaultMessageLis tenerContainer$AsyncMessageListenerInvoker.execute OngoingLoop(DefaultMessageListenerContainer.java:9 68)
23:57:57,696 ERROR [STDERR] at org.springframework.jms.listener.DefaultMessageLis tenerContainer$AsyncMessageListenerInvoker.run(Def aultMessageListenerContainer.java:870)
23:57:57,696 ERROR [STDERR] at java.lang.Thread.run(Unknown Source)

The exception is encountered in QueueListener class created in above post.

[curiousmind]

Problem seems to be very weired, each time i start/stop jboss I am getting either of the above exception.
I am using Spring Integration 2.0.0 M and SPRING 3.
can someone please let me know whats going on? why i am getting these random exceptions?
Also how can I achieve the task, listed above?

[iwein]

You should probably not use jms:channel, but the jms gateways or channel adapters for this. I don't know what JBoss is doing, but the dispatcher has no subscribers error happens if you send a message to a direct channel that doesn't have subscribers.

And yes, the error message could be better, there's an issue open for that.

[Mark Fisher]

The ClassCastException that you are seeing indicates that the received JMS Message is not a TextMessage... it's a JBoss "SpyMessage". I'm not sure if that's what you are intending to send, but if so, you might want to consider a custom MessageConverter implementation to deal with that type.

[curiousmind]

Hi iwein,

Can you please tell me why I shouldn't use jms:channel here? can you please provide more details about same, it would help me to understand spring integration better.
Also I am sending message to Queue and not to Channel directly.
How can I achieve the same functionlity by using gateways or channel adapters, my primary intention is not to write any java code but getting this task done with just configuration.

Well now I am getting this error consistently, not sure why
org.springframework.integration.message.MessageDel iveryException: Dispatcher has no subscribers.

[curiousmind]

Can someone please help me to figure out where is the problem?

[Mark Fisher]

The <jms:channel/> is really intended for use where the producer and consumer are both Spring Integration components. For one-way interaction, you can use the <jms:message-driven-channel-adapter/> and <jms: outbound-channel-adapter/>. Any JMS producer could send to that inbound (message-driven) channel-adapter's Destination, and the Spring Integration components subscribed to its channel would receive the converted Message. For outbound, it's also uni-directional but in the other direction (from a Spring Integration component to a JMS Destination).

What is your ultimate goal here? I guess you are using that bridge as a first step prior to adding some other component between the JMS Destinations?

[curiousmind]

Well mark, thanks for your reply. As explained earlier, I want to take a message from one queue as and when it arrives and pass it to another queue.
Isnt that possible without writing a single line of code and just with the configuration?
Well the reason behind that is, instead of using SPRING integration what if I just write a message listener bean and then as and when message comes the bean will just write that message to another queue.
Now the question is why I should use SPRING integration in such scenario, if I will have to write Java code. I have taken very basic example here.

[Mark Fisher]

Okay. For that use-case, you should use the inbound/outbound adapter pair. If you are not planning to add any components between them, you don't need a bridge as long as they share the same channel reference. The jms:channel on the other hand is simply for replacing a "channel" in Spring Integration with one that is backed by a JMS Queue (to be used when the producer and consumer are both Spring Integration components).

[curiousmind]

Well finally, I could able to achieve the desired functionality, but now there is one more problem. Have a look at the following configuration


<jee:jndi-lookup jndi-name="ConnectionFactory" id="connectionFactory"
environment-ref="activeMqProps" expected-type="javax.jms.ConnectionFactory" />

<jee:jndi-lookup jndi-name="ConnectionFactory" id="jbossConnectionFactory"
environment-ref="jndiJbossProps" expected-type="javax.jms.ConnectionFactory" />

<jee:jndi-lookup jndi-name="MyQueue" id="activeMqQueue"
environment-ref="activeMqProps" expected-type="javax.jms.Queue" />

<jee:jndi-lookup jndi-name="queue/B" id="jBossQueueTarget"
environment-ref="jndiJbossProps" expected-type="javax.jms.Queue" />

<util:properties id="activeMqProps" location="classpath:jndi.properties"/>
<util:properties id="jndiJbossProps" location="classpath:jndi-jboss.properties"/>

<jms:inbound-channel-adapter destination="activeMqQueue" channel="jbossIn" >
<integration:poller>
<integration:interval-trigger interval="1000"/>
</integration:poller>
</jms:inbound-channel-adapter>

<jms:outbound-channel-adapter destination="jBossQueueTarget" channel="jbossIn">
</jms:outbound-channel-adapter>

<integration:channel id="jbossIn" />


When I send a message from Servlet to ActiveMQ - MyQueue, it dynamically creates queue.B in Active MQ.
While sometimes the message is sent to Jboss Queue "queue/B" and sometimes the message is sent to Active MQ - "queue.B"
Can someone please tell me why this is happenning, why the dynamic queue is getting created in Active MQ? why messages are redirected to Active MQ rather than Jboss Queue?

[Mark Fisher]

If you don't explicitly provide a "connection-factory" attribute on those channel adapters, it will use the one named "connectionFactory". In your case, that means they should always be using the ActiveMQ ConnectionFactory rather than the JBoss one.

[curiousmind]

Hi Mark,

I tried setting the connection factory property to appropriate values. but there was no difference. It continued to create queue on Active mq and redirect messages to Active MQ queue.

[Mark Fisher]

Can you show the new revised configuration?

[curiousmind]

Hi Mark,

Sorry I forgot to post it in previous post.
Here is the new configuration



<jee:jndi-lookup jndi-name="ConnectionFactory" id="connectionFactory"
environment-ref="activeMqProps" expected-type="javax.jms.ConnectionFactory" />

<jee:jndi-lookup jndi-name="ConnectionFactory" id="jbossConnectionFactory"
environment-ref="jndiJbossProps" expected-type="javax.jms.ConnectionFactory" />

<jee:jndi-lookup jndi-name="MyQueue" id="activeMqQueue"
environment-ref="activeMqProps" expected-type="javax.jms.Queue" />

<jee:jndi-lookup jndi-name="queue/B" id="jBossQueueTarget"
environment-ref="jndiJbossProps" expected-type="javax.jms.Queue" />

<util:properties id="activeMqProps" location="classpath:jndi.properties"/>
<util:properties id="jndiJbossProps" location="classpath:jndi-jboss.properties"/>

<jms:inbound-channel-adapter destination="activeMqQueue" channel="jbossIn" connection-factory="connectionFactory">
<integration:poller>
<integration:interval-trigger interval="1000"/>
</integration:poller>
</jms:inbound-channel-adapter>

<jms:outbound-channel-adapter destination="jBossQueueTarget" channel="jbossIn" connection-factory="jbossConnectionFactory">
</jms:outbound-channel-adapter>

<integration:channel id="jbossIn" />



and here is the corresponding properties files

#jndi-jboss.properties
java.naming.provider.url=jnp://localhost:1099
java.naming.factory.initial=org.jnp.interfaces.Nam ingContextFactory
java.naming.factory.url.pkgs=org.jboss.naming.clie nt


#jndi.properties
java.naming.factory.initial = org.apache.activemq.jndi.ActiveMQInitialContextFac tory
java.naming.provider.url = tcp://localhost:61616
queue.MyQueue = example.A

[Mark Fisher]

Can you post the relevant excerpt from the Servlet code that sends the Message as well?... including its configuration if possible?

[curiousmind]

Here it goes....


Properties props = new Properties();
props.put(Context.INITIAL_CONTEXT_FACTORY, "org.apache.activemq.jndi.ActiveMQInitialContextFac tory");
props.put(Context.PROVIDER_URL, "tcp://localhost:61616");
props.put("queue.MyQueue", "example.A");
Context ctx = new InitialContext(props);
QueueConnectionFactory queueFactory = (QueueConnectionFactory)ctx.lookup("ConnectionFactory");
QueueConnection queueCon = queueFactory.createQueueConnection();
Queue queue = (Queue)ctx.lookup("MyQueue");
QueueSession queueSession = queueCon.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
QueueSender queueSender = queueSession.createSender(queue);

TextMessage textMessage = queueSession.createTextMessage();
textMessage.setText("A critical Message.");
queueSender.send(textMessage);
queueSession.close();
queueCon.close();


Sorry for delay...

[curiousmind]

hey mark, is there anything wrong with the code.. can you pls reply....

[Mark Fisher]

Nothing obvious is jumping out at me, but it seems like somewhere you must be creating the other JBoss Queue or at least providing its name ("queue.B"). Are there any other missing pieces?... maybe in the objects configured in the JNDI registry themselves?

[curiousmind]

Hi Mark,

I could able to resolve the problem. The solution is strange. Initially, I was using the default queue that comes with Acive MQ to post messages.

I created new queue in Active MQ and started posting messages to that queue and it worked well. no more redirection of messages and no more creation of new queue. Not sure how the default queues are configured in Active MQ.

Thanks alot for your help.

Would you recommend any books or online documentation which would help me to get more insight of Spring Integration (aprat from the reference documentation that comes with project)?

[Mark Fisher]

I'm really glad to hear that you've solved the problem!

As far as books, a few of the Spring Integration committers (including myself) have started writing "Spring Integration in Action". The hard copy will be published in 2010, but you can read the material as it evolves through Manning's Early Access Program. Some chapters are available already (4 as of right now), and new chapters will be added as they become available.

For today only (since we released Spring 3.0), they have a 45% discount special:
http://www.facebook.com/notes/manning-publications/spring-30-are-you-ready-45-off-any-spring-book/212674327550

[MmarcoM]

Curiousmind,
if you have time and patience, you could try to setup your own mqueues and run a standalone spring integ application to test your JMS functionality...
you will avoid any JBoss JMS configuration errors...

plus once you have setup your little JMS application, you could easily plug it in any JMS containr you want.

hth
marco

[curiousmind]

Well I was trying to apply Law of reusability :D
So thought I could use what is already there... but I ended up wasting too much of time because of that...
Anyways my problem has resolved now... and I know it can be used for any JMS provider...
Thanks for comments....

@Marker
Thanks for the book references...