Add support for using more than 1 sender thread.

3 files changed, 28 insertions, 6 deletions

diff --git a/vespaclient-java/src/main/java/com/yahoo/vespafeeder/Arguments.java b/vespaclient-java/src/main/java/com/yahoo/vespafeeder/Arguments.java
index 86512dfab73..0d23af1fec5 100644
--- a/vespaclient-java/src/main/java/com/yahoo/vespafeeder/Arguments.java
+++ b/vespaclient-java/src/main/java/com/yahoo/vespafeeder/Arguments.java
@@ -39,14 +39,15 @@ public class Arguments {
     }
 
     private FeederConfig.Builder feederConfigBuilder = new FeederConfig.Builder();
-    private List<String> files = new ArrayList<String>();
+    private List<String> files = new ArrayList<>();
     private String dumpDocumentsFile = null;
     private String mode = "standard";
     private boolean validateOnly = false;
     private boolean verbose = false;
-    SessionFactory sessionFactory = null;
+    SessionFactory sessionFactory;
     MessagePropertyProcessor propertyProcessor = null;
     private String priority = null;
+    private int numThreads = 1;
 
     public MessagePropertyProcessor getPropertyProcessor() {
         return propertyProcessor;
@@ -83,6 +84,7 @@ public class Arguments {
                 "                                feeding them.\n" +
                 "  --dumpDocuments <filename>    Specify a file where documents in the put are serialized.\n" +
                 "  --priority arg                Specify priority of sent messages (see documentation for priority values)\n" +
+                "  --numthreads arg              Specify how many threads to use for sending. Default is 1.\n" +
                 "  --create-if-non-existent      Enable setting of create-if-non-existent to true on all document updates in the given xml feed.\n" +
                 "  -v [ --verbose ]              Enable verbose output of progress.\n");
     }
@@ -152,6 +154,8 @@ public class Arguments {
                 verbose = true;
             } else if ("--priority".equals(arg)) {
                 priority = getParam(args, arg);
+            } else if ("--numthreads".equals(arg)) {
+                numThreads = Integer.parseInt(getParam(args, arg));
             } else {
                 files.add(arg);
             }
@@ -183,6 +187,10 @@ public class Arguments {
         return priority;
     }
 
+    public int getNumThreads() {
+        return numThreads;
+    }
+
     public SessionFactory getSessionFactory() {
         return sessionFactory;
     }
diff --git a/vespaclient-java/src/main/java/com/yahoo/vespafeeder/VespaFeeder.java b/vespaclient-java/src/main/java/com/yahoo/vespafeeder/VespaFeeder.java
index 0a926f6aae2..f80567709c4 100755
--- a/vespaclient-java/src/main/java/com/yahoo/vespafeeder/VespaFeeder.java
+++ b/vespaclient-java/src/main/java/com/yahoo/vespafeeder/VespaFeeder.java
@@ -13,8 +13,14 @@ import com.yahoo.log.LogSetup;
 import com.yahoo.concurrent.SystemTimer;
 import com.yahoo.vespaclient.ClusterList;
 
-import java.io.*;
-import java.util.*;
+import java.io.BufferedInputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.PrintStream;
+import java.util.List;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
 
@@ -82,7 +88,7 @@ public class VespaFeeder {
         if (args.getFiles().isEmpty()) {
             InputStreamRequest req = new InputStreamRequest(input);
             setProperties(req, input);
-            FeedResponse response = (FeedResponse)handler.handle(req.toRequest(), createProgressCallback(output));
+            FeedResponse response = (FeedResponse)handler.handle(req.toRequest(), createProgressCallback(output), args.getNumThreads());
             if ( ! response.isSuccess()) {
                 throw renderErrors(response.getErrorList());
             }
@@ -100,7 +106,7 @@ public class VespaFeeder {
                 final BufferedInputStream inputSnooper = new BufferedInputStream(new FileInputStream(fileName));
                 setProperties(req, inputSnooper);
                 inputSnooper.close();
-                FeedResponse response = (FeedResponse)handler.handle(req.toRequest(), createProgressCallback(output));
+                FeedResponse response = (FeedResponse)handler.handle(req.toRequest(), createProgressCallback(output), args.getNumThreads());
                 if (!response.isSuccess()) {
                     throw renderErrors(response.getErrorList());
                 }
diff --git a/vespaclient-java/src/test/java/com/yahoo/vespafeeder/VespaFeederTestCase.java b/vespaclient-java/src/test/java/com/yahoo/vespafeeder/VespaFeederTestCase.java
index d1b7397de34..4de286398e9 100644
--- a/vespaclient-java/src/test/java/com/yahoo/vespafeeder/VespaFeederTestCase.java
+++ b/vespaclient-java/src/test/java/com/yahoo/vespafeeder/VespaFeederTestCase.java
@@ -68,6 +68,14 @@ public class VespaFeederTestCase {
     }
 
     @Test
+    public void requireThatnumThreadsBeParsed() throws Exception {
+        String argsS="--numthreads 5";
+        Arguments arguments = new Arguments(argsS.split(" "), DummySessionFactory.createWithAutoReply());
+        assertEquals(5, arguments.getNumThreads());
+        assertEquals(1, new Arguments("".split(" "), DummySessionFactory.createWithAutoReply()).getNumThreads());
+    }
+
+    @Test
     public void testHelp() throws Exception {
         String argsS="-h";