activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From gtu...@apache.org
Subject [1/3] activemq git commit: https://issues.apache.org/jira/browse/AMQ-5513 - have lastDeliveredSequence -1 indicate nothing was received, respect this value in queue sub remove and durable sub deactivate. 0 still constitutes no info - so old clients see s
Date Fri, 09 Jan 2015 13:32:34 GMT
Repository: activemq
Updated Branches:
  refs/heads/trunk 7948d6905 -> dbb1d8b83


https://issues.apache.org/jira/browse/AMQ-5513 - have lastDeliveredSequence -1 indicate nothing
was received, respect this value in queue sub remove and durable sub deactivate. 0 still constitutes
no info - so old clients see same behaviour as before


Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/dbb1d8b8
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/dbb1d8b8
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/dbb1d8b8

Branch: refs/heads/trunk
Commit: dbb1d8b83d2ee5c5a38c55834a70cd21a2dd2ad0
Parents: 05f6cd6
Author: gtully <gary.tully@gmail.com>
Authored: Fri Jan 9 13:28:58 2015 +0000
Committer: gtully <gary.tully@gmail.com>
Committed: Fri Jan 9 13:30:59 2015 +0000

----------------------------------------------------------------------
 .../broker/region/DurableTopicSubscription.java | 14 +++---
 .../apache/activemq/broker/region/Queue.java    |  4 +-
 .../apache/activemq/broker/region/Topic.java    |  2 +-
 .../activemq/broker/region/TopicRegion.java     |  4 +-
 .../activemq/ActiveMQMessageConsumer.java       |  2 +-
 .../org/apache/activemq/JmsRedeliveredTest.java | 47 ++++++++++++++++++++
 6 files changed, 61 insertions(+), 12 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/activemq/blob/dbb1d8b8/activemq-broker/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java
index 4c19c62..8df2819 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java
@@ -184,7 +184,7 @@ public class DurableTopicSubscription extends PrefetchSubscription implements
Us
         }
     }
 
