Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

added changes from http://jira.activemq.org/jira//browse/AMQ-660 

to allow destination conversation on outbound messages with replyTo destinations



git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@393383 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Robert Davies 2006-04-12 04:46:22 +00:00
parent d964145532
commit 28647f1790
5 changed files with 204 additions and 104 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
activemq-core/src/main/java/org/apache/activemq/network/jms/DestinationBridge.java
View File

@ -34,7 +34,7 @@ import edu.emory.mathcs.backport.java.util.concurrent.atomic.AtomicBoolean;
* *
* @version $Revision: 1.1.1.1 $ * @version $Revision: 1.1.1.1 $
*/ */
abstract class DestinationBridge implements Service,MessageListener{ public abstract class DestinationBridge implements Service,MessageListener{
private static final Log log=LogFactory.getLog(DestinationBridge.class); private static final Log log=LogFactory.getLog(DestinationBridge.class);
protected MessageConsumer consumer; protected MessageConsumer consumer;
protected AtomicBoolean started=new AtomicBoolean(false); protected AtomicBoolean started=new AtomicBoolean(false);
@ -95,31 +95,34 @@ abstract class DestinationBridge implements Service,MessageListener{
} }
public void onMessage(Message message){ public void onMessage(Message message){
if(started.get()&&message!=null){ if(started.get()&&message!=null){
try{ try{
if(doHandleReplyTo){ Message converted;
Destination replyTo=message.getJMSReplyTo(); if(doHandleReplyTo){
if(replyTo!=null){ Destination replyTo = message.getJMSReplyTo();
replyTo=processReplyToDestination(replyTo); if(replyTo != null){
message.setJMSReplyTo(replyTo); converted = jmsMessageConvertor.convert(message, processReplyToDestination(replyTo));
} } else {
}else { converted = jmsMessageConvertor.convert(message);
message.setJMSReplyTo(null); }
} } else {
Message converted=jmsMessageConvertor.convert(message); message.setJMSReplyTo(null);
sendMessage(converted); converted = jmsMessageConvertor.convert(message);
message.acknowledge(); }
}catch(JMSException e){ sendMessage(converted);
log.error("failed to forward message: "+message,e); message.acknowledge();
try{ }catch(JMSException e){
stop(); log.error("failed to forward message: "+message,e);
}catch(Exception e1){ try{
log.warn("Failed to stop cleanly",e1); stop();
} }catch(Exception e1){
} log.warn("Failed to stop cleanly",e1);
} }
}
}
} }
/** /**
* @return Returns the doHandleReplyTo. * @return Returns the doHandleReplyTo.
*/ */

5
activemq-core/src/main/java/org/apache/activemq/network/jms/JmsMesageConvertor.java
View File

@ -16,6 +16,8 @@
*/ */
package org.apache.activemq.network.jms; package org.apache.activemq.network.jms;
import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException; import javax.jms.JMSException;
import javax.jms.Message; import javax.jms.Message;
@ -34,5 +36,8 @@ public interface JmsMesageConvertor {
*/ */
public Message convert(Message message) throws JMSException; public Message convert(Message message) throws JMSException;
public Message convert(Message message, Destination replyTo) throws JMSException;
public void setConnection(Connection connection);
} }

118
activemq-core/src/main/java/org/apache/activemq/network/jms/JmsQueueConnector.java
View File

