You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ch...@apache.org on 2007/10/16 18:22:29 UTC
svn commit: r585187 - in
/activemq/trunk/activemq-core/src/main/java/org/apache/activemq:
advisory/AdvisoryBroker.java broker/region/Topic.java
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 {