activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From chir...@apache.org
Subject svn commit: r585187 - in /activemq/trunk/activemq-core/src/main/java/org/apache/activemq: advisory/AdvisoryBroker.java broker/region/Topic.java
Date Tue, 16 Oct 2007 16:22:29 GMT
Author: chirino
Date: Tue Oct 16 09:22:08 2007
New Revision: 585187

URL: http://svn.apache.org/viewvc?rev=585187&view=rev
Log:
Updated the flow control logic for the Topic case to match what we are using for the Queue
case

Modified:
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java?rev=585187&r1=585186&r2=585187&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java
Tue Oct 16 09:22:08 2007
@@ -38,6 +38,7 @@
 import org.apache.activemq.command.MessageId;
 import org.apache.activemq.command.ProducerId;
 import org.apache.activemq.command.ProducerInfo;
+import org.apache.activemq.state.ProducerState;
 import org.apache.activemq.util.IdGenerator;
 import org.apache.activemq.util.LongSequenceGenerator;
 import org.apache.commons.logging.Log;
@@ -286,6 +287,7 @@
             final ProducerBrokerExchange producerExchange = new ProducerBrokerExchange();
             producerExchange.setConnectionContext(context);
             producerExchange.setMutable(true);
+            producerExchange.setProducerState(new ProducerState(new ProducerInfo()));
             try {
                 context.setProducerFlowControl(false);
                 next.send(producerExchange, advisoryMessage);

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java?rev=585187&r1=585186&r2=585187&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java
Tue Oct 16 09:22:08 2007
@@ -42,6 +42,8 @@
 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.command.SubscriptionInfo;
 import org.apache.activemq.filter.MessageEvaluationContext;
 import org.apache.activemq.store.MessageRecoveryListener;
@@ -279,13 +281,16 @@
     public void send(final ProducerBrokerExchange producerExchange, final Message message)
throws Exception {
         final ConnectionContext context = producerExchange.getConnectionContext();
 
+        final ProducerInfo producerInfo = producerExchange.getProducerState().getInfo();
+        final boolean sendProducerAck = !message.isResponseRequired() && producerInfo.getWindowSize()
> 0 && !context.isInRecoveryMode();
+
         // There is delay between the client sending it and it arriving at the
         // destination.. it may have expired.
         if (broker.isExpired(message)) {
             broker.messageExpired(context, message);
             destinationStatistics.getMessages().decrement();
-            if ((!message.isResponseRequired() || producerExchange.getProducerState().getInfo().getWindowSize()
> 0) && !context.isInRecoveryMode()) {
-                ProducerAck ack = new ProducerAck(producerExchange.getProducerState().getInfo().getProducerId(),
message.getSize());
+            if (sendProducerAck) {
+                ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message.getSize());
                 context.getConnection().dispatchAsync(ack);
             }
             return;
@@ -299,33 +304,39 @@
             // 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() {
+                            
+                            try {
 
-                            // While waiting for space to free up... the message
-                            // may have expired.
-                            if (broker.isExpired(message)) {
-                                broker.messageExpired(context, message);
-                                destinationStatistics.getMessages().decrement();
+                                // While waiting for space to free up... the
+                                // message may have expired.
+                                if (broker.isExpired(message)) {
+                                    broker.messageExpired(context, message);
+                                    destinationStatistics.getMessages().decrement();
+                                } else {
+                                    doMessageSend(producerExchange, message);
+                                }
 
-                                if (!message.isResponseRequired() && !context.isInRecoveryMode())
{
-                                    ProducerAck ack = new ProducerAck(producerExchange.getProducerState().getInfo().getProducerId(),
message.getSize());
+                                if (sendProducerAck) {
+                                    ProducerAck ack = new ProducerAck(producerInfo.getProducerId(),
message.getSize());
                                     context.getConnection().dispatchAsync(ack);
+                                } else {
+                                    Response response = new Response();
+                                    response.setCorrelationId(message.getCommandId());
+                                    context.getConnection().dispatchAsync(response);
                                 }
-                                return;
-                            }
 
-                            try {
-                                doMessageSend(producerExchange, message);
                             } catch (Exception e) {
-                                if (message.isResponseRequired() && !context.isInRecoveryMode())
{
+                                if (!sendProducerAck && !context.isInRecoveryMode())
{
                                     ExceptionResponse response = new ExceptionResponse(e);
                                     response.setCorrelationId(message.getCommandId());
                                     context.getConnection().dispatchAsync(response);
                                 }
                             }
+                            
                         }
                     });
 
@@ -362,6 +373,10 @@
         }
 
         doMessageSend(producerExchange, message);
+        if (sendProducerAck) {
+            ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message.getSize());
+            context.getConnection().dispatchAsync(ack);
+        }
     }
 
     void doMessageSend(final ProducerBrokerExchange producerExchange, final Message message)
throws IOException, Exception {



Mime
View raw message