Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

AMQ-1976, individual messages passing through a network bridge need an individual ack; for an individual ack, it is ok to have more than one message dispatched 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@705281 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Gary Tully 2008-10-16 16:48:52 +00:00
parent bc7e759662
commit fdee6e81e2
3 changed files with 36 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
View File

@ -411,18 +411,16 @@ public abstract class PrefetchSubscription extends AbstractSubscription {
* @param lastAckedMsg * @param lastAckedMsg
* @throws JMSException if it does not match * @throws JMSException if it does not match
*/ */
protected void assertAckMatchesDispatched(MessageAck ack) protected void assertAckMatchesDispatched(MessageAck ack) throws JMSException {
throws JMSException {
MessageId firstAckedMsg = ack.getFirstMessageId(); MessageId firstAckedMsg = ack.getFirstMessageId();
MessageId lastAckedMsg = ack.getLastMessageId(); MessageId lastAckedMsg = ack.getLastMessageId();
int checkCount = 0; int checkCount = 0;
boolean checkFoundStart = false; boolean checkFoundStart = false;
boolean checkFoundEnd = false; boolean checkFoundEnd = false;
for (MessageReference node : dispatched) { for (MessageReference node : dispatched) {
if( firstAckedMsg == null ) { if (firstAckedMsg == null) {
checkFoundStart=true; checkFoundStart = true;
} else if (!checkFoundStart && firstAckedMsg.equals(node.getMessageId())) { } else if (!checkFoundStart && firstAckedMsg.equals(node.getMessageId())) {
checkFoundStart = true; checkFoundStart = true;
} }
@ -437,12 +435,18 @@ public abstract class PrefetchSubscription extends AbstractSubscription {
} }
} }
if (!checkFoundStart && firstAckedMsg != null) if (!checkFoundStart && firstAckedMsg != null)
throw new JMSException("Unmatched acknowledege: Could not find Message-ID "+firstAckedMsg+" in dispatched-list (start of ack)"); throw new JMSException("Unmatched acknowledege: " + ack
+ "; Could not find Message-ID " + firstAckedMsg
+ " in dispatched-list (start of ack)");
if (!checkFoundEnd && lastAckedMsg != null) if (!checkFoundEnd && lastAckedMsg != null)
throw new JMSException("Unmatched acknowledege: Could not find Message-ID "+lastAckedMsg+" in dispatched-list (end of ack)"); throw new JMSException("Unmatched acknowledege: " + ack
if (ack.getMessageCount() != checkCount) { + "; Could not find Message-ID " + lastAckedMsg
throw new JMSException("Unmatched acknowledege: Expected message count ("+ack.getMessageCount()+ + " in dispatched-list (end of ack)");
") differs from count in dispatched-list ("+checkCount+")"); if (ack.getMessageCount() != checkCount && ack.isStandardAck()) {
throw new JMSException("Unmatched acknowledege: " + ack
+ "; Expected message count (" + ack.getMessageCount()
+ ") differs from count in dispatched-list (" + checkCount
+ ")");
} }
} }

4
activemq-core/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
View File

@ -647,7 +647,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge {
else{ else{
LOG.info("Message not forwarded on to remote, because message came from remote"); LOG.info("Message not forwarded on to remote, because message came from remote");
} }
localBroker.oneway(new MessageAck(md, MessageAck.STANDARD_ACK_TYPE, 1)); localBroker.oneway(new MessageAck(md, MessageAck.INDIVIDUAL_ACK_TYPE, 1));
dequeueCounter.incrementAndGet(); dequeueCounter.incrementAndGet();
} else { } else {
@ -664,7 +664,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge {
ExceptionResponse er = (ExceptionResponse)response; ExceptionResponse er = (ExceptionResponse)response;
serviceLocalException(er.getException()); serviceLocalException(er.getException());
} else { } else {
localBroker.oneway(new MessageAck(md, MessageAck.STANDARD_ACK_TYPE, 1)); localBroker.oneway(new MessageAck(md, MessageAck.INDIVIDUAL_ACK_TYPE, 1));
dequeueCounter.incrementAndGet(); dequeueCounter.incrementAndGet();
} }
} catch (IOException e) { } catch (IOException e) {

4
activemq-core/src/test/java/org/apache/activemq/usecases/QueueMemoryFullMultiBrokersTest.java
View File

@ -39,7 +39,6 @@ public class QueueMemoryFullMultiBrokersTest extends JmsMultipleBrokersTestSuppo
public static final int MESSAGE_COUNT = 2000; public static final int MESSAGE_COUNT = 2000;
public void testQueueNetworkWithConsumerFull() throws Exception { public void testQueueNetworkWithConsumerFull() throws Exception {
if (true) return;
bridgeAllBrokers(); bridgeAllBrokers();
startAllBrokers(); startAllBrokers();
@ -67,8 +66,6 @@ public class QueueMemoryFullMultiBrokersTest extends JmsMultipleBrokersTestSuppo
assertTrue("All messages are consumed and acked from source:" + internalQueue, internalQueue.getMessages().isEmpty()); assertTrue("All messages are consumed and acked from source:" + internalQueue, internalQueue.getMessages().isEmpty());
assertEquals("messages source:" + internalQueue, 0, internalQueue.getDestinationStatistics().getMessages().getCount()); assertEquals("messages source:" + internalQueue, 0, internalQueue.getDestinationStatistics().getMessages().getCount());
assertEquals("inflight source:" + internalQueue, 0, internalQueue.getDestinationStatistics().getInflight().getCount()); assertEquals("inflight source:" + internalQueue, 0, internalQueue.getDestinationStatistics().getInflight().getCount());
} }
public void setUp() throws Exception { public void setUp() throws Exception {
@ -82,7 +79,6 @@ public class QueueMemoryFullMultiBrokersTest extends JmsMultipleBrokersTestSuppo
} }
BrokerService broker2 = brokers.get("Broker2").broker; BrokerService broker2 = brokers.get("Broker2").broker;
applyMemoryLimitPolicy(broker2); applyMemoryLimitPolicy(broker2);
} }
private void applyMemoryLimitPolicy(BrokerService broker) { private void applyMemoryLimitPolicy(BrokerService broker) {