package forklift.integration;

import forklift.Forklift;
import forklift.connectors.ConnectorException;
import forklift.consumer.Consumer;
import forklift.exception.StartupException;
import forklift.producers.ForkliftProducerI;
import forklift.producers.ProducerException;
import org.junit.Test;

public class ConsumerActionTests extends BaseIntegrationTest {

    /**
     * Test subscribing to a topic, unsubscribing, then resubscribing to the topic.
     */
    @Test
    public void topicTurnoverTest() throws StartupException, ConnectorException, InterruptedException, ProducerException {
        Forklift forklift = serviceManager.newManagedForkliftInstance();
        ForkliftProducerI producer = forklift.getConnector().getQueueProducer("forklift-string-topic");
        sentMessageIds.add(producer.send("message1"));
        final Consumer c1 = new Consumer(StringConsumer.class, forklift);
        // Shutdown the consumer after all the messages have been processed.
        c1.setOutOfMessages((listener) -> {
            listener.shutdown();
        });
        // Start the consumer.
        c1.listen();
        Thread.sleep(3000);
        sentMessageIds.add(producer.send("message2"));
        final Consumer c2 = new Consumer(StringConsumer.class, forklift);
        // Shutdown the consumer after all the messages have been processed.
        c2.setOutOfMessages((listener) -> {
            listener.shutdown();
        });
        // Start the consumer.
        c2.listen();
        messageAsserts();
    }

}

@@ -885,32 +885,33 @@
 ass, forklift);%0A
+%0A
         // Shutd
@@ -1007,32 +1007,148 @@
 (listener) -%3E %7B%0A
+            timeouts++;%0A%0A            if (sentMessageIds.equals(consumedMessageIds) %7C%7C timeouts %3E maxTimeouts) %7B%0A    
             list
@@ -1156,32 +1156,46 @@
 ner.shutdown();%0A
+            %7D%0A
         %7D);%0A    
@@ -1246,36 +1246,9 @@
 ();%0A
-        Thread.sleep(3000);
 %0A
+
     
@@ -1372,16 +1372,17 @@
 klift);%0A
+%0A
         
@@ -1442,32 +1442,54 @@
 been processed.%0A
+        timeouts = 0;%0A
         c2.setOu
@@ -1508,32 +1508,148 @@
 (listener) -%3E %7B%0A
+            timeouts++;%0A%0A            if (sentMessageIds.equals(consumedMessageIds) %7C%7C timeouts %3E maxTimeouts) %7B%0A    
             list
@@ -1657,32 +1657,46 @@
 ner.shutdown();%0A
+            %7D%0A
         %7D);%0A    

Fix a test that should've been fixed long ago with the others