Author: gtully
Date: Thu Oct 16 09:48:52 2008
New Revision: 705281

URL: http://svn.apache.org/viewvc?rev=705281&view=rev
Log:
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

Modified:
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/QueueMemoryFullMultiBrokersTest.java

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java?rev=705281&r1=705280&r2=705281&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java Thu Oct 16 09:48:52 2008
@@ -411,40 +411,44 @@
      * @param lastAckedMsg
      * @throws JMSException if it does not match
      */
-	protected void assertAckMatchesDispatched(MessageAck ack)
-			throws JMSException {
+	protected void assertAckMatchesDispatched(MessageAck ack) throws JMSException {
         MessageId firstAckedMsg = ack.getFirstMessageId();
-		MessageId lastAckedMsg = ack.getLastMessageId();
+        MessageId lastAckedMsg = ack.getLastMessageId();
+        int checkCount = 0;
+        boolean checkFoundStart = false;
+        boolean checkFoundEnd = false;
+        for (MessageReference node : dispatched) {
+
+            if (firstAckedMsg == null) {
+                checkFoundStart = true;
+            } else if (!checkFoundStart && firstAckedMsg.equals(node.getMessageId())) {
+                checkFoundStart = true;
+            }
+
+            if (checkFoundStart) {
+                checkCount++;
+            }
 
-		int checkCount = 0;
-		boolean checkFoundStart = false;
-		boolean checkFoundEnd = false;
-		for (MessageReference node : dispatched) {
-			
-			if( firstAckedMsg == null ) {
-				checkFoundStart=true;
-			} else if (!checkFoundStart && firstAckedMsg.equals(node.getMessageId())) {
-				checkFoundStart = true;
-			}
-
-			if (checkFoundStart) {
-				checkCount++;
-			}
-
-			if (lastAckedMsg != null && lastAckedMsg.equals(node.getMessageId())) {
-				checkFoundEnd = true;
-				break;
-			}
-		}
-		if (!checkFoundStart && firstAckedMsg != null)
-			throw new JMSException("Unmatched acknowledege: Could not find Message-ID "+firstAckedMsg+" in dispatched-list (start of ack)");
-		if (!checkFoundEnd && lastAckedMsg != null)
-			throw new JMSException("Unmatched acknowledege: Could not find Message-ID "+lastAckedMsg+" in dispatched-list (end of ack)");
-		if (ack.getMessageCount() != checkCount) {
-			throw new JMSException("Unmatched acknowledege: Expected message count ("+ack.getMessageCount()+
-					") differs from count in dispatched-list ("+checkCount+")");
-		}
-	}
+            if (lastAckedMsg != null && lastAckedMsg.equals(node.getMessageId())) {
+                checkFoundEnd = true;
+                break;
+            }
+        }
+        if (!checkFoundStart && firstAckedMsg != null)
+            throw new JMSException("Unmatched acknowledege: " + ack
+                    + "; Could not find Message-ID " + firstAckedMsg
+                    + " in dispatched-list (start of ack)");
+        if (!checkFoundEnd && lastAckedMsg != null)
+            throw new JMSException("Unmatched acknowledege: " + ack
+                    + "; Could not find Message-ID " + lastAckedMsg
+                    + " in dispatched-list (end of ack)");
+        if (ack.getMessageCount() != checkCount && ack.isStandardAck()) {
+            throw new JMSException("Unmatched acknowledege: " + ack
+                    + "; Expected message count (" + ack.getMessageCount()
+                    + ") differs from count in dispatched-list (" + checkCount
+                    + ")");
+        }
+    }
 
     /**
      * @param context

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java?rev=705281&r1=705280&r2=705281&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java Thu Oct 16 09:48:52 2008
@@ -647,7 +647,7 @@
                             else{
                               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();                          
 
                         } else {
@@ -664,7 +664,7 @@
                                             ExceptionResponse er = (ExceptionResponse)response;
                                             serviceLocalException(er.getException());
                                         } else {
-                                            localBroker.oneway(new MessageAck(md, MessageAck.STANDARD_ACK_TYPE, 1));
+                                            localBroker.oneway(new MessageAck(md, MessageAck.INDIVIDUAL_ACK_TYPE, 1));
                                             dequeueCounter.incrementAndGet();
                                         }
                                     } catch (IOException e) {

Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/QueueMemoryFullMultiBrokersTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/QueueMemoryFullMultiBrokersTest.java?rev=705281&r1=705280&r2=705281&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/QueueMemoryFullMultiBrokersTest.java (original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/QueueMemoryFullMultiBrokersTest.java Thu Oct 16 09:48:52 2008
@@ -39,7 +39,6 @@
     public static final int MESSAGE_COUNT = 2000;
    
     public void testQueueNetworkWithConsumerFull() throws Exception {
-        if (true) return;
         
         bridgeAllBrokers();
         startAllBrokers();
@@ -67,8 +66,6 @@
         assertTrue("All messages are consumed and acked from source:" + internalQueue, internalQueue.getMessages().isEmpty());
         assertEquals("messages source:" + internalQueue, 0, internalQueue.getDestinationStatistics().getMessages().getCount());
         assertEquals("inflight source:" + internalQueue, 0, internalQueue.getDestinationStatistics().getInflight().getCount());
-        
-
     }
 
     public void setUp() throws Exception {
@@ -82,7 +79,6 @@
         }
         BrokerService broker2 = brokers.get("Broker2").broker;
         applyMemoryLimitPolicy(broker2);
-        
     }
 
     private void applyMemoryLimitPolicy(BrokerService broker) {