PLEASE HELP ME MDB and Remote Topic
Hi all; i have some MDBs; i want that these MDBs are listening on a remote topic on a server with ip address: 192.168.xxx.yyy; now by following tips in this link:
http://www.jboss.org/wiki/Wiki.jsp?page=HowDoIConfigureAnMDBToTalkToARemoteQueue
i have these files ( note: i'm using JBoss 3.2.5 ):
jboss-service.xml:
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE server PUBLIC "-//JBoss//DTD MBean Service 3.2//EN" "http://www.jboss.org/j2ee/dtd/jboss-service_3_2.dtd">
<server>
<mbean code="org.jboss.mq.server.jmx.Topic" name="jboss.mq.destination:service=Topic,name=angelo">
<depends optional-attribute-name="DestinationManager">jboss.mq:service=DestinationManager</depends>
<depends optional-attribute-name="SecurityManager">jboss.mq:service=SecurityManager</depends>
</mbean>
<mbean code="org.jboss.jms.jndi.JMSProviderLoader" name="jboss.mq:service=JMSProviderLoader,name=RemoteJMSProvider,server=ContentAuthor">
<attribute name="ProviderName">RemoteJMSProvider</attribute>
<attribute name="ProviderAdapterClass">org.jboss.jms.jndi.JNDIProviderAdapter</attribute>
<attribute name="QueueFactoryRef">XAConnectionFactory</attribute>
<attribute name="TopicFactoryRef">XAConnectionFactory</attribute>
<attribute name="Properties">java.naming.factory.initial=org.jnp.interfaces.NamingContextFactory
java.naming.factory.url.pkgs=org.jnp.interfaces
java.naming.provider.url=192.168.xxx.yyy:1099
</attribute>
</mbean>
</server>
jboss.xml:
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE jboss PUBLIC "-//JBoss//DTD JBOSS 3.2//EN" "http://www.jboss.org/j2ee/dtd/jboss_3_2.dtd">
<jboss>
<enterprise-beans>
<message-driven>
<ejb-name>TestBean</ejb-name>
<destination-jndi-name>topic/angelo</destination-jndi-name>
<!-- I must include this if not the deployment fails -->
<mdb-subscription-id>mySubscription</mdb-subscription-id>
<invoker-bindings>
<invoker>
<invoker-proxy-binding-name>RemoteTopic</invoker-proxy-binding-name>
<jndi-name>topic/angelo</jndi-name>
</invoker>
</invoker-bindings>
</message-driven>
</enterprise-beans>
<invoker-proxy-bindings>
<invoker-proxy-binding>
<name>RemoteTopic</name>
<invoker-mbean>remote</invoker-mbean>
<proxy-factory>org.jboss.ejb.plugins.jms.JMSContainerInvoker</proxy-factory>
<proxy-factory-config>
<JMSProviderAdapterJNDI>RemoteJMSProvider</JMSProviderAdapterJNDI>
<ServerSessionPoolFactoryJNDI>StdJMSPool</ServerSessionPoolFactoryJNDI>
<MinimumSize>1</MinimumSize>
<KeepAliveMillis>30000</KeepAliveMillis>
<MaxMessages>1</MaxMessages>
<MDBConfig>
<ReconnectIntervalSec>10</ReconnectIntervalSec>
<DLQConfig>
<DestinationQueue>queue/DLQ</DestinationQueue>
<MaxTimesRedelivered>10</MaxTimesRedelivered>
<TimeToLive>0</TimeToLive>
</DLQConfig>
</MDBConfig>
</proxy-factory-config>
</invoker-proxy-binding>
</invoker-proxy-bindings>
</jboss>
ejb-jar.xml:
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE ejb-jar PUBLIC "-//Sun Microsystems, Inc.//DTD Enterprise JavaBeans 2.0//EN" "http://java.sun.com/dtd/ejb-jar_2_0.dtd">
<ejb-jar>
<display-name>TestBeanJMS</display-name>
<enterprise-beans>
<message-driven>
<ejb-name>TestBean</ejb-name>
<ejb-class>testjms.TestBean</ejb-class>
<transaction-type>Container</transaction-type>
<message-driven-destination>
<destination-type>javax.jms.Topic</destination-type>
<subscription-durability>Durable</subscription-durability>
</message-driven-destination>
</message-driven>
</enterprise-beans>
<assembly-descriptor>
<container-transaction>
<method>
<ejb-name>TestBean</ejb-name>
<method-name>*</method-name>
</method>
<trans-attribute>Required</trans-attribute>
</container-transaction>
</assembly-descriptor>
</ejb-jar>
This is my sample MDB code:
package testjms;
import javax.ejb.MessageDrivenBean;
import javax.jms.MessageListener;
import javax.ejb.MessageDrivenContext;
import javax.ejb.CreateException;
import javax.jms.Message;
public class TestBean implements MessageDrivenBean, MessageListener {
MessageDrivenContext messageDrivenContext;
public void ejbCreate() {
}
public void ejbRemove() {
}
public void onMessage(Message message) {
System.out.println( "Ricevuto: "+ message );
}
public void setMessageDrivenContext(MessageDrivenContext
messageDrivenContext) {
this.messageDrivenContext = messageDrivenContext;
}
}this is my publisher code ( generated by JBuilder ):
package testjms;
import javax.naming.InitialContext;
import javax.jms.MessageListener;
import javax.naming.Context;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicConnection;
import javax.jms.TopicSession;
import javax.jms.TopicPublisher;
import javax.jms.TopicSubscriber;
import javax.jms.Topic;
import javax.naming.NamingException;
import java.util.Hashtable;
import java.io.Serializable;
import javax.jms.Message;
/**
<p>Title: </p>
<p>Description: </p>
<p>Copyright: Copyright (c) 2004</p>
<p>Company: </p>
@author not attributable
@version 1.0
* Use this class to publish and subscribe to messages.
* To send a text message:
* <code>
* Test test = new Test();
* test.publishText("Hello world");
* test.close(); //Release resources
* </code>
* To receive a message:
* <code>
* Test test = new Test();
* test.getTopicSubscriber();
* </code>
*/
public class Test implements MessageListener {
private static Context context;
private boolean transacted = false;
private int acknowledgementMode = javax.jms.Session.AUTO_ACKNOWLEDGE;
private TopicConnectionFactory topicConnectionFactory;
private TopicConnection topicConnection;
private TopicSession topicSession;
private TopicPublisher topicPublisher;
private TopicSubscriber topicSubscriber;
private Topic topic;
private String topicConnectionFactoryName = "ConnectionFactory";
private String publishTopicName = "angelo";
private String subscribeTopicName = "angelo";
private String clientId = "angeloClient";
private String durableName = "test";
private boolean durable = true;
public void setTransacted(boolean transacted) {
this.transacted = transacted;
}
public TopicSubscriber getTopicSubscriber() throws Exception {
if (topicSubscriber == null) {
if (isDurable()) {
topicSubscriber = getTopicSession(true).createDurableSubscriber(
getSubscribeTopic(), getDurableName());
} else {
topicSubscriber = getTopicSession(true).createSubscriber(
getSubscribeTopic());
}
topicSubscriber.setMessageListener(this);
getTopicConnection(true).start();
}
return topicSubscriber;
}
public TopicPublisher getTopicPublisher() throws Exception {
if (topicPublisher == null) {
topicPublisher = getTopicSession(false).createPublisher(
getPublishTopic());
}
return topicPublisher;
}
public Topic getPublishTopic() throws Exception {
if (topic == null) {
Object obj = getContext().lookup(publishTopicName);
topic = (Topic) obj;
}
return topic;
}
public TopicSession getTopicSession(boolean consumer) throws Exception {
if (topicSession == null) {
topicSession = getTopicConnection(consumer).createTopicSession(
isTransacted(), getAcknowledgementMode());
}
return topicSession;
}
public TopicConnection getTopicConnection(boolean consumer) throws
Exception {
if (topicConnection == null) {
topicConnection = getTopicConnectionFactory().createTopicConnection();
topicConnection.start();
if (isDurable() && consumer) {
topicConnection.setClientID(clientId);
}
}
return topicConnection;
}
public TopicConnectionFactory getTopicConnectionFactory() throws Exception {
if (topicConnectionFactory == null) {
Object obj = getContext().lookup(topicConnectionFactoryName);
topicConnectionFactory = (TopicConnectionFactory) obj;
}
return topicConnectionFactory;
}
public void setDurable(boolean durable) {
this.durable = durable;
}
public boolean isDurable() {
return durable;
}
public void setDurableName(String durableName) {
this.durableName = durableName;
}
public String getDurableName() {
return durableName;
}
public void setClientId(String clientId) {
this.clientId = clientId;
}
public String getClientId() {
return clientId;
}
public void setSubscribeTopicName(String subscribeTopicName) {
this.subscribeTopicName = subscribeTopicName;
}
public String getSubscribeTopicName() {
return subscribeTopicName;
}
public void setPublishTopicName(String publishTopicName) {
this.publishTopicName = publishTopicName;
}
public String getPublishTopicName() {
return publishTopicName;
}
public void setTopicConnectionFactoryName(String topicConnectionFactoryName) {
this.topicConnectionFactoryName = topicConnectionFactoryName;
}
public String getTopicConnectionFactoryName() {
return topicConnectionFactoryName;
}
public void setAcknowledgementMode(int acknowledgementMode) {
this.acknowledgementMode = acknowledgementMode;
}
public int getAcknowledgementMode() {
return acknowledgementMode;
}
public boolean isTransacted() {
return transacted;
}
private Context getInitialContext() throws NamingException {
Hashtable environment = new Hashtable();
environment.put(Context.INITIAL_CONTEXT_FACTORY,
"org.jnp.interfaces.NamingContextFactory");
environment.put(Context.URL_PKG_PREFIXES,
"org.jboss.naming:org.jnp.interfaces");
environment.put(Context.PROVIDER_URL, "jnp://localhost:1099");
return new InitialContext(environment);
}
private Context getContext() throws Exception {
if (context == null) {
try {
context = getInitialContext();
} catch (Exception ex) {
ex.printStackTrace();
throw ex;
}
}
return context;
}
public Topic getSubscribeTopic() throws Exception {
if (topic == null) {
Object obj = getContext().lookup(subscribeTopicName);
topic = (Topic) obj;
}
return topic;
}
public void publishText(String message) throws Exception {
javax.jms.TextMessage textMessage = getTopicSession(false).
createTextMessage();
textMessage.clearBody();
textMessage.setText(message);
getTopicPublisher().publish(textMessage);
if (isTransacted()) {
getTopicSession(false).commit();
}
}
public void publishObject(Serializable message) throws Exception {
javax.jms.ObjectMessage objectMessage = getTopicSession(false).
createObjectMessage();
objectMessage.clearBody();
objectMessage.setObject(message);
getTopicPublisher().publish(objectMessage);
if (isTransacted()) {
getTopicSession(false).commit();
}
}
public void onMessage(Message message) {
if (message instanceof javax.jms.BytesMessage) {
javax.jms.BytesMessage bytesMessage = (javax.jms.BytesMessage)
message;
/** @todo Process bytesMessage here */
} else if (message instanceof javax.jms.MapMessage) {
javax.jms.MapMessage mapMessage = (javax.jms.MapMessage) message;
/** @todo Process mapMessage here */
} else if (message instanceof javax.jms.ObjectMessage) {
javax.jms.ObjectMessage objectMessage = (javax.jms.ObjectMessage)
message;
/** @todo Process objectMessage here */
} else if (message instanceof javax.jms.StreamMessage) {
javax.jms.StreamMessage streamMessage = (javax.jms.StreamMessage)
message;
/** @todo Process streamMessage here */
} else if (message instanceof javax.jms.TextMessage) {
javax.jms.TextMessage objectMessage = (javax.jms.TextMessage)
message;
/** @todo Process textMessage here */
}
if (isTransacted()) {
try {
getTopicSession(false).commit();
} catch (Exception ex) {
ex.printStackTrace();
}
}
}
public void close() throws Exception {
if (topicPublisher != null) {
topicPublisher.close();
}
if (topicSubscriber != null) {
topicSubscriber.close();
}
if (topicSession != null) {
topicSession.close();
}
if (topicConnection != null) {
topicConnection.close();
}
}
public static void main(String[] args) {
Test test = new Test();
try {
test.publishText("Hello world");
test.close(); //Release resources
} catch (Exception ex) {
}
}
}It seems to me that all is correct, infact i have no error o warning, but the MDB receive no message.... can anybody tell me where i'm wrong?
Thanks to all
