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

Make amq 5.x compatible again with 4.x clients 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@580382 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Hiram R. Chirino 2007-09-28 15:09:33 +00:00
parent 19a83ae3cb
commit 3d024d6b4f
2 changed files with 12 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java
View File

@ -637,6 +637,11 @@ public class TransportConnection implements Service, Connection, Task, CommandVi
}
public Response processAddConnection(ConnectionInfo info) throws Exception {
// Older clients should have been defaulting this field to true.. but they were not.
if( wireFormatInfo!=null && wireFormatInfo.getVersion() <= 2 ) {
info.setClientMaster(true);
}
TransportConnectionState state;

12
activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java
View File

@ -48,6 +48,7 @@ import org.apache.activemq.command.Message;
import org.apache.activemq.command.MessageAck;
import org.apache.activemq.command.MessageId;
import org.apache.activemq.command.ProducerAck;
import org.apache.activemq.command.ProducerInfo;
import org.apache.activemq.command.Response;
import org.apache.activemq.filter.BooleanExpression;
import org.apache.activemq.filter.MessageEvaluationContext;
@ -349,12 +350,13 @@ public class Queue implements Destination, Task {
// There is delay between the client sending it and it arriving at the
// destination.. it may have expired.
final boolean sendProducerAck = (!message.isResponseRequired() || producerExchange.getProducerState().getInfo().getWindowSize() > 0) && !context.isInRecoveryMode();
final ProducerInfo producerInfo = producerExchange.getProducerState().getInfo();
final boolean sendProducerAck = !message.isResponseRequired() && producerInfo.getWindowSize() > 0 && !context.isInRecoveryMode();
if (message.isExpired()) {
broker.messageExpired(context, message);
destinationStatistics.getMessages().decrement();
if (sendProducerAck) {
ProducerAck ack = new ProducerAck(producerExchange.getProducerState().getInfo().getProducerId(), message.getSize());
ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message.getSize());
context.getConnection().dispatchAsync(ack);
}
return;
@ -367,7 +369,7 @@ public class Queue implements Destination, Task {
// We can avoid blocking due to low usage if the producer is sending
// a sync message or
// if it is using a producer window
if (producerExchange.getProducerState().getInfo().getWindowSize() > 0 || message.isResponseRequired()) {
if (producerInfo.getWindowSize() > 0 || message.isResponseRequired()) {
synchronized (messagesWaitingForSpace) {
messagesWaitingForSpace.add(new Runnable() {
public void run() {
@ -384,7 +386,7 @@ public class Queue implements Destination, Task {
}
if (sendProducerAck) {
ProducerAck ack = new ProducerAck(producerExchange.getProducerState().getInfo().getProducerId(), message.getSize());
ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message.getSize());
context.getConnection().dispatchAsync(ack);
} else {
Response response = new Response();
@ -435,7 +437,7 @@ public class Queue implements Destination, Task {
}
doMessageSend(producerExchange, message);
if (sendProducerAck) {
ProducerAck ack = new ProducerAck(producerExchange.getProducerState().getInfo().getProducerId(), message.getSize());
ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message.getSize());
context.getConnection().dispatchAsync(ack);
}
}