Recieving response from MDB

[Prakash Dwivedi]

Hello Friends,
i am trying to develop a program where i put a message in a request queue, from the request queue mdb picks up the message, does some processing and puts the response back in response queue. The calling program tries to read the message from the response queue using QueueReciever and JMSCorrelation Id. But some how things are not working right. My QueueReceiver is not able to pick the message from the response queue.

Following is the code written in the SessionBean which is sending and recieving message.

  
public String sendRecvMessage(String strMsg){
String msgId = sendMessage(strMsg);
for(int i=0;i<100000;i++);
return getMessage(msgId);
}

public String getMessage(String msgId){

try{
InitialContext ctx= new InitialContext(); 
QueueConnectionFactory conFact= (QueueConnectionFactory) ctx.lookup("jms/QCF");
String selector = "JMSCorrelationID = '"+msgId+"'";
Queue responseQueue= (Queue)  ctx.lookup("jms/ResponseQueue");
QueueConnection qCon= conFact.createQueueConnection();
qCon.start();
QueueSession session= qCon.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
QueueReceiver reciever= session.createReceiver(responseQueue,selector);
TextMessage recMsg= (TextMessage)reciever.receive(30000);
String str ="";
if(recMsg == null)
str =  "Null recieved";
else
str = recMsg.getText();
System.out.println("Message Recd" + str);
qCon.close();
session.close();
return str;
}catch(Exception ex){
ex.printStackTrace();
return ex.toString();
}
}

public String sendMessage(String strMsg){
try{
InitialContext ctx= new InitialContext(); 
QueueConnectionFactory conFact= (QueueConnectionFactory) ctx.lookup("jms/QCF");
Queue requestQueue= (Queue)  ctx.lookup("jms/RequestQueue");
QueueConnection qCon= conFact.createQueueConnection();
qCon.start();
QueueSession session= qCon.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
QueueSender sender= session.createSender(requestQueue);
TextMessage msg= session.createTextMessage(strMsg);
sender.send(msg);
String msgId= msg.getJMSMessageID();    
qCon.close();
session.close();
return msgId;
}catch(Exception ex){
ex.printStackTrace();
return ex.toString();
}
}

Following is the code written in MDB

 
public void onMessage(javax.jms.Message msg) {
TextMessage tMsg = (TextMessage)msg;

try{
StringBuffer sb= new StringBuffer(tMsg.getText());
String msgId= msg.getJMSMessageID();
InitialContext ctx= new InitialContext(); 
QueueConnectionFactory conFact= (QueueConnectionFactory) ctx.lookup("jms/QCF");
Queue requestQueue= (Queue)  ctx.lookup("jms/ResponseQueue");
QueueConnection qCon= conFact.createQueueConnection();
qCon.start();
QueueSession session= qCon.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
QueueSender sender= session.createSender(requestQueue);
msg= session.createTextMessage(sb.reverse().toString());
msg.setJMSCorrelationID(msgId);
sender.send(msg);
qCon.close();
session.close();    

}catch(Exception ex){
ex.printStackTrace();
}
}

Plz help i am unable to understand where i am wrong. I am using WSAD 5.1.2 Developer Edition on Windows.
[ July 02, 2005: Message edited by: Prakash Dwivedi ]

[Osuwari Inu]

The code seems ok. I'm guessing the problem is the same one I used to have way back.
Have you tried adding some debugging statements to your code? If you put them in the correct places I think you might see something not quite unlike the following scenario:
- Session bean: Sending message
- Session bean: Waiting for reply
- Session bean: Timed out waiting for reply
- MDB: Received message
- MDB: Sending reply
- MDB: Reply sent

What happens here is that both the send and the receive in your test code are done in the same transaction. If the send method does not throw an exception, that only means that the transport layer has accepted your message, not that it has actually been sent. (The mail is in the mailbox).
For the message to be sent, the current transaction has to complete succesfully. Which is done (at the earliest) after sendRecvMessage is finished.

The easiest way to solve this problem and make your test case work, is to give the receive (getMessage) method a transaction attribute of RequiresNew. This will start a new transaction, closing the previous one, allowing the message to be send.

