Fixed the regression in bulk (#110)

Author: danishjamal104

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

@@ -7,7 +7,7 @@
 import org.apache.camel.builder.RouteBuilder;
 import org.springframework.stereotype.Component;
 
-@Component
+//@Component
 public class ExternalApiCallRoute extends RouteBuilder {
 
     @Override

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

@@ -32,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);
 
-            InputStream csvFileInputStream = fileTransferService.streamFile(filename, bucketName);
+            byte[] csvFile = fileTransferService.downloadFile(filename, bucketName);
             File file = new File(filename);
             try (FileOutputStream fos = new FileOutputStream(file)) {
-                fos.write(csvFileInputStream.read());
+                fos.write(csvFile);
             }
             exchange.setProperty(LOCAL_FILE_PATH, file.getAbsolutePath());
             logger.info("File downloaded");

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

@@ -3,6 +3,7 @@
 import static org.mifos.processor.bulk.camel.config.CamelProperties.BATCH_ID_HEADER;
 import static org.mifos.processor.bulk.camel.config.CamelProperties.EXTERNAL_ENDPOINT;
 import static org.mifos.processor.bulk.camel.config.CamelProperties.EXTERNAL_ENDPOINT_FAILED;
+import static org.mifos.processor.bulk.camel.config.CamelProperties.HEADER_CLIENT_CORRELATION_ID;
 import static org.mifos.processor.bulk.camel.config.CamelProperties.IS_PAYMENT_MODE_VALID;
 import static org.mifos.processor.bulk.camel.config.CamelProperties.LOCAL_FILE_PATH;
 import static org.mifos.processor.bulk.camel.config.CamelProperties.OVERRIDE_HEADER;
@@ -132,7 +133,6 @@ public void configure() throws Exception {
                   logger.info("REQUEST_ID: {}", transaction.getRequestId());
                     exchange.setProperty(TRANSACTION_LIST_ELEMENT, transaction);
                 }).setHeader("Platform-TenantId", exchangeProperty(TENANT_NAME))
-                .setHeader("X-CorrelationID", exchangeProperty(REQUEST_ID))
                 .to("direct:dynamic-payload-setter")
                 .to("direct:external-api-call").to("direct:external-api-response-handler").end() // end loop block
                 .endChoice();
@@ -152,9 +152,10 @@ public void configure() throws Exception {
         from("direct:external-api-response-handler").id("direct:external-api-response-handler")
                 .log("Starting route direct:external-api-response-handler").choice().when(header("CamelHttpResponseCode").isEqualTo(200))
                 .process(exchange -> {
-                    logger.info("reached here");
+                    logger.info("INIT_SUB_BATCH_FAILED is false");
                     exchange.setProperty(INIT_SUB_BATCH_FAILED, false);
                 }).otherwise().process(exchange -> {
+                    logger.info("INIT_SUB_BATCH_FAILED is false");
                     exchange.setProperty(INIT_SUB_BATCH_FAILED, true);
                 }).endChoice();
 
@@ -191,8 +192,13 @@ public void configure() throws Exception {
                 .log(LoggingLevel.DEBUG, "Making API call to endpoint ${exchangeProperty.extEndpoint} and body: ${body}")
                 .setHeader(Exchange.CONTENT_TYPE, constant("application/json"))
                 .setHeader(BATCH_ID_HEADER, simple("${exchangeProperty." + BATCH_ID + "}"))
+                .setHeader(HEADER_CLIENT_CORRELATION_ID, simple("${exchangeProperty." + REQUEST_ID + "}"))
+                .process(exchange -> {
+                    log.debug("Variables: {}", exchange.getProperties());
+                    log.debug("Emergency: {}", exchange.getIn().getHeaders());
+                })
                 .toD(ChannelURL + "${exchangeProperty.extEndpoint}" + "?bridgeEndpoint=true&throwExceptionOnFailure=false")
-                .log(LoggingLevel.DEBUG, "Response body: ${body}").otherwise().endChoice();
+                .log(LoggingLevel.INFO, "Response body: ${body}").otherwise().endChoice();
 
         from("direct:validate-payment-mode").id("direct:validate-payment-mode").log("Starting route direct:validate-payment-mode")
                 .process(exchange -> {

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

@@ -2,7 +2,6 @@
 
 import static org.mifos.processor.bulk.camel.config.CamelProperties.*;
 import static org.mifos.processor.bulk.zeebe.ZeebeVariables.*;
-
 import java.io.BufferedReader;
 import java.io.File;
 import java.io.FileInputStream;
@@ -203,7 +202,7 @@ private void setup() {
                 .setProperty(OVERRIDE_HEADER, constant(true)) // default header in CSV file will be used
                 .to("direct:update-file-v2")
                 .otherwise()
-                .log(LoggingLevel.DEBUG, "No update");
+                .log(LoggingLevel.INFO, "No update");
 
         from("direct:start-batch-process-csv")
                 .id("direct:start-batch-process-csv")
@@ -261,7 +260,10 @@ private void setup() {
                     variables.put(REGISTERING_INSTITUTE_ID, exchange.getProperty(REGISTERING_INSTITUTE_ID));
                     variables.put(IS_FILE_VALID, true);
                     setConfigProperties(variables);
+
 					logger.debug("Zeebe variables published: {}", variables);
+                    log.debug("Variables published to zeebe: {}", variables);
+
                     JSONObject response = new JSONObject();
 
                     try {

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

@@ -8,6 +8,7 @@
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import com.amazonaws.util.IOUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -24,6 +25,19 @@ public class AwsFileTransferImpl implements FileTransferService {
     @Autowired
     private AmazonS3 s3Client;
 
+    @Override
+    public byte[] downloadFile(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;
+    }
+
     @Override
     public String uploadFile(MultipartFile file, String bucketName) {
 

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

@@ -5,7 +5,9 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.file.Files;
+import java.nio.file.Paths;
 
+import com.azure.storage.blob.models.BlobProperties;
 import com.azure.storage.blob.specialized.BlobInputStream;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -25,6 +27,20 @@ public class AzureFileTransferImpl implements FileTransferService {
     @Autowired
     BlobClientBuilder client;
 
+    @Override
+    public byte[] downloadFile(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;
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+        return null;
+    }
+
     @Override
     public String uploadFile(MultipartFile file, String bucketName) {
 

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

@@ -8,6 +8,8 @@
 @Service
 public interface FileTransferService {
 
+    byte[] downloadFile(String fileName, String bucketName);
+
     String uploadFile(MultipartFile file, String bucketName);
 
     String uploadFile(File file, String bucketName);

src/main/resources/application.yaml

@@ -163,3 +163,19 @@ management:
         enabled: true
       readiness:
         enabled: true
+
+payment-mode:
+  mappings:
+    - id: "GSMA"
+      type: "PAYMENT"
+      endpoint: "/channel/gsma/transfer"
+    - id: "MOJALOOP"
+      type: "PAYMENT"
+      endpoint: "/channel/transfer"
+    - id: "SLCB"
+      type: "BULK"
+      endpoint: "bulk_connector_{MODE}-{dfspid}"
+    - id: "CLOSEDLOOP"
+      type: "BULK"
+      endpoint: "bulk_connector_{MODE}-{dfspid}"
+      debulkingDfspid: "lion"