[prev in list] [next in list] [prev in thread] [next in thread] 

List:       activemq-commits
Subject:    svn commit: r580382 - in
From:       chirino () apache ! org
Date:       2007-09-28 15:09:33
Message-ID: 20070928150934.4105F1A9832 () eris ! apache ! org
[Download RAW message or body]

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/apach \
e/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);  }
     }


[prev in list] [next in list] [prev in thread] [next in thread]