svn commit: r580382 - in /activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker: TransportConnection.java region/Queue.java

[ch...@apache.org]

Author: chirino
Date: Fri Sep 28 08:09:33 2007
New Revision: 580382

URL: http://svn.apache.org/viewvc?rev=580382&view=rev
Log:
Make amq 5.x compatible again with 4.x clients

Modified:
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java?rev=580382&r1=580381&r2=580382&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java Fri Sep 28 08:09:33 2007
@@ -637,6 +637,11 @@
     }
 
     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;
 

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java?rev=580382&r1=580381&r2=580382&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java Fri Sep 28 08:09:33 2007
@@ -48,6 +48,7 @@
 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 @@
         // 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 @@
             // 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 @@
                                 }
 
                                 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 @@
         }
         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);
         }
     }