draft for callback

ankita10r · Dec 15, 2023 · 9545245 · 9545245
1 parent b4248a2
commit 9545245
Show file tree

Hide file tree

Showing 16 changed files with 374 additions and 355 deletions.
diff --git a/src/main/java/org/mifos/processor/bulk/api/definition/BatchTransactions.java b/src/main/java/org/mifos/processor/bulk/api/definition/BatchTransactions.java
@@ -1,5 +1,12 @@
 package org.mifos.processor.bulk.api.definition;
 
+import org.springframework.web.bind.annotation.*;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import java.io.IOException;
+
+import static org.mifos.processor.bulk.camel.config.CamelProperties.*;
 import static org.mifos.processor.bulk.camel.config.CamelProperties.HEADER_CLIENT_CORRELATION_ID;
 import static org.mifos.processor.bulk.camel.config.CamelProperties.HEADER_PLATFORM_TENANT_ID;
 import static org.mifos.processor.bulk.camel.config.CamelProperties.HEADER_PROGRAM_ID;
@@ -19,8 +26,11 @@ public interface BatchTransactions {
     @PostMapping(value = "/batchtransactions", produces = "application/json")
     String batchTransactions(HttpServletRequest httpServletRequest, HttpServletResponse httpServletResponse,
             @RequestHeader(value = HEADER_CLIENT_CORRELATION_ID) String requestId,
-            @RequestHeader(value = FILE_NAME, required = false) String fileName, @RequestHeader(value = PURPOSE) String purpose,
-            @RequestHeader(value = HEADER_TYPE) String type, @RequestHeader(value = HEADER_PLATFORM_TENANT_ID) String tenant,
+            @RequestHeader(value = FILE_NAME, required = false) String fileName,
+            @RequestHeader(value = PURPOSE) String purpose,
+            @RequestParam(value = HEADER_TYPE) String type,
+            @RequestHeader(value = HEADER_PLATFORM_TENANT_ID) String tenant,
+            @RequestHeader(value = CALLBACK, required = false) String callbackUrl,
             @RequestHeader(value = HEADER_REGISTERING_INSTITUTE_ID, required = false) String registeringInstitutionId,
             @RequestHeader(value = HEADER_PROGRAM_ID, required = false) String programId) throws IOException;
 

diff --git a/src/main/java/org/mifos/processor/bulk/api/implementation/BatchTransactionsController.java b/src/main/java/org/mifos/processor/bulk/api/implementation/BatchTransactionsController.java
@@ -2,6 +2,7 @@
 
 import static org.mifos.processor.bulk.camel.config.CamelProperties.HEADER_PROGRAM_ID;
 import static org.mifos.processor.bulk.camel.config.CamelProperties.HEADER_REGISTERING_INSTITUTE_ID;
+import static org.mifos.processor.bulk.zeebe.ZeebeVariables.CALLBACK;
 import static org.mifos.processor.bulk.zeebe.ZeebeVariables.FILE_NAME;
 import static org.mifos.processor.bulk.zeebe.ZeebeVariables.HEADER_CLIENT_CORRELATION_ID;
 import static org.mifos.processor.bulk.zeebe.ZeebeVariables.HEADER_PLATFORM_TENANT_ID;
@@ -39,6 +40,21 @@
 import org.springframework.web.bind.annotation.RestController;
 import org.springframework.web.multipart.MultipartException;
 
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import java.io.IOException;
+import static org.mifos.processor.bulk.camel.config.CamelProperties.*;
+
+import java.nio.charset.Charset;
+import java.util.List;
+import java.util.Optional;
+import java.util.UUID;
+import static org.mifos.processor.bulk.camel.config.CamelProperties.HEADER_PROGRAM_ID;
+import static org.mifos.processor.bulk.camel.config.CamelProperties.HEADER_REGISTERING_INSTITUTE_ID;
+import static org.mifos.processor.bulk.zeebe.ZeebeVariables.*;
+import static org.mifos.processor.bulk.zeebe.ZeebeVariables.HEADER_CLIENT_CORRELATION_ID;
+import static org.mifos.processor.bulk.zeebe.ZeebeVariables.HEADER_PLATFORM_TENANT_ID;
+
 @Slf4j
 @RestController
 public class BatchTransactionsController implements BatchTransactions {
@@ -63,12 +79,12 @@ public class BatchTransactionsController implements BatchTransactions {
     @SneakyThrows
     @Override
     public String batchTransactions(HttpServletRequest httpServletRequest, HttpServletResponse httpServletResponse, String requestId,
-            String fileName, String purpose, String type, String tenant, String registeringInstitutionId, String programId) {
+                                    String fileName, String purpose, String type, String tenant, String callbackUrl, String registeringInstitutionId, String programId) throws IOException {
 
         log.info("Inside api logic");
         Headers.HeaderBuilder headerBuilder = new Headers.HeaderBuilder().addHeader(HEADER_CLIENT_CORRELATION_ID, requestId)
                 .addHeader(PURPOSE, purpose).addHeader(HEADER_TYPE, type).addHeader(HEADER_PLATFORM_TENANT_ID, tenant)
-                .addHeader(HEADER_REGISTERING_INSTITUTE_ID, registeringInstitutionId).addHeader(HEADER_PROGRAM_ID, programId);
+                .addHeader(CALLBACK, callbackUrl).addHeader(HEADER_REGISTERING_INSTITUTE_ID, registeringInstitutionId).addHeader(HEADER_PROGRAM_ID, programId);
 
         Optional<String> validationResponse = isValidRequest(httpServletRequest, fileName, type);
         if (validationResponse.isPresent()) {

diff --git a/src/main/java/org/mifos/processor/bulk/camel/config/CamelProperties.java b/src/main/java/org/mifos/processor/bulk/camel/config/CamelProperties.java
@@ -83,4 +83,6 @@ private CamelProperties() {}
     public static final String EVENT_TYPE = "eventType";
     public static final String DUPLICATE_TRANSACTION_LIST = "duplicateTransactionList";
     public static final String ORIGINAL_TRANSACTION_LIST = "originalTransactionList";
+    public static final String CALLBACK = "X-CallbackURL";
+
 }
diff --git a/src/main/java/org/mifos/processor/bulk/camel/routes/BatchAggregateRoute.java b/src/main/java/org/mifos/processor/bulk/camel/routes/BatchAggregateRoute.java
@@ -52,7 +52,9 @@ public void configure() throws Exception {
                 .log("Starting route direct:batch-aggregate-response-handler")
                 // .setBody(exchange -> exchange.getIn().getBody(String.class))
                 .choice().when(header("CamelHttpResponseCode").isEqualTo("200")).log(LoggingLevel.INFO, "Batch summary request successful")
-                .unmarshal().json(JsonLibrary.Jackson, BatchDTO.class).process(exchange -> {
+                .log("Response body: ${body}")
+                .unmarshal().json(JsonLibrary.Jackson, BatchDTO.class)
+                .process(exchange -> {
                     BatchDTO batchAggregateResponse = exchange.getIn().getBody(BatchDTO.class);
                     int percentage = (int) (((double) batchAggregateResponse.getSuccessful() / batchAggregateResponse.getTotal()) * 100);
 

diff --git a/src/main/java/org/mifos/processor/bulk/camel/routes/InitSubBatchRoute.java b/src/main/java/org/mifos/processor/bulk/camel/routes/InitSubBatchRoute.java
@@ -198,12 +198,12 @@ public void configure() throws Exception {
                 .setHeader(HEADER_CLIENT_CORRELATION_ID, simple("${exchangeProperty." + REQUEST_ID + "}"))
                 .setHeader(HEADER_REGISTERING_INSTITUTE_ID, simple("${exchangeProperty." + HEADER_REGISTERING_INSTITUTE_ID + "}"))
                 .process(exchange -> {
-                    log.debug("Variables: {}", exchange.getProperties());
-                    log.debug("Emergency: {}", exchange.getIn().getHeaders());
+                    log.info("Variables: {}", exchange.getProperties());
+                    log.info("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 -> {

diff --git a/src/main/java/org/mifos/processor/bulk/camel/routes/ProcessorStartRoute.java b/src/main/java/org/mifos/processor/bulk/camel/routes/ProcessorStartRoute.java
@@ -13,36 +13,7 @@
 import static org.mifos.processor.bulk.camel.config.CamelProperties.RESULT_TRANSACTION_LIST;
 import static org.mifos.processor.bulk.camel.config.CamelProperties.TENANT_NAME;
 import static org.mifos.processor.bulk.camel.config.CamelProperties.TRANSACTION_LIST;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.APPROVAL_ENABLED;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.AUTHORIZATION_ENABLED;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.BATCH_AGGREGATE_ENABLED;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.BATCH_ID;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.BULK_NOTIF_FAILURE;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.BULK_NOTIF_SUCCESS;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.CALLBACK_URL;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.CLIENT_CORRELATION_ID;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.COMPLETION_THRESHOLD;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.COMPLETION_THRESHOLD_CHECK_ENABLED;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.DE_DUPLICATION_ENABLE;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.FILE_NAME;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.FORMATTING_ENABLED;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.IS_FILE_VALID;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.MAX_CALLBACK_RETRY;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.MAX_STATUS_RETRY;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.MERGE_ENABLED;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.NOTE;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.ORDERING_ENABLED;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.PARTY_LOOKUP_ENABLED;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.PAYER_IDENTIFIER_TYPE;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.PAYER_IDENTIFIER_VALUE;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.PHASES;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.PHASE_COUNT;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.PROGRAM_NAME;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.PURPOSE;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.REQUEST_ID;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.SPLITTING_ENABLED;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.TENANT_ID;
-import static org.mifos.processor.bulk.zeebe.ZeebeVariables.THRESHOLD_DELAY;
+import static org.mifos.processor.bulk.zeebe.ZeebeVariables.*;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import java.io.BufferedReader;
@@ -149,6 +120,9 @@ private void setup() {
             String requestId = exchange.getIn().getHeader("requestId", String.class);
             String purpose = exchange.getIn().getHeader("purpose", String.class);
             String batchId = UUID.randomUUID().toString();
+            // extracting and setting callback Url
+            String callbackUrl = exchange.getIn().getHeader("X-CallbackURL", String.class);
+            exchange.setProperty(CALLBACK, callbackUrl);
             exchange.setProperty(BATCH_ID, batchId);
             exchange.setProperty(FILE_NAME, fileName);
             exchange.setProperty(REQUEST_ID, requestId);
@@ -232,6 +206,7 @@ private void setup() {
                     String requestId = exchange.getProperty(REQUEST_ID, String.class);
                     String purpose = exchange.getProperty(PURPOSE, String.class);
                     String batchId = exchange.getProperty(BATCH_ID, String.class);
+                    String callbackUrl = exchange.getProperty(CALLBACK,String.class);
                     String note = null;
 
                     if (purpose == null || purpose.isEmpty()) {
@@ -248,7 +223,7 @@ private void setup() {
                     logger.debug("File absolute path: {}", file.getAbsolutePath());
 
                     boolean verifyData = verifyData(file);
-                    logger.debug("Data verification result {}", verifyData);
+                    logger.info("Data verification result {}", verifyData);
                     if (!verifyData) {
                         note = "Invalid data in file data processing stopped";
                     }
@@ -257,9 +232,7 @@ private void setup() {
 
                     logger.debug("File uploaded {}", nm);
 
-                    // extracting and setting callback Url
-                    String callbackUrl = exchange.getIn().getHeader("X-Callback-URL", String.class);
-                    exchange.setProperty(CALLBACK_URL, callbackUrl);
+
 
                     List<Integer> phases = phaseUtils.getValues();
                     logger.debug(phases.toString());
@@ -269,7 +242,7 @@ private void setup() {
                     variables.put(REQUEST_ID, requestId);
                     variables.put(PURPOSE, purpose);
                     variables.put(TENANT_ID, exchange.getProperty(TENANT_NAME));
-                    variables.put(CALLBACK_URL, callbackUrl);
+                    variables.put(CALLBACK, callbackUrl);
                     variables.put(PHASES, phases);
                     variables.put(PHASE_COUNT, phases.size());
                     variables.put(NOTE, note);
@@ -322,10 +295,13 @@ private void setup() {
                     String filename = exchange.getIn().getHeader("filename", String.class);
                     String requestId = exchange.getIn().getHeader("X-CorrelationID", String.class);
                     String purpose = exchange.getIn().getHeader("Purpose", String.class);
-                    String type = exchange.getIn().getHeader("Type", String.class);
+                    String type = exchange.getIn().getHeader("type", String.class);
                     String clientCorrelationId = exchange.getIn().getHeader(HEADER_CLIENT_CORRELATION_ID, String.class);
                     String registeringInstitutionId = exchange.getIn().getHeader(HEADER_REGISTERING_INSTITUTE_ID, String.class);
                     String programId = exchange.getIn().getHeader(HEADER_PROGRAM_ID, String.class);
+                    // extracting and setting callback Url
+                    String callbackUrl = exchange.getIn().getHeader("X-CallbackURL", String.class);
+                    exchange.setProperty(CALLBACK, callbackUrl);
                     exchange.setProperty(FILE_NAME, filename);
                     exchange.setProperty(REQUEST_ID, requestId);
                     exchange.setProperty(PURPOSE, purpose);
@@ -396,9 +372,9 @@ private boolean verifyData(File file) throws IOException {
         String line;
         br.readLine();
         while ((line = br.readLine()) != null) {
-            String[] row = line.split(",");
+            String[] row = line.split(",", -1);
             if (row.length != columnNames.size()) {
-                logger.debug("Row invalid {} {}", row.length, columnNames.size());
+                logger.info("Row invalid {} {}", row.length, columnNames.size());
                 return false;
             }
             if (!verifyRow(row)) {
@@ -415,14 +391,14 @@ private boolean verifyRow(String[] row) {
                 int j = row[i].indexOf("MSISDN");
                 if (!(j == row.length)) {
                     if (!row[j + 1].matches("^[0-9]*$")) {
-                        logger.debug("MSISDN invalid");
+                        logger.info("MSISDN invalid");
                         return false;
                     }
                 }
             } else if (row[i].contains("amount")) {
                 int j = row[i].indexOf("amount");
                 if (!row[j].matches("^[0-9]*$")) {
-                    logger.debug("Amount invalid");
+                    logger.info("Amount invalid");
                     return false;
                 }
 
@@ -437,12 +413,12 @@ private boolean verifyCsv(File csvData) throws IOException {
         String[] columns = new String[0];
         if (header != null) {
             columns = header.split(",");
-            logger.debug("Columns in the csv file are {}", Arrays.toString(columns));
+            logger.info("Columns in the csv file are {}", Arrays.toString(columns));
         }
         int i = 0;
         while (i < columns.length) {
             if (columnNames.contains(columns[i])) {
-                logger.debug("Column name {} is at index {} ", columns[i], columnNames.indexOf(columns[i]));
+                logger.info("Column name {} is at index {} ", columns[i], columnNames.indexOf(columns[i]));
                 i++;
 
             } else {