Apache/activemq
1
0
Fork 0
mirror of https://github.com/apache/activemq.git synced 2025-03-01 13:49:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

Updated the flow control logic for the Topic case to match what we are using for the Queue case

Browse Source 
git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@585187 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Hiram R. Chirino 2007-10-16 16:22:08 +00:00
parent 8931aac98e
commit 9da7e31a33
2 changed files with 36 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
activemq-core/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java
View File

@ -38,6 +38,7 @@ import org.apache.activemq.command.DestinationInfo;
import org.apache.activemq.command.MessageId; import org.apache.activemq.command.MessageId;
import org.apache.activemq.command.ProducerId; import org.apache.activemq.command.ProducerId;
import org.apache.activemq.command.ProducerInfo; import org.apache.activemq.command.ProducerInfo;
import org.apache.activemq.state.ProducerState;
import org.apache.activemq.util.IdGenerator; import org.apache.activemq.util.IdGenerator;
import org.apache.activemq.util.LongSequenceGenerator; import org.apache.activemq.util.LongSequenceGenerator;
import org.apache.commons.logging.Log; import org.apache.commons.logging.Log;
@ -286,6 +287,7 @@ public class AdvisoryBroker extends BrokerFilter {
final ProducerBrokerExchange producerExchange = new ProducerBrokerExchange(); final ProducerBrokerExchange producerExchange = new ProducerBrokerExchange();
producerExchange.setConnectionContext(context); producerExchange.setConnectionContext(context);
producerExchange.setMutable(true); producerExchange.setMutable(true);
producerExchange.setProducerState(new ProducerState(new ProducerInfo()));
try { try {
context.setProducerFlowControl(false); context.setProducerFlowControl(false);
next.send(producerExchange, advisoryMessage); next.send(producerExchange, advisoryMessage);

45
activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java
View File

@ -42,6 +42,8 @@ import org.apache.activemq.command.Message;
import org.apache.activemq.command.MessageAck; import org.apache.activemq.command.MessageAck;
import org.apache.activemq.command.MessageId; import org.apache.activemq.command.MessageId;
import org.apache.activemq.command.ProducerAck; 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.command.SubscriptionInfo;
import org.apache.activemq.filter.MessageEvaluationContext; import org.apache.activemq.filter.MessageEvaluationContext;
import org.apache.activemq.store.MessageRecoveryListener; import org.apache.activemq.store.MessageRecoveryListener;
@ -279,13 +281,16 @@ public class Topic implements Destination {
public void send(final ProducerBrokerExchange producerExchange, final Message message) throws Exception { public void send(final ProducerBrokerExchange producerExchange, final Message message) throws Exception {
final ConnectionContext context = producerExchange.getConnectionContext(); 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 // There is delay between the client sending it and it arriving at the
// destination.. it may have expired. // destination.. it may have expired.
if (broker.isExpired(message)) { if (broker.isExpired(message)) {
broker.messageExpired(context, message); broker.messageExpired(context, message);
destinationStatistics.getMessages().decrement(); destinationStatistics.getMessages().decrement();
if ((!message.isResponseRequired() || producerExchange.getProducerState().getInfo().getWindowSize() > 0) && !context.isInRecoveryMode()) { if (sendProducerAck) {
ProducerAck ack = new ProducerAck(producerExchange.getProducerState().getInfo().getProducerId(), message.getSize()); ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message.getSize());
context.getConnection().dispatchAsync(ack); context.getConnection().dispatchAsync(ack);
} }
return; return;
@ -299,33 +304,39 @@ public class Topic implements Destination {
// We can avoid blocking due to low usage if the producer is sending // We can avoid blocking due to low usage if the producer is sending
// a sync message or // a sync message or
// if it is using a producer window // if it is using a producer window
if (producerExchange.getProducerState().getInfo().getWindowSize() > 0 || message.isResponseRequired()) { if (producerInfo.getWindowSize() > 0 || message.isResponseRequired()) {
synchronized (messagesWaitingForSpace) { synchronized (messagesWaitingForSpace) {
messagesWaitingForSpace.add(new Runnable() { messagesWaitingForSpace.add(new Runnable() {
public void run() { public void run() {
// While waiting for space to free up... the message try {
// may have expired.
// While waiting for space to free up... the
// message may have expired.
if (broker.isExpired(message)) { if (broker.isExpired(message)) {
broker.messageExpired(context, message); broker.messageExpired(context, message);
destinationStatistics.getMessages().decrement(); destinationStatistics.getMessages().decrement();
} else {
if (!message.isResponseRequired() && !context.isInRecoveryMode()) {
ProducerAck ack = new ProducerAck(producerExchange.getProducerState().getInfo().getProducerId(), message.getSize());
context.getConnection().dispatchAsync(ack);
}
return;
}
try {
doMessageSend(producerExchange, message); doMessageSend(producerExchange, message);
}
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);
}
} catch (Exception e) { } catch (Exception e) {
if (message.isResponseRequired() && !context.isInRecoveryMode()) { if (!sendProducerAck && !context.isInRecoveryMode()) {
ExceptionResponse response = new ExceptionResponse(e); ExceptionResponse response = new ExceptionResponse(e);
response.setCorrelationId(message.getCommandId()); response.setCorrelationId(message.getCommandId());
context.getConnection().dispatchAsync(response); context.getConnection().dispatchAsync(response);
} }
} }
} }
}); });
@ -362,6 +373,10 @@ public class Topic implements Destination {
} }
doMessageSend(producerExchange, message); 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 { void doMessageSend(final ProducerBrokerExchange producerExchange, final Message message) throws IOException, Exception {