Added the TOUCH command to NSQMessage to allow long running consume…

…rs to notify nsqd that they are still working on a message so it won't be requeued.

pom.xml

@@ -6,7 +6,7 @@
 
   <groupId>com.github.brainlag</groupId>
   <artifactId>nsq-client</artifactId>
-  <version>1.0.0.RC2</version>
+  <version>1.0.0.RC3</version>
 
   <name>JavaNSQClient</name>
   <description>Fast Java client for NSQ.</description>
@@ -172,8 +172,8 @@
 
     <dependency>
       <groupId>io.netty</groupId>
-      <artifactId>netty-tcnative</artifactId>
-      <version>1.1.33.Fork11</version>
+      <artifactId>netty-tcnative-boringssl-static</artifactId>
+      <version>1.1.33.Fork16</version>
       <classifier>${os.detected.classifier}</classifier>
       <scope>test</scope>
     </dependency>

src/main/java/com/github/brainlag/nsq/NSQMessage.java

@@ -24,6 +24,14 @@ public void finished() {
         }
     }
 
+    public void touch() {
+        try {
+            connection.command(NSQCommand.instance("TOUCH " + new String(id, "ascii")));
+        } catch (UnsupportedEncodingException e) {
+            LogManager.getLogger(this).error("ASCII charset is not supported by your JVM?", e);
+        }
+    }
+
     /**
      * indicates a problem with processing, puts it back on the queue.
      */

src/test/java/com/github/brainlag/nsq/NSQConsumerTest.java

@@ -0,0 +1,60 @@
+package com.github.brainlag.nsq;
+
+import com.github.brainlag.nsq.exceptions.NSQException;
+import com.github.brainlag.nsq.lookup.DefaultNSQLookup;
+import com.github.brainlag.nsq.lookup.NSQLookup;
+import org.apache.logging.log4j.LogManager;
+import org.junit.Test;
+
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.assertTrue;
+
+public class NSQConsumerTest {
+
+    //duration to wait before auto-requeing a message setting in nsqd, defined with -msg-timeout:
+    //Set your timeout -msg-timeout="5s" command line when starting nsqd or changed this constant
+    //to the default of 60000.
+    private static final long NSQ_MSG_TIMEOUT = 5000;
+
+    @Test
+    public void testLongRunningConsumer() throws NSQException, TimeoutException, InterruptedException {
+        AtomicInteger counter = new AtomicInteger(0);
+        NSQLookup lookup = new DefaultNSQLookup();
+        lookup.addLookupAddress("localhost", 4161);
+
+        NSQConsumer consumer = new NSQConsumer(lookup, "test1", "testconsumer", (message) -> {
+            LogManager.getLogger(this).info("Processing message: " + new String(message.getMessage()));
+            counter.incrementAndGet();
+
+            long sleepTime = NSQ_MSG_TIMEOUT / 5;
+            for(int i = 0; i < 5; i++) {
+                try {
+                    Thread.sleep(sleepTime + 500);
+                } catch (InterruptedException e) {
+                    e.printStackTrace();
+                }
+                message.touch();
+            }
+            message.finished();
+            assertTrue( message.getAttempts() == 1);
+        });
+        consumer.start();
+
+        NSQProducer producer = new NSQProducer();
+        producer.addAddress("localhost", 4150);
+        producer.start();
+        String msg = "test-one-message";
+        producer.produce("test1", msg.getBytes());
+        producer.shutdown();
+
+        Thread.sleep(NSQ_MSG_TIMEOUT*2);
+        assertTrue(counter.get() == 1);
+        consumer.shutdown();
+    }
+
+
+
+
+}