replace byte[] with inputStream (#101)

Author: Anover000

src/main/java/org/mifos/processor/bulk/HealthCheck.java

@@ -6,6 +6,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.dataformat.csv.CsvMapper;
 import com.fasterxml.jackson.dataformat.csv.CsvSchema;
+import java.io.InputStream;
 import java.util.UUID;
 import org.apache.camel.Exchange;
 import org.apache.camel.LoggingLevel;
@@ -61,11 +62,11 @@ public void configure() {
                     exchange.setProperty(BATCH_ID, batchId);
 
                     // TODO: How to get sender information? Hard coded in Channel connector?
-                    byte[] csvFile = fileTransferService.downloadFile(fileName, bucketName);
+                    InputStream csvFileInputStream = fileTransferService.streamFile(fileName, bucketName);
 
                     CsvSchema schema = CsvSchema.emptySchema().withHeader();
                     MappingIterator<TransactionOlder> readValues = csvMapper.readerWithSchemaFor(TransactionOlder.class).with(schema)
-                            .readValues(csvFile);
+                            .readValues(csvFileInputStream);
 
                     while (readValues.hasNext()) {
                         TransactionOlder current = readValues.next();

src/main/java/org/mifos/processor/bulk/api/ApiOriginFilter.java

@@ -22,7 +22,7 @@ public void doFilter(ServletRequest request, ServletResponse response, FilterCha
         HttpServletRequest req = (HttpServletRequest) request;
         String tenant = req.getHeader("Platform-TenantId");
         logger.debug("Tenant Name is : {}", tenant);
-        logger.info("Client IP Address: {}",req.getRemoteHost());
+        logger.info("Client IP Address: {}", req.getRemoteHost());
     }
 
 }

src/main/java/org/mifos/processor/bulk/camel/routes/FileRoute.java

@@ -5,6 +5,7 @@
 
 import java.io.File;
 import java.io.FileOutputStream;
+import java.io.InputStream;
 import org.mifos.processor.bulk.file.FileTransferService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
@@ -31,10 +32,10 @@ public void configure() throws Exception {
         from("direct:download-file").id("direct:download-file").log("Started download-file route").process(exchange -> {
             String filename = exchange.getProperty(SERVER_FILE_NAME, String.class);
 
-            byte[] csvFile = fileTransferService.downloadFile(filename, bucketName);
+            InputStream csvFileInputStream = fileTransferService.streamFile(filename, bucketName);
             File file = new File(filename);
             try (FileOutputStream fos = new FileOutputStream(file)) {
-                fos.write(csvFile);
+                fos.write(csvFileInputStream.read());
             }
             exchange.setProperty(LOCAL_FILE_PATH, file.getAbsolutePath());
             logger.info("File downloaded");

src/main/java/org/mifos/processor/bulk/camel/routes/HealthRoute.java

@@ -10,8 +10,11 @@ public class HealthRoute extends BaseRouteBuilder {
     public void configure() throws Exception {
 
         // todo remove once camel APIs are migrated to spring
-        from("rest:GET:/actuator/health/liveness")
-                .id("rest:GET:/actuator/health/liveness")
-                .setBody(exchange -> new JSONObject(){{ put("status", "UP"); }}.toString());
+        from("rest:GET:/actuator/health/liveness").id("rest:GET:/actuator/health/liveness").setBody(exchange -> new JSONObject() {
+
+            {
+                put("status", "UP");
+            }
+        }.toString());
     }
 }

src/main/java/org/mifos/processor/bulk/file/AwsFileTransferImpl.java

@@ -4,10 +4,10 @@
 import com.amazonaws.services.s3.model.PutObjectRequest;
 import com.amazonaws.services.s3.model.S3Object;
 import com.amazonaws.services.s3.model.S3ObjectInputStream;
-import com.amazonaws.util.IOUtils;
 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
+import java.io.InputStream;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -41,16 +41,10 @@ public String uploadFile(File file, String bucketName) {
     }
 
     @Override
-    public byte[] downloadFile(String fileName, String bucketName) {
+    public InputStream streamFile(String fileName, String bucketName) {
         S3Object s3Object = s3Client.getObject(bucketName, fileName);
         S3ObjectInputStream inputStream = s3Object.getObjectContent();
-        try {
-            byte[] content = IOUtils.toByteArray(inputStream);
-            return content;
-        } catch (IOException e) {
-            e.printStackTrace();
-        }
-        return null;
+        return inputStream;
     }
 
     @Override

src/main/java/org/mifos/processor/bulk/file/AzureFileTransferImpl.java

@@ -1,11 +1,12 @@
 package org.mifos.processor.bulk.file;
 
 import com.azure.storage.blob.BlobClientBuilder;
-import com.azure.storage.blob.models.BlobProperties;
 import java.io.File;
 import java.io.IOException;
+import java.io.InputStream;
 import java.nio.file.Files;
-import java.nio.file.Paths;
+
+import com.azure.storage.blob.specialized.BlobInputStream;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -51,13 +52,13 @@ public String uploadFile(File file, String bucketName) {
     }
 
     @Override
-    public byte[] downloadFile(String fileName, String bucketName) {
+    public InputStream streamFile(String fileName, String bucketName) {
         try {
             File temp = new File("/temp/" + fileName);
-            BlobProperties properties = client.containerName(bucketName).blobName(fileName).buildClient().downloadToFile(temp.getPath());
-            byte[] content = Files.readAllBytes(Paths.get(temp.getPath()));
-            temp.delete();
-            return content;
+            BlobInputStream csvInputStream = client.containerName(bucketName).blobName(fileName).buildClient().openInputStream();
+//            byte[] content = Files.Paths.get(temp.getPath()));
+//            temp.delete();
+            return csvInputStream;
         } catch (Exception e) {
             e.printStackTrace();
         }
@@ -68,4 +69,5 @@ public byte[] downloadFile(String fileName, String bucketName) {
     public void deleteFile(String fileName, String bucketName) {
         client.containerName(bucketName).blobName(fileName).buildClient().delete();
     }
+
 }

src/main/java/org/mifos/processor/bulk/file/FileTransferService.java

@@ -1,6 +1,7 @@
 package org.mifos.processor.bulk.file;
 
 import java.io.File;
+import java.io.InputStream;
 import org.springframework.stereotype.Service;
 import org.springframework.web.multipart.MultipartFile;
 
@@ -11,7 +12,7 @@ public interface FileTransferService {
 
     String uploadFile(File file, String bucketName);
 
-    byte[] downloadFile(String fileName, String bucketName);
+    InputStream streamFile(String fileName, String bucketName);
 
     void deleteFile(String fileName, String bucketName);
 

src/main/java/org/mifos/processor/bulk/zeebe/ZeebeWorkers.java

@@ -9,6 +9,7 @@
 import com.fasterxml.jackson.dataformat.csv.CsvMapper;
 import com.fasterxml.jackson.dataformat.csv.CsvSchema;
 import io.camunda.zeebe.client.ZeebeClient;
+import java.io.InputStream;
 import java.util.Map;
 import org.apache.camel.CamelContext;
 import org.apache.camel.Exchange;
@@ -70,10 +71,10 @@ private void workerBulkProcessor() {
             String fileName = (String) variables.get("fileName");
 
             // TODO: How to get sender information? Hard coded in Channel connector?
-            byte[] csvFile = fileTransferService.downloadFile(fileName, bucketName);
+            InputStream csvFileInputStream = fileTransferService.streamFile(fileName, bucketName);
 
             CsvSchema schema = CsvSchema.emptySchema().withHeader();
-            MappingIterator<Transaction> readValues = csvMapper.readerWithSchemaFor(Transaction.class).with(schema).readValues(csvFile);
+            MappingIterator<Transaction> readValues = csvMapper.readerWithSchemaFor(Transaction.class).with(schema).readValues(csvFileInputStream);
 
             /*
              * while (readValues.hasNext()) { Transaction current = readValues.next(); current.setBatchId(batchId); if