-    public void deactivate(boolean keepDurableSubsActive) throws Exception {
+    public void deactivate(boolean keepDurableSubsActive, long lastDeliveredSequenceId) throws
Exception {
         LOG.debug("Deactivating keepActive={}, {}", keepDurableSubsActive, this);
         active.set(false);
         offlineTimestamp.set(System.currentTimeMillis());
@@ -214,11 +214,13 @@ public class DurableTopicSubscription extends PrefetchSubscription implements
Us
 
                 for (final MessageReference node : dispatched) {
                     // Mark the dispatched messages as redelivered for next time.
-                    Integer count = redeliveredMessages.get(node.getMessageId());
-                    if (count != null) {
-                        redeliveredMessages.put(node.getMessageId(), Integer.valueOf(count.intValue()
+ 1));
-                    } else {
-                        redeliveredMessages.put(node.getMessageId(), Integer.valueOf(1));
+                    if (lastDeliveredSequenceId == 0 || (lastDeliveredSequenceId > 0 &&
node.getMessageId().getBrokerSequenceId() <= lastDeliveredSequenceId)) {
+                        Integer count = redeliveredMessages.get(node.getMessageId());
+                        if (count != null) {
+                            redeliveredMessages.put(node.getMessageId(), Integer.valueOf(count.intValue()
+ 1));
+                        } else {
+                            redeliveredMessages.put(node.getMessageId(), Integer.valueOf(1));
+                        }
                     }
                     if (keepDurableSubsActive && pending.isTransient()) {
                         pending.addMessageFirst(node);

http://git-wip-us.apache.org/repos/asf/activemq/blob/dbb1d8b8/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java b/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java
index f5f2efe..89b9081 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java
@@ -547,12 +547,12 @@ public class Queue extends BaseDestination implements Task, UsageListener,
Index
                 List<MessageReference> unAckedMessages = sub.remove(context, this);
 
                 // locate last redelivered in unconsumed list (list in delivery rather than
seq order)
-                if (lastDeiveredSequenceId != 0) {
+                if (lastDeiveredSequenceId > 0) {
                     for (MessageReference ref : unAckedMessages) {
                         if (ref.getMessageId().getBrokerSequenceId() == lastDeiveredSequenceId)
{
                             lastDeliveredRef = ref;
                             markAsRedelivered = true;
-                            LOG.debug("found lastDeliveredSeqID: {}, message reference: {}",
lastDeiveredSequenceId, ref.getMessageId());
+                            LOG.error("found lastDeliveredSeqID: {}, message reference: {}",
lastDeiveredSequenceId, ref.getMessageId());
                             break;
                         }
                     }

http://git-wip-us.apache.org/repos/asf/activemq/blob/dbb1d8b8/activemq-broker/src/main/java/org/apache/activemq/broker/region/Topic.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/Topic.java b/activemq-broker/src/main/java/org/apache/activemq/broker/region/Topic.java
index cd144c3..eff9619 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/Topic.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/Topic.java
@@ -203,7 +203,7 @@ public class Topic extends BaseDestination implements Task {
             if (removed != null) {
                 destinationStatistics.getConsumers().decrement();
                 // deactivate and remove
-                removed.deactivate(false);
+                removed.deactivate(false, 0l);
                 consumers.remove(removed);
             }
         }

http://git-wip-us.apache.org/repos/asf/activemq/blob/dbb1d8b8/activemq-broker/src/main/java/org/apache/activemq/broker/region/TopicRegion.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/TopicRegion.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/TopicRegion.java
index cea5bb7..383f240 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/TopicRegion.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/TopicRegion.java
@@ -155,7 +155,7 @@ public class TopicRegion extends AbstractRegion {
                     if ((sub.context != context) || (sub.info != info)) {
                         sub.info = info;
                         sub.context = context;
-                        sub.deactivate(keepDurableSubsActive);
+                        sub.deactivate(keepDurableSubsActive, info.getLastDeliveredSequenceId());
                     }
                     subscriptions.put(info.getConsumerId(), sub);
                 }
@@ -185,7 +185,7 @@ public class TopicRegion extends AbstractRegion {
                 // as what is in the sub. otherwise, during linksteal
                 // sub will get new context, but will be removed here
                 if (sub.getContext() == context)
-                    sub.deactivate(keepDurableSubsActive);
+                    sub.deactivate(keepDurableSubsActive, info.getLastDeliveredSequenceId());
             }
         } else {
             super.removeConsumer(context, info);

http://git-wip-us.apache.org/repos/asf/activemq/blob/dbb1d8b8/activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
----------------------------------------------------------------------
diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
b/activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
index f43b56d..0808ead 100755
--- a/activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
+++ b/activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
@@ -147,7 +147,7 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer,
StatsC
     AtomicInteger inProgressClearRequiredFlag = new AtomicInteger(0);
 
     private MessageAck pendingAck;
-    private long lastDeliveredSequenceId;
+    private long lastDeliveredSequenceId = -1;
 
     private IOException failureError;
 

http://git-wip-us.apache.org/repos/asf/activemq/blob/dbb1d8b8/activemq-unit-tests/src/test/java/org/apache/activemq/JmsRedeliveredTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/JmsRedeliveredTest.java
b/activemq-unit-tests/src/test/java/org/apache/activemq/JmsRedeliveredTest.java
index 63c5911..72a1a28 100755
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/JmsRedeliveredTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/JmsRedeliveredTest.java
@@ -16,6 +16,7 @@
  */
 package org.apache.activemq;
 
+import java.util.concurrent.TimeUnit;
 import javax.jms.Connection;
 import javax.jms.DeliveryMode;
 import javax.jms.Destination;
@@ -400,6 +401,52 @@ public class JmsRedeliveredTest extends TestCase {
         session.close();
     }
 
+    public void testNoReceiveConsumerDoesNotIncrementRedelivery() throws Exception {
+        connection.setClientID(getName());
+        connection.start();
+
+        Session session = connection.createSession(true, Session.CLIENT_ACKNOWLEDGE);
+        Queue queue = session.createQueue("queue-" + getName());
+        MessageConsumer consumer = session.createConsumer(queue);
+
+        MessageProducer producer = createProducer(session, queue);
+        producer.send(createTextMessage(session));
+        session.commit();
+
+        TimeUnit.SECONDS.sleep(1);
+        consumer.close();
+
+        consumer = session.createConsumer(queue);
+        Message msg = consumer.receive(1000);
+        assertNotNull(msg);
+
+        assertFalse("Message should not be redelivered.", msg.getJMSRedelivered());
+        session.close();
+    }
+
+    public void testNoReceiveDurableConsumerDoesNotIncrementRedelivery() throws Exception
{
+        connection.setClientID(getName());
+        connection.start();
+
+        Session session = connection.createSession(true, Session.CLIENT_ACKNOWLEDGE);
+        Topic topic = session.createTopic("topic-" + getName());
+        MessageConsumer consumer = session.createDurableSubscriber(topic, "sub");
+
+        MessageProducer producer = createProducer(session, topic);
+        producer.send(createTextMessage(session));
+        session.commit();
+
+        TimeUnit.SECONDS.sleep(1);
+        consumer.close();
+
+        consumer = session.createDurableSubscriber(topic, "sub");
+        Message msg = consumer.receive(1000);
+        assertNotNull(msg);
+
+        assertFalse("Message should not be redelivered.", msg.getJMSRedelivered());
+        session.close();
+    }
+
     /**
      * Creates a text message.
      * 


Mime
View raw message