add ability to change the underlying executor service in OutputStream's

Signed-off-by: Andrea Cioni <[email protected]>

Author: andreacioni

spring-batch-s3/src/main/java/org/springframework/batch/extensions/s3/stream/S3MultipartOutputStream.java

@@ -44,7 +44,7 @@ public class S3MultipartOutputStream extends OutputStream {
 
 	private final PipedOutputStream pipedOutputStream;
 
-	private final ExecutorService singleThreadExecutor;
+	private  ExecutorService singleThreadExecutor;
 
 	private volatile boolean uploading;
 
@@ -57,7 +57,6 @@ public S3MultipartOutputStream(S3Client s3Client, String bucketName, String key)
 	public S3MultipartOutputStream(S3Uploader s3Uploader) throws IOException {
 		this.pipedInputStream = new PipedInputStream();
 		this.pipedOutputStream = new PipedOutputStream(this.pipedInputStream);
-		this.singleThreadExecutor = Executors.newSingleThreadExecutor();
 		this.uploading = false;
 		this.multipartUpload = s3Uploader;
 	}
@@ -73,6 +72,10 @@ public void write(int b) throws IOException {
 	}
 
 	private void startUpload() {
+		if(this.singleThreadExecutor == null) {
+			this.singleThreadExecutor = Executors.newSingleThreadExecutor();
+		}
+
 		this.singleThreadExecutor.execute(() -> {
 			try {
 				this.multipartUpload.upload(this.pipedInputStream);
@@ -114,4 +117,7 @@ public void close() throws IOException {
 		super.close();
 	}
 
+	public void setSingleThreadExecutor(ExecutorService singleThreadExecutor) {
+		this.singleThreadExecutor = singleThreadExecutor;
+	}
 }

spring-batch-s3/src/main/java/org/springframework/batch/extensions/s3/stream/S3OutputStream.java

@@ -54,7 +54,7 @@ public class S3OutputStream extends OutputStream {
 
 	private final PipedOutputStream pipedOutputStream;
 
-	private final ExecutorService singleThreadExecutor;
+	private ExecutorService singleThreadExecutor;
 
 	private volatile boolean uploading;
 
@@ -66,7 +66,6 @@ public S3OutputStream(S3Client s3, String bucketName, String key) throws IOExcep
 		this.key = key;
 		this.pipedInputStream = new PipedInputStream();
 		this.pipedOutputStream = new PipedOutputStream(this.pipedInputStream);
-		this.singleThreadExecutor = Executors.newSingleThreadExecutor();
 		this.uploading = false;
 	}
 
@@ -80,6 +79,10 @@ public void write(int b) throws IOException {
 	}
 
 	private void runUploadThread() {
+		if(this.singleThreadExecutor == null) {
+			this.singleThreadExecutor = Executors.newSingleThreadExecutor();
+		}
+
 		this.singleThreadExecutor.execute(() -> {
 			try {
 				RequestBody body = RequestBody