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

when sending - use quality of service applicable to message and subscriber 

git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@366191 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Robert Davies 2006-01-05 14:58:35 +00:00
parent 95a2aa6b8f
commit ef3b50e18b
1 changed files with 13 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -31,11 +31,13 @@ import org.apache.activemq.command.ConnectionInfo;
import org.apache.activemq.command.ConsumerId; import org.apache.activemq.command.ConsumerId;
import org.apache.activemq.command.ConsumerInfo; import org.apache.activemq.command.ConsumerInfo;
import org.apache.activemq.command.DataStructure; import org.apache.activemq.command.DataStructure;
import org.apache.activemq.command.ExceptionResponse;
import org.apache.activemq.command.Message; import org.apache.activemq.command.Message;
import org.apache.activemq.command.MessageAck; import org.apache.activemq.command.MessageAck;
import org.apache.activemq.command.MessageDispatch; import org.apache.activemq.command.MessageDispatch;
import org.apache.activemq.command.ProducerInfo; import org.apache.activemq.command.ProducerInfo;
import org.apache.activemq.command.RemoveInfo; import org.apache.activemq.command.RemoveInfo;
import org.apache.activemq.command.Response;
import org.apache.activemq.command.SessionInfo; import org.apache.activemq.command.SessionInfo;
import org.apache.activemq.command.ShutdownInfo; import org.apache.activemq.command.ShutdownInfo;
import org.apache.activemq.command.WireFormatInfo; import org.apache.activemq.command.WireFormatInfo;
@ -300,7 +302,7 @@ public class DemandForwardingBridge implements Bridge {
} }
} }
protected void serviceLocalException(IOException error) { protected void serviceLocalException(Throwable error) {
log.info("Network connection between " + localBroker + " and " + remoteBroker + " shutdown: "+error.getMessage(), error); log.info("Network connection between " + localBroker + " and " + remoteBroker + " shutdown: "+error.getMessage(), error);
ServiceSupport.dispose(this); ServiceSupport.dispose(this);
} }
@ -347,8 +349,16 @@ public class DemandForwardingBridge implements Bridge {
if( trace ) if( trace )
log.trace("bridging " + localBroker + " -> " + remoteBroker + ": "+message); log.trace("bridging " + localBroker + " -> " + remoteBroker + ": "+message);
if (!message.isPersistent() || !sub.remoteInfo.isDurable()){
remoteBroker.oneway( message ); remoteBroker.oneway( message );
}else{
Response response = remoteBroker.request(message);
if (response.isException()) {
ExceptionResponse er = (ExceptionResponse) response;
serviceLocalException(er.getException());
}
}
localBroker.oneway(new MessageAck(md, MessageAck.STANDARD_ACK_TYPE, demandConsumerDispatched)); localBroker.oneway(new MessageAck(md, MessageAck.STANDARD_ACK_TYPE, demandConsumerDispatched));
} }
@ -373,7 +383,7 @@ public class DemandForwardingBridge implements Bridge {
log.warn("Unexpected local command: "+command); log.warn("Unexpected local command: "+command);
} }
} }
} catch (IOException e) { } catch (Exception e) {
serviceLocalException(e); serviceLocalException(e);
} }
} }