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

https://issues.apache.org/jira/browse/AMQ-3331 - make bridge alwaysSyncSend=true the default 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1441545 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Gary Tully 2013-02-01 17:21:59 +00:00
parent 6028ec4d35
commit f383ca1de4
2 changed files with 15 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
View File

@ -965,25 +965,12 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
+ message.getDestination() + ", brokerPath: " + Arrays.toString(message.getBrokerPath()) + ", message: " + message);
}
if (!configuration.isAlwaysSyncSend() && !message.isPersistent()) {
// If the message was originally sent using async send, we will
// preserve that QOS by bridging it using an async send (small chance
// of message loss).
try {
remoteBroker.oneway(message);
localBroker.oneway(new MessageAck(md, MessageAck.INDIVIDUAL_ACK_TYPE, 1));
dequeueCounter.incrementAndGet();
} finally {
sub.decrementOutstandingResponses();
}
} else {
if (message.isPersistent() || configuration.isAlwaysSyncSend()) {
// The message was not sent using async send, so we should only
// ack the local broker when we get confirmation that the remote
// broker has received the message.
ResponseCallback callback = new ResponseCallback() {
remoteBroker.asyncRequest(message, new ResponseCallback() {
@Override
public void onCompletion(FutureResponse future) {
try {
@ -1001,9 +988,19 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
sub.decrementOutstandingResponses();
}
}
};
});
remoteBroker.asyncRequest(message, callback);
} else {
// If the message was originally sent using async send, we will
// preserve that QOS by bridging it using an async send (small chance
// of message loss).
try {
remoteBroker.oneway(message);
localBroker.oneway(new MessageAck(md, MessageAck.INDIVIDUAL_ACK_TYPE, 1));
dequeueCounter.incrementAndGet();
} finally {
sub.decrementOutstandingResponses();
}
}
} else {
if (LOG.isDebugEnabled()) {

2
activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeConfiguration.java
View File

@ -54,7 +54,7 @@ public class NetworkBridgeConfiguration {
private boolean suppressDuplicateQueueSubscriptions = false;
private boolean suppressDuplicateTopicSubscriptions = true;
private boolean alwaysSyncSend = false;
private boolean alwaysSyncSend = true;
private boolean staticBridge = false;
private boolean useCompression = false;
private boolean advisoryForFailedForward = false;