https://jira.jboss.org/browse/HORNETQ-410

hornetq · Jun 8, 2010 · 07e3ac6 · 07e3ac6
1 parent 71c1fa3
commit 07e3ac6
Show file tree

Hide file tree

Showing 2 changed files with 43 additions and 37 deletions.
diff --git a/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java b/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java
@@ -453,13 +453,16 @@ public void setTransferring(final boolean transferring)
    }
 
    public void receiveCredits(final int credits) throws Exception
-   {
+   {      
       if (credits == -1)
       {
          // No flow control
          availableCredits = null;
+
+         //There may be messages already in the queue
+         promptDelivery();
       }
-      else if(credits == 0)
+      else if (credits == 0)
       {
          //reset, used on slow consumers
          availableCredits.set(0);

diff --git a/tests/src/org/hornetq/tests/integration/client/ConsumerTest.java b/tests/src/org/hornetq/tests/integration/client/ConsumerTest.java
@@ -66,41 +66,7 @@ protected void tearDown() throws Exception
       super.tearDown();
    }
 
-//   public void testQueueSpin() throws Exception
-//   {
-//      ClientSessionFactory sf = createInVMFactory();
-//
-//      ClientSession session1 = sf.createSession();
-//      
-//      ClientSession session2 = sf.createSession();
-//
-//      session1.createQueue(QUEUE, QUEUE, null, false);
-//
-//      ClientProducer producer = session1.createProducer(QUEUE);
-//
-//      final int numMessages = 100;
-//
-//      for (int i = 0; i < numMessages; i++)
-//      {
-//         ClientMessage message = createTextMessage("m" + i, session1);
-//         producer.send(message);
-//      }
-//
-//      ClientConsumer consumer1 = session1.createConsumer(QUEUE);
-//      
-//      ClientConsumer consumer2 = session2.createConsumer(QUEUE, new SimpleString("foo=wibble"));
-//      
-//      session2.start();
-//      
-//      consumer2.receive();
-//      
-//      Thread.sleep(30000);
-//      
-//      session1.close();
-//      
-//      session2.close();
-//   }
-
+
    public void testConsumerAckImmediateAutoCommitTrue() throws Exception
    {
       ClientSessionFactory sf = createInVMFactory();
@@ -301,6 +267,43 @@ public void onMessage(final ClientMessage message)
       Assert.assertTrue(latch.await(5, TimeUnit.SECONDS));
       sessionRec.close();
    }
+
+   // https://jira.jboss.org/browse/HORNETQ-410
+   public void testConsumeWithNoConsumerFlowControl() throws Exception
+   {
+      ClientSessionFactory sf = createInVMFactory();
+
+      sf.setConsumerWindowSize(-1);
+
+      ClientSession session = sf.createSession(false, true, true);
+
+      session.createQueue(QUEUE, QUEUE, null, false);
+
+      session.start();
+
+      ClientProducer producer = session.createProducer(QUEUE);
+
+      final int numMessages = 100;
+
+      for (int i = 0; i < numMessages; i++)
+      {
+         ClientMessage message = createTextMessage("m" + i, session);
+         producer.send(message);
+      }
+
+      ClientConsumer consumer = session.createConsumer(QUEUE);
+
+      for (int i = 0; i < numMessages; i++)
+      {
+         ClientMessage message = consumer.receive(10000);
+         assertNotNull(message);
+         message.acknowledge();
+      }
+
+      session.close();
+      sf.close();
+
+   }
 
    public void testClearListener() throws Exception
    {