Merge pull request #20 from hansdude/master

Added api to helpers to control max parallelism.

Author: rpmoore

src/main/java/com/spectralogic/ds3client/helpers/Ds3ClientHelpers.java

@@ -91,6 +91,11 @@ public interface WriteJob extends Job {
          * Note that it's possible for the {@code modifier} to be called simultaneously from multiple threads.
          */
         public WriteJob withRequestModifier(PutRequestModifier modifier);
+
+        /**
+         * Sets the maximum number of requests to execute at a time when fulfilling the job.
+         */
+        public WriteJob withMaxParallelRequests(int maxParallelRequests);
     }
 
     public interface ReadJob extends Job {
@@ -109,6 +114,11 @@ public interface ReadJob extends Job {
          * Note that it's possible for the {@code modifier} to be called simultaneously from multiple threads.
          */
         public ReadJob withRequestModifier(GetRequestModifier modifier);
+        
+        /**
+         * Sets the maximum number of requests to execute at a time when fulfilling the job.
+         */
+        public ReadJob withMaxParallelRequests(int maxParallelRequests);
     }
 
     /**

src/main/java/com/spectralogic/ds3client/helpers/JobImpl.java

@@ -35,7 +35,7 @@
 import com.spectralogic.ds3client.serializer.XmlProcessingException;
 
 abstract class JobImpl implements Job {
-    private static final int THREAD_COUNT = 10;
+    private int maxParallelRequests = 20;
 
     private final Ds3ClientFactory clientFactory;
     private final UUID jobId;
@@ -63,6 +63,10 @@ public String getBucketName() {
         return this.bucketName;
     }
 
+    protected void setMaxParallelRequests(final int maxParallelRequests) {
+        this.maxParallelRequests = maxParallelRequests;
+    }
+    
     interface Transferrer {
         public void Transfer(Ds3Client client, UUID jobId, String bucketName, Ds3Object ds3Object)
                 throws SignatureException, IOException;
@@ -77,7 +81,7 @@ protected void transferAll(final Transferrer transferrer)
 
     private void transferObjects(final Transferrer transferrer, final Objects objects)
             throws SignatureException, IOException, XmlProcessingException {
-        final ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(THREAD_COUNT));
+        final ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(this.maxParallelRequests));
         try {
             final Ds3Client client = this.clientFactory.GetClientForServerId(objects.getServerId());
             final List<ListenableFuture<?>> tasks = new ArrayList<>();

src/main/java/com/spectralogic/ds3client/helpers/ReadJobImpl.java

@@ -70,4 +70,10 @@ public ReadJob withRequestModifier(final GetRequestModifier modifier) {
         this.modifier = modifier;
         return this;
     }
+
+    @Override
+    public ReadJob withMaxParallelRequests(final int maxParallelRequests) {
+        this.setMaxParallelRequests(maxParallelRequests);
+        return this;
+    }
 }

src/main/java/com/spectralogic/ds3client/helpers/WriteJobImpl.java

@@ -69,4 +69,10 @@ public WriteJob withRequestModifier(final PutRequestModifier modifier) {
         this.modifier = modifier;
         return this;
     }
+
+    @Override
+    public WriteJob withMaxParallelRequests(final int maxParallelRequests) {
+        this.setMaxParallelRequests(maxParallelRequests);
+        return this;
+    }
 }