@ -16,9 +16,6 @@
*/ */
package org.apache.activemq.network.jms; package org.apache.activemq.network.jms;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import javax.jms.Connection; import javax.jms.Connection;
import javax.jms.Destination; import javax.jms.Destination;
import javax.jms.JMSException; import javax.jms.JMSException;
@ -28,6 +25,9 @@ import javax.jms.QueueConnectionFactory;
import javax.jms.QueueSession; import javax.jms.QueueSession;
import javax.jms.Session; import javax.jms.Session;
import javax.naming.NamingException; import javax.naming.NamingException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/** /**
* A Bridge to other JMS Queue providers * A Bridge to other JMS Queue providers
* *
@ -46,16 +46,14 @@ public class JmsQueueConnector extends JmsConnector{
private InboundQueueBridge[] inboundQueueBridges; private InboundQueueBridge[] inboundQueueBridges;
private OutboundQueueBridge[] outboundQueueBridges; private OutboundQueueBridge[] outboundQueueBridges;
public boolean init(){ public boolean init(){
boolean result=super.init(); boolean result=super.init();
if(result){ if(result){
try{ try{
initializeForeignQueueConnection(); initializeForeignQueueConnection();
initializeLocalQueueConnection(); initializeLocalQueueConnection();
initializeInboundJmsMessageConvertor();
initializeOutboundJmsMessageConvertor();
initializeInboundQueueBridges(); initializeInboundQueueBridges();
initializeOutboundQueueBridges(); initializeOutboundQueueBridges();
}catch(Exception e){ }catch(Exception e){
@ -250,6 +248,14 @@ public class JmsQueueConnector extends JmsConnector{
localQueueConnection.start(); localQueueConnection.start();
} }
protected void initializeInboundJmsMessageConvertor(){
inboundMessageConvertor.setConnection(localQueueConnection);
}
protected void initializeOutboundJmsMessageConvertor(){
outboundMessageConvertor.setConnection(outboundQueueConnection);
}
protected void initializeInboundQueueBridges() throws JMSException{ protected void initializeInboundQueueBridges() throws JMSException{
if(inboundQueueBridges!=null){ if(inboundQueueBridges!=null){
QueueSession outboundSession = outboundQueueConnection.createQueueSession(false,Session.AUTO_ACKNOWLEDGE); QueueSession outboundSession = outboundQueueConnection.createQueueSession(false,Session.AUTO_ACKNOWLEDGE);
@ -287,7 +293,6 @@ public class JmsQueueConnector extends JmsConnector{
bridge.setProducerQueue(foreignQueue); bridge.setProducerQueue(foreignQueue);
bridge.setProducerConnection(outboundQueueConnection); bridge.setProducerConnection(outboundQueueConnection);
bridge.setConsumerConnection(localQueueConnection); bridge.setConsumerConnection(localQueueConnection);
bridge.setDoHandleReplyTo(false);
if(bridge.getJmsMessageConvertor()==null){ if(bridge.getJmsMessageConvertor()==null){
bridge.setJmsMessageConvertor(getOutboundMessageConvertor()); bridge.setJmsMessageConvertor(getOutboundMessageConvertor());
} }
@ -299,38 +304,71 @@ public class JmsQueueConnector extends JmsConnector{
} }
} }
protected Destination createReplyToBridge(Destination destination, Connection consumerConnection, Connection producerConnection){ protected Destination createReplyToBridge(Destination destination, Connection replyToProducerConnection, Connection replyToConsumerConnection){
Queue queue = (Queue)destination; Queue replyToProducerQueue =(Queue)destination;
OutboundQueueBridge bridge = (OutboundQueueBridge) replyToBridges.get(queue); boolean isInbound = replyToProducerConnection.equals(localQueueConnection);
if (bridge == null){
bridge = new OutboundQueueBridge(){ if(isInbound){
//we only handle replyTo destinations - inbound InboundQueueBridge bridge = (InboundQueueBridge) replyToBridges.get(replyToProducerQueue);
protected Destination processReplyToDestination (Destination destination){ if (bridge == null){
return null; bridge = new InboundQueueBridge(){
} protected Destination processReplyToDestination (Destination destination){
}; return null;
try{ }
QueueSession localSession = localQueueConnection.createQueueSession(false,Session.AUTO_ACKNOWLEDGE); };
Queue localQueue = localSession.createTemporaryQueue(); try{
localSession.close(); QueueSession replyToConsumerSession = ((QueueConnection)replyToConsumerConnection).createQueueSession(false,Session.AUTO_ACKNOWLEDGE);
bridge.setConsumerQueue(localQueue); Queue replyToConsumerQueue = replyToConsumerSession.createTemporaryQueue();
bridge.setProducerQueue(queue); replyToConsumerSession.close();
bridge.setProducerConnection(outboundQueueConnection); bridge.setConsumerQueue(replyToConsumerQueue);
bridge.setConsumerConnection(localQueueConnection); bridge.setProducerQueue(replyToProducerQueue);
bridge.setDoHandleReplyTo(false); bridge.setProducerConnection((QueueConnection)replyToProducerConnection);
if(bridge.getJmsMessageConvertor()==null){ bridge.setConsumerConnection((QueueConnection)replyToConsumerConnection);
bridge.setJmsMessageConvertor(getOutboundMessageConvertor()); bridge.setDoHandleReplyTo(false);
} if(bridge.getJmsMessageConvertor()==null){
bridge.setJmsConnector(this); bridge.setJmsMessageConvertor(getInboundMessageConvertor());
bridge.start(); }
log.info("Created replyTo bridge for " + queue); bridge.setJmsConnector(this);
}catch(Exception e){ bridge.start();
log.error("Failed to create replyTo bridge for queue: " + queue,e); log.info("Created replyTo bridge for " + replyToProducerQueue);
return null; }catch(Exception e){
} log.error("Failed to create replyTo bridge for queue: " + replyToProducerQueue, e);
replyToBridges.put(queue, bridge); return null;
} }
return bridge.getConsumerQueue(); replyToBridges.put(replyToProducerQueue, bridge);
}
return bridge.getConsumerQueue();
}else{
OutboundQueueBridge bridge = (OutboundQueueBridge) replyToBridges.get(replyToProducerQueue);
if (bridge == null){
bridge = new OutboundQueueBridge(){
protected Destination processReplyToDestination (Destination destination){
return null;
}
};
try{
QueueSession replyToConsumerSession = ((QueueConnection)replyToConsumerConnection).createQueueSession(false,Session.AUTO_ACKNOWLEDGE);
Queue replyToConsumerQueue = replyToConsumerSession.createTemporaryQueue();
replyToConsumerSession.close();
bridge.setConsumerQueue(replyToConsumerQueue);
bridge.setProducerQueue(replyToProducerQueue);
bridge.setProducerConnection((QueueConnection)replyToProducerConnection);
bridge.setConsumerConnection((QueueConnection)replyToConsumerConnection);
bridge.setDoHandleReplyTo(false);
if(bridge.getJmsMessageConvertor()==null){
bridge.setJmsMessageConvertor(getOutboundMessageConvertor());
}
bridge.setJmsConnector(this);
bridge.start();
log.info("Created replyTo bridge for " + replyToProducerQueue);
}catch(Exception e){
log.error("Failed to create replyTo bridge for queue: " + replyToProducerQueue, e);
return null;
}
replyToBridges.put(replyToProducerQueue, bridge);
}
return bridge.getConsumerQueue();
}
} }
protected Queue createActiveMQQueue(QueueSession session,String queueName) throws JMSException{ protected Queue createActiveMQQueue(QueueSession session,String queueName) throws JMSException{

112
activemq-core/src/main/java/org/apache/activemq/network/jms/JmsTopicConnector.java
View File

@ -25,6 +25,7 @@ import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory; import javax.jms.TopicConnectionFactory;
import javax.jms.TopicSession; import javax.jms.TopicSession;
import javax.naming.NamingException; import javax.naming.NamingException;
import org.apache.commons.logging.Log; import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory; import org.apache.commons.logging.LogFactory;
@ -46,16 +47,14 @@ public class JmsTopicConnector extends JmsConnector{
private InboundTopicBridge[] inboundTopicBridges; private InboundTopicBridge[] inboundTopicBridges;
private OutboundTopicBridge[] outboundTopicBridges; private OutboundTopicBridge[] outboundTopicBridges;
public boolean init(){ public boolean init(){
boolean result=super.init(); boolean result=super.init();
if(result){ if(result){
try{ try{
initializeForeignTopicConnection(); initializeForeignTopicConnection();
initializeLocalTopicConnection(); initializeLocalTopicConnection();
initializeInboundJmsMessageConvertor();
initializeOutboundJmsMessageConvertor();
initializeInboundTopicBridges(); initializeInboundTopicBridges();
initializeOutboundTopicBridges(); initializeOutboundTopicBridges();
}catch(Exception e){ }catch(Exception e){
@ -251,6 +250,14 @@ public class JmsTopicConnector extends JmsConnector{
localTopicConnection.start(); localTopicConnection.start();
} }
protected void initializeInboundJmsMessageConvertor(){
inboundMessageConvertor.setConnection(localTopicConnection);
}
protected void initializeOutboundJmsMessageConvertor(){
outboundMessageConvertor.setConnection(outboundTopicConnection);
}
protected void initializeInboundTopicBridges() throws JMSException{ protected void initializeInboundTopicBridges() throws JMSException{
if(inboundTopicBridges!=null){ if(inboundTopicBridges!=null){
TopicSession outboundSession = outboundTopicConnection.createTopicSession(false,Session.AUTO_ACKNOWLEDGE); TopicSession outboundSession = outboundTopicConnection.createTopicSession(false,Session.AUTO_ACKNOWLEDGE);
@ -288,7 +295,6 @@ public class JmsTopicConnector extends JmsConnector{
bridge.setProducerTopic(foreignTopic); bridge.setProducerTopic(foreignTopic);
bridge.setProducerConnection(outboundTopicConnection); bridge.setProducerConnection(outboundTopicConnection);
bridge.setConsumerConnection(localTopicConnection); bridge.setConsumerConnection(localTopicConnection);
bridge.setDoHandleReplyTo(false);
if(bridge.getJmsMessageConvertor()==null){ if(bridge.getJmsMessageConvertor()==null){
bridge.setJmsMessageConvertor(getOutboundMessageConvertor()); bridge.setJmsMessageConvertor(getOutboundMessageConvertor());
} }
@ -300,39 +306,71 @@ public class JmsTopicConnector extends JmsConnector{
} }
} }
protected Destination createReplyToBridge(Destination destination, Connection consumerConnection, Connection producerConnection){ protected Destination createReplyToBridge(Destination destination, Connection replyToProducerConnection, Connection replyToConsumerConnection){
Topic topic =(Topic)destination; Topic replyToProducerTopic =(Topic)destination;
boolean isInbound = replyToProducerConnection.equals(localTopicConnection);
OutboundTopicBridge bridge = (OutboundTopicBridge) replyToBridges.get(topic); if(isInbound){
if (bridge == null){ InboundTopicBridge bridge = (InboundTopicBridge) replyToBridges.get(replyToProducerTopic);
bridge = new OutboundTopicBridge(){ if (bridge == null){
//we only handle replyTo destinations - inbound bridge = new InboundTopicBridge(){
protected Destination processReplyToDestination (Destination destination){ protected Destination processReplyToDestination (Destination destination){
return null; return null;
} }
}; };
try{ try{
TopicSession localSession = localTopicConnection.createTopicSession(false,Session.AUTO_ACKNOWLEDGE); TopicSession replyToConsumerSession = ((TopicConnection)replyToConsumerConnection).createTopicSession(false,Session.AUTO_ACKNOWLEDGE);
Topic localTopic = localSession.createTemporaryTopic(); Topic replyToConsumerTopic = replyToConsumerSession.createTemporaryTopic();
localSession.close(); replyToConsumerSession.close();
bridge.setConsumerTopic(localTopic); bridge.setConsumerTopic(replyToConsumerTopic);
bridge.setProducerTopic(topic); bridge.setProducerTopic(replyToProducerTopic);
bridge.setProducerConnection(outboundTopicConnection); bridge.setProducerConnection((TopicConnection)replyToProducerConnection);
bridge.setConsumerConnection(localTopicConnection); bridge.setConsumerConnection((TopicConnection)replyToConsumerConnection);
bridge.setDoHandleReplyTo(false); bridge.setDoHandleReplyTo(false);
if(bridge.getJmsMessageConvertor()==null){ if(bridge.getJmsMessageConvertor()==null){
bridge.setJmsMessageConvertor(getOutboundMessageConvertor()); bridge.setJmsMessageConvertor(getInboundMessageConvertor());
} }
bridge.setJmsConnector(this); bridge.setJmsConnector(this);
bridge.start(); bridge.start();
log.info("Created replyTo bridge for " + topic); log.info("Created replyTo bridge for " + replyToProducerTopic);
}catch(Exception e){ }catch(Exception e){
log.error("Failed to create replyTo bridge for topic: " + topic,e); log.error("Failed to create replyTo bridge for topic: " + replyToProducerTopic, e);
return null; return null;
} }
replyToBridges.put(topic, bridge); replyToBridges.put(replyToProducerTopic, bridge);
} }
return bridge.getConsumerTopic(); return bridge.getConsumerTopic();
}else{
OutboundTopicBridge bridge = (OutboundTopicBridge) replyToBridges.get(replyToProducerTopic);
if (bridge == null){
bridge = new OutboundTopicBridge(){
protected Destination processReplyToDestination (Destination destination){
return null;
}
};
try{
TopicSession replyToConsumerSession = ((TopicConnection)replyToConsumerConnection).createTopicSession(false,Session.AUTO_ACKNOWLEDGE);
Topic replyToConsumerTopic = replyToConsumerSession.createTemporaryTopic();
replyToConsumerSession.close();
bridge.setConsumerTopic(replyToConsumerTopic);
bridge.setProducerTopic(replyToProducerTopic);
bridge.setProducerConnection((TopicConnection)replyToProducerConnection);
bridge.setConsumerConnection((TopicConnection)replyToConsumerConnection);
bridge.setDoHandleReplyTo(false);
if(bridge.getJmsMessageConvertor()==null){
bridge.setJmsMessageConvertor(getOutboundMessageConvertor());
}
bridge.setJmsConnector(this);
bridge.start();
log.info("Created replyTo bridge for " + replyToProducerTopic);
}catch(Exception e){
log.error("Failed to create replyTo bridge for topic: " + replyToProducerTopic, e);
return null;
}
replyToBridges.put(replyToProducerTopic, bridge);
}
return bridge.getConsumerTopic();
}
} }
protected Topic createActiveMQTopic(TopicSession session,String topicName) throws JMSException{ protected Topic createActiveMQTopic(TopicSession session,String topicName) throws JMSException{

16
activemq-core/src/main/java/org/apache/activemq/network/jms/SimpleJmsMessageConvertor.java
View File

@ -16,6 +16,8 @@
*/ */
package org.apache.activemq.network.jms; package org.apache.activemq.network.jms;
import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException; import javax.jms.JMSException;
import javax.jms.Message; import javax.jms.Message;
@ -39,5 +41,19 @@ public class SimpleJmsMessageConvertor implements JmsMesageConvertor {
return message; return message;
} }
public Message convert(Message message, Destination replyTo) throws JMSException{
Message msg = convert(message);
if(replyTo != null) {
msg.setJMSReplyTo(replyTo);
}else{
msg.setJMSReplyTo(null);
}
return msg;
}
public void setConnection(Connection connection){
//do nothing
}
} }