Some constructive criticism: (or: things you didn't ask for )
In stead of looping like:
for(int i=0;i<100000;i++);
to wait for a small period, use:
Thread.sleep( 100 ); // in milliseconds

This will not cost cpu time. The looping for time is really ASM/Pascal age

Try closing things in the opposite order of opening:
Open connection
Open session
Close session
Close connection

This may avoid unexpected exceptions and generally is considered good coding practice.

Hope this helps

[Ravichandramohan Pelluru]

Hi,

currently i am doing websphere migraiton from 4.0 to 5.1.0

i am also facing the same problem what you have

Problem description:

A stateless session bean puts a message in a request queue, From the request queue a C++ Program (independent component) picks up the message,
does some processing and Puts the response back in reply queue.
The session bean then tries to read the message from the reply queue using QueueReciever and JMSCorrelationID in a separate JMS Session.
But QueueReceiver is not able to pick the message from the reply queue. Using createBrowse I am able to read the messages in the reply queue.

Observations:
1) Same code is working in Websphere 4.0, not working in websphere 5.1.0
3) I created one jsp which have JMS code to get the message using JMSCorrelationID. It is working fine.
Note: it is reading the message from web container not from EJB Container

Environmental Details:
=====================
JDK 1.4.2_05-b04
Websphere 5.1.0 [base version]
MQ Server 5.3

Stateless session bean:


private QueueConnection qc = null;
private QueuerequestQ = null;
private QueuereplyQ = null;

public void ejbCreate() throws javax.ejb.CreateException
{
InitialContext ctx = new InitialContext();
qcf = (QueueConnectionFactory) ctx.lookup("jms/QCF");
requestQ = (Queue) ctx.lookup("jms/RequestQueue");
replyQ = (Queue) ctx.lookup("jms/ReplyQueue");
qc = qcf.createQueueConnection();
}


public void ejbRemove()
{
if(qc != null)
qc.close();
}

public String processRequest(String strMessage)
{
String CorrelationId = sendMessage(strMessage);
String strReceiveMsg = receiveMessage(CorrelationId);
return strReceiveMsg;
}

private String sendMessage(String cDataStream) throws javax.ejb.EJBException,javax.jms.JMSException,javax.naming.NamingException
{
QueueSession session = null;
QueueSender sender = null;
TextMessage message = null;
String correlationId = null;
try {
// create a session.
session = qc.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

// create a QueueSender
sender = session.createSender(requestQ);

// create a message to send to the queue...
message = session.createTextMessage(cDataStream);

// send
sender.send(message);

// store the messageID as correlationID
correlationId = message.getJMSMessageID();

}
catch (JMSException je)
{
je.printStackTrace();
throw je;
}
finally
{
try {
if (sender != null)
sender.close();
if (session != null)
session.close();
} catch (JMSException je) {
je.printStackTrace();
throw je;
}
}

return correlationId;
}

private String receiveMessage(String correlationId) throws javax.ejb.EJBException,javax.jms.JMSException,javax.naming.NamingException
{
QueueSession session = null;
QueueReceiver receiver = null;
TextMessage message = null;
String messageSelector = "JMSCorrelationID=\'" + correlationId + "\'";
String msg = null;

try {
// start a connection
qc.start();

// create a session.
session = qc.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

// create a QueueReceiver
receiver = session.createReceiver(replyQ,messageSelector);

// receive a message from the queue, wait 10 seconds
try
{
message = (TextMessage) receiver.receive(10*1000);
msg = message.getText();
}
catch(Exception e)
{
e.printStackTrace();
}

}
catch (JMSException je)
{
je.printStackTrace();
throw je;
}
finally
{
try {
if (receiver != null)
receiver.close();
if (session != null)
session.close();
if (qc != null)
qc.stop();
} catch (JMSException je)
{
je.printStackTrace();
}

}
return msg;
}


Plz help i am unable to understand where i am wrong.
------------------------
Thanks & Regards
Ravichandramohan

[Jaikiran Pai]

Hi,
Please refer the following link, which talks about transactions for sending JMS messages. Actually, you cant send and receive messages in the same transaction.

http://e-docs.bea.com/wls/docs70/faq/jms.html#252684

Briefly, the above link says the following:

Q. Why can't I receive a message that I send within a container-managed transaction?

A. If you are using container-managed transactions, the original message sent from the EJB will never be sent. Here is what is happening.


Container starts transaction.

Start method.

Generate new message.

Send message (message isn't sent - it's buffered until transaction commit).

Do a blocking receive on a queue.

End method.

Transaction Commit never Reached because original message was never sent because you can't get past blocking receive.
The solution is to either use bean-managed transactions, or to break the send and receive into two separate methods.

[Osuwari Inu]

The solution is to either use bean-managed transactions, or to break the send and receive into two separate methods.

And then set a REQUIRESNEW transaction attribute on the receive method!

Oh and one other fun detail I found out after we deployed the application on WebSphere on AIX. Specifying a negative timeout (bug in my own round robin receive algoritm) on Windows would result in an IllegalArgumentException, whilst on AIX it would be interpreted as 0 (block indefinitely). A minor but _very_ annoying detail.

Be warned