Limit max in flight

Author: Eivind Siqveland Larsen

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

@@ -7,6 +7,7 @@
 
 import java.net.InetAddress;
 import java.net.UnknownHostException;
+import java.util.Optional;
 
 public class NSQConfig {
 
@@ -23,6 +24,7 @@ public enum Compression {NO_COMPRESSION, DEFLATE, SNAPPY}
     private Compression compression = Compression.NO_COMPRESSION;
     private Integer deflateLevel = null;
     private Integer sampleRate = null;
+    private Optional<Integer> maxInFlight = Optional.empty();
     private String userAgent = null;
     private Integer msgTimeout = null;
     private SslContext sslContext = null;
@@ -62,6 +64,15 @@ public Integer getOutputBufferSize() {
         return outputBufferSize;
     }
 
+    public NSQConfig setMaxInFlight(final int maxInFlight) {
+        this.maxInFlight = Optional.of(maxInFlight);
+        return this;
+    }
+
+    public Optional<Integer> getMaxInFlight() {
+        return maxInFlight;
+    }
+
     public void setOutputBufferSize(final Integer outputBufferSize) {
         this.outputBufferSize = outputBufferSize;
     }
@@ -118,7 +129,6 @@ public void setMsgTimeout(final Integer msgTimeout) {
         this.msgTimeout = msgTimeout;
     }
 
-
     public SslContext getSslContext() {
         return sslContext;
     }

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

@@ -39,7 +39,7 @@ public class NSQConsumer implements Closeable {
     private final AtomicLong totalMessages = new AtomicLong(0l);
 
     private boolean started = false;
-    private int messagesPerBatch = 200;
+    private int messagesPerBatch;
     private long lookupPeriod = 60 * 1000; // how often to recheck for new nodes (and clean up non responsive nodes)
     private ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
     private Executor executor = Executors.newCachedThreadPool();
@@ -62,7 +62,7 @@ public NSQConsumer(final NSQLookup lookup, final String topic, final String chan
         this.config = config;
         this.callback = callback;
         this.errorCallback = errCallback;
-
+        this.messagesPerBatch = config.getMaxInFlight().orElse(200);
     }
 
     public NSQConsumer start() {