https://issues.jboss.org/browse/HORNETQ-465 - deleteMessages won't cl…

…ear lastValueQueue
clebertsuconic · Nov 8, 2013 · a78836c · a78836c
1 parent 933f80e
commit a78836c
Show file tree

Hide file tree

Showing 3 changed files with 44 additions and 1 deletion.
diff --git a/hornetq-server/src/main/java/org/hornetq/core/server/impl/LastValueQueue.java b/hornetq-server/src/main/java/org/hornetq/core/server/impl/LastValueQueue.java
@@ -155,6 +155,23 @@ public synchronized void addHead(final MessageReference ref)
       }
    }
 
+
+   @Override
+   protected void refRemoved(MessageReference ref)
+   {
+      synchronized (this)
+      {
+         SimpleString prop = ref.getMessage().getSimpleStringProperty(Message.HDR_LAST_VALUE_NAME);
+
+         if (prop != null)
+         {
+            map.remove(prop);
+         }
+      }
+
+      super.refRemoved(ref);
+   }
+
    private class HolderReference implements MessageReference
    {
       private final SimpleString prop;

diff --git a/hornetq-server/src/main/java/org/hornetq/core/server/impl/QueueImpl.java b/hornetq-server/src/main/java/org/hornetq/core/server/impl/QueueImpl.java
@@ -2061,7 +2061,7 @@ else if (status == HandleStatus.NO_MATCH)
    /**
     * @param ref
     */
-   private void refRemoved(MessageReference ref)
+   protected void refRemoved(MessageReference ref)
    {
       queueMemorySize.addAndGet(-ref.getMessageMemoryEstimate());
       if (ref.isPaged())

diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/server/LVQTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/server/LVQTest.java
@@ -498,6 +498,32 @@ public void testMultipleAcksPersistedCorrectly() throws Exception
       assertEquals(0, queue.getDeliveringCount());
    }
 
+   @Test
+   public void testRemoveMessageThroughManagement() throws Exception
+   {
+
+      Queue queue = server.locateQueue(qName1);
+      ClientProducer producer = clientSession.createProducer(address);
+      ClientConsumer consumer = clientSession.createConsumer(qName1);
+      SimpleString rh = new SimpleString("SMID1");
+      ClientMessage m1 = createTextMessage(clientSession, "m1");
+      m1.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
+      m1.setDurable(true);
+      producer.send(m1);
+
+      queue.deleteAllReferences();
+
+      producer.send(m1);
+
+      clientSession.start();
+      ClientMessage m = consumer.receive(1000);
+      Assert.assertNotNull(m);
+      m.acknowledge();
+      Assert.assertEquals(m.getBodyBuffer().readString(), "m1");
+
+      assertEquals(0, queue.getDeliveringCount());
+   }
+
    @Test
    public void testMultipleAcksPersistedCorrectly2() throws Exception
    {