mirror of https://github.com/apache/activemq.git
put back support for request/reply across networks
git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@395810 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
parent
4f0e6d51eb
commit
065c46f19d
|
@ -22,6 +22,7 @@ import edu.emory.mathcs.backport.java.util.concurrent.atomic.AtomicBoolean;
|
|||
|
||||
import org.apache.activemq.advisory.AdvisorySupport;
|
||||
import org.apache.activemq.command.ActiveMQDestination;
|
||||
import org.apache.activemq.command.ActiveMQTempDestination;
|
||||
import org.apache.activemq.command.ActiveMQTopic;
|
||||
import org.apache.activemq.command.BrokerId;
|
||||
import org.apache.activemq.command.BrokerInfo;
|
||||
|
@ -31,6 +32,7 @@ import org.apache.activemq.command.ConnectionInfo;
|
|||
import org.apache.activemq.command.ConsumerId;
|
||||
import org.apache.activemq.command.ConsumerInfo;
|
||||
import org.apache.activemq.command.DataStructure;
|
||||
import org.apache.activemq.command.DestinationInfo;
|
||||
import org.apache.activemq.command.ExceptionResponse;
|
||||
import org.apache.activemq.command.KeepAliveInfo;
|
||||
import org.apache.activemq.command.Message;
|
||||
|
@ -219,7 +221,11 @@ public abstract class DemandForwardingBridgeSupport implements Bridge {
|
|||
+destinationFilter));
|
||||
demandConsumerInfo.setPrefetchSize(prefetchSize);
|
||||
remoteBroker.oneway(demandConsumerInfo);
|
||||
|
||||
//we want infomation about Destinations as well
|
||||
ConsumerInfo destinationInfo = new ConsumerInfo(remoteSessionInfo,2);
|
||||
destinationInfo.setDestination(AdvisorySupport.TEMP_DESTINATION_COMPOSITE_ADVISORY_TOPIC);
|
||||
destinationInfo.setPrefetchSize(prefetchSize);
|
||||
remoteBroker.oneway(destinationInfo);
|
||||
startedLatch.countDown();
|
||||
}
|
||||
}
|
||||
|
@ -331,7 +337,34 @@ public abstract class DemandForwardingBridgeSupport implements Bridge {
|
|||
log.trace("Ignoring sub " + info + " already subscribed to matching destination");
|
||||
}
|
||||
}
|
||||
if(data.getClass()==RemoveInfo.class){
|
||||
else if (data.getClass()==DestinationInfo.class){
|
||||
// It's a destination info - we want to pass up
|
||||
//infomation about temporary destinations
|
||||
DestinationInfo destInfo = (DestinationInfo) data;
|
||||
BrokerId[] path=destInfo.getBrokerPath();
|
||||
if((path!=null&&path.length>= networkTTL)){
|
||||
if(log.isTraceEnabled())
|
||||
log.trace("Ignoring Subscription " + destInfo + " restricted to " + networkTTL + " network hops only");
|
||||
return;
|
||||
}
|
||||
if(contains(destInfo.getBrokerPath(),localBrokerPath[0])){
|
||||
// Ignore this consumer as it's a consumer we locally sent to the broker.
|
||||
if(log.isTraceEnabled())
|
||||
log.trace("Ignoring sub " + destInfo + " already routed through this broker once");
|
||||
return;
|
||||
}
|
||||
|
||||
destInfo.setConnectionId(localConnectionInfo.getConnectionId());
|
||||
if (destInfo.getDestination() instanceof ActiveMQTempDestination){
|
||||
//re-set connection id so comes from here
|
||||
ActiveMQTempDestination tempDest = (ActiveMQTempDestination) destInfo.getDestination();
|
||||
tempDest.setConnectionId(localSessionInfo.getSessionId().getConnectionId());
|
||||
}
|
||||
destInfo.setBrokerPath(appendToBrokerPath(destInfo.getBrokerPath(),getRemoteBrokerPath()));
|
||||
localBroker.oneway(destInfo);
|
||||
|
||||
}
|
||||
else if(data.getClass()==RemoveInfo.class){
|
||||
ConsumerId id=(ConsumerId) ((RemoveInfo) data).getObjectId();
|
||||
removeDemandSubscription(id);
|
||||
}
|
||||
|
@ -761,6 +794,8 @@ public abstract class DemandForwardingBridgeSupport implements Bridge {
|
|||
protected abstract void addRemoteBrokerToBrokerPath(ConsumerInfo info) throws IOException;
|
||||
|
||||
protected abstract void serviceRemoteBrokerInfo(Command command) throws IOException;
|
||||
|
||||
protected abstract BrokerId[] getRemoteBrokerPath();
|
||||
|
||||
public String getPassword() {
|
||||
return password;
|
||||
|
|
Loading…
Reference in New Issue