Skip to content

Commit febdbee

Browse files
shkr7shkr7
authored
[PHEE-587] Convert BD-011 routes to Java/Spring (#151)
* [PHEE-587] Convert BD-011 route to Java/Spring * Added spotless change * Added csv and incoming data java routes
1 parent 03a9566 commit febdbee

2 files changed

Lines changed: 219 additions & 179 deletions

File tree

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

Lines changed: 8 additions & 179 deletions
Original file line numberDiff line numberDiff line change
@@ -1,62 +1,41 @@
11
package org.mifos.processor.bulk.camel.routes;
22

33
import static org.mifos.processor.bulk.camel.config.CamelProperties.BATCH_REQUEST_TYPE;
4-
import static org.mifos.processor.bulk.camel.config.CamelProperties.HEADER_CLIENT_CORRELATION_ID;
54
import static org.mifos.processor.bulk.camel.config.CamelProperties.HEADER_PLATFORM_TENANT_ID;
6-
import static org.mifos.processor.bulk.camel.config.CamelProperties.HEADER_PROGRAM_ID;
7-
import static org.mifos.processor.bulk.camel.config.CamelProperties.HEADER_REGISTERING_INSTITUTE_ID;
85
import static org.mifos.processor.bulk.camel.config.CamelProperties.IS_UPDATED;
96
import static org.mifos.processor.bulk.camel.config.CamelProperties.LOCAL_FILE_PATH;
107
import static org.mifos.processor.bulk.camel.config.CamelProperties.OVERRIDE_HEADER;
11-
import static org.mifos.processor.bulk.camel.config.CamelProperties.PROGRAM_ID;
12-
import static org.mifos.processor.bulk.camel.config.CamelProperties.REGISTERING_INSTITUTE_ID;
13-
import static org.mifos.processor.bulk.camel.config.CamelProperties.RESULT_TRANSACTION_LIST;
148
import static org.mifos.processor.bulk.camel.config.CamelProperties.TENANT_NAME;
15-
import static org.mifos.processor.bulk.camel.config.CamelProperties.TRANSACTION_LIST;
169
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.APPROVAL_ENABLED;
1710
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.AUTHORIZATION_ENABLED;
1811
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.BATCH_AGGREGATE_ENABLED;
1912
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.BATCH_ID;
2013
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.BULK_NOTIF_FAILURE;
2114
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.BULK_NOTIF_SUCCESS;
2215
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.CALLBACK;
23-
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.CALLBACK_URL;
24-
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.CLIENT_CORRELATION_ID;
2516
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.COMPLETION_THRESHOLD;
2617
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.COMPLETION_THRESHOLD_CHECK_ENABLED;
2718
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.DE_DUPLICATION_ENABLE;
2819
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.FILE_NAME;
2920
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.FORMATTING_ENABLED;
30-
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.IS_FILE_VALID;
3121
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.MAX_CALLBACK_RETRY;
3222
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.MAX_STATUS_RETRY;
3323
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.MERGE_ENABLED;
34-
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.NOTE;
3524
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.ORDERING_ENABLED;
3625
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.PARTY_LOOKUP_ENABLED;
37-
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.PAYER_IDENTIFIER_TYPE;
38-
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.PAYER_IDENTIFIER_VALUE;
39-
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.PHASES;
40-
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.PHASE_COUNT;
41-
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.PROGRAM_NAME;
4226
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.PURPOSE;
4327
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.REQUEST_ID;
4428
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.SPLITTING_ENABLED;
45-
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.TENANT_ID;
4629
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.THRESHOLD_DELAY;
4730

48-
import com.fasterxml.jackson.core.JsonProcessingException;
49-
import io.camunda.zeebe.client.api.command.ClientStatusException;
5031
import java.io.BufferedReader;
5132
import java.io.File;
5233
import java.io.FileInputStream;
5334
import java.io.FileReader;
5435
import java.io.IOException;
5536
import java.io.InputStream;
5637
import java.io.InputStreamReader;
57-
import java.util.ArrayList;
5838
import java.util.Arrays;
59-
import java.util.HashMap;
6039
import java.util.List;
6140
import java.util.Map;
6241
import java.util.UUID;
@@ -65,13 +44,10 @@
6544
import org.apache.tika.Tika;
6645
import org.json.JSONObject;
6746
import org.mifos.processor.bulk.config.BudgetAccountConfig;
68-
import org.mifos.processor.bulk.config.Program;
69-
import org.mifos.processor.bulk.config.RegisteringInstitutionConfig;
7047
import org.mifos.processor.bulk.connectors.service.ProcessorStartRouteService;
7148
import org.mifos.processor.bulk.file.FileTransferService;
7249
import org.mifos.processor.bulk.properties.TenantImplementation;
7350
import org.mifos.processor.bulk.properties.TenantImplementationProperties;
74-
import org.mifos.processor.bulk.schema.Transaction;
7551
import org.mifos.processor.bulk.utility.PhaseUtils;
7652
import org.mifos.processor.bulk.utility.Utils;
7753
import org.mifos.processor.bulk.zeebe.ZeebeProcessStarter;
@@ -82,7 +58,6 @@
8258
import org.springframework.beans.factory.annotation.Qualifier;
8359
import org.springframework.beans.factory.annotation.Value;
8460
import org.springframework.stereotype.Component;
85-
import org.springframework.util.StringUtils;
8661

8762
@Component
8863
public class ProcessorStartRoute extends BaseRouteBuilder {
@@ -181,145 +156,16 @@ private void setup() {
181156
// [LOCAL_FILE_PATH] is already set in [direct:validateFileSyncResponse] route
182157
.setProperty(LOCAL_FILE_PATH, exchangeProperty(FILE_NAME)).to("direct:get-transaction-array")
183158
// make sure new data is set under the exchange variable [RESULT_TRANSACTION_LIST]
184-
.process(exchange -> {
185-
String registeringInstituteId = exchange.getProperty(REGISTERING_INSTITUTE_ID, String.class);
186-
String programId = exchange.getProperty(PROGRAM_ID, String.class);
187-
logger.debug("Inst id: {}, prog id: {}", registeringInstituteId, programId);
188-
if (!(StringUtils.hasText(registeringInstituteId) && StringUtils.hasText(programId))) {
189-
// this will make sure the file is not updated since there is no update in data
190-
logger.debug("InstitutionId or programId is null");
191-
192-
exchange.setProperty(IS_UPDATED, false);
193-
return;
194-
}
195-
List<Transaction> transactionList = exchange.getProperty(TRANSACTION_LIST, List.class);
196-
logger.debug("Size: {}", transactionList.size());
197-
RegisteringInstitutionConfig registeringInstitutionConfig = budgetAccountConfig
198-
.getByRegisteringInstituteId(registeringInstituteId);
199-
if (registeringInstitutionConfig == null) {
200-
logger.debug("Element in nested in config: {}",
201-
budgetAccountConfig.getRegisteringInstitutions().get(0).getPrograms().size());
202-
logger.debug("Registering institute id is null");
203-
204-
exchange.setProperty(IS_UPDATED, false);
205-
return;
206-
}
207-
Program program = registeringInstitutionConfig.getByProgramId(programId);
208-
if (program == null) {
209-
// this will make sure the file is not updated since there is no update in data
210-
logger.debug("Program is null");
211-
exchange.setProperty(IS_UPDATED, false);
212-
return;
213-
}
214-
List<Transaction> resultTransactionList = new ArrayList<>();
215-
216-
transactionList.forEach(transaction -> {
217-
transaction.setPayerIdentifierType(program.getIdentifierType());
218-
transaction.setPayerIdentifier(program.getIdentifierValue());
219-
resultTransactionList.add(transaction);
220-
try {
221-
logger.debug("Txn: {}", objectMapper.writeValueAsString(transaction));
222-
} catch (JsonProcessingException e) {
223-
throw new RuntimeException(e);
224-
}
225-
});
226-
exchange.setProperty(RESULT_TRANSACTION_LIST, resultTransactionList);
227-
exchange.setProperty(IS_UPDATED, true);
228-
exchange.setProperty(PROGRAM_NAME, program.getName());
229-
exchange.setProperty(PAYER_IDENTIFIER_TYPE, program.getIdentifierType());
230-
exchange.setProperty(PAYER_IDENTIFIER_VALUE, program.getIdentifierValue());
231-
}).choice()
159+
.bean(ProcessorStartRouteService.class, "updateIncomingData").choice()
232160
// update only when previous(edit function) makes any changes to data
233161
.when(exchange -> exchange.getProperty(IS_UPDATED, Boolean.class))
234162
// warning: changing this flag can break things
235163
.setProperty(OVERRIDE_HEADER, constant(true)) // default header in CSV file will be used
236164
.to("direct:update-file-v2").otherwise().log(LoggingLevel.INFO, "No update");
237165

238166
from("direct:start-batch-process-csv").id("direct:start-batch-process-csv").log("Starting route direct:start-batch-process-csv")
239-
.to("direct:update-incoming-data").process(exchange -> {
240-
String fileName = exchange.getProperty(FILE_NAME, String.class);
241-
String requestId = exchange.getProperty(REQUEST_ID, String.class);
242-
String purpose = exchange.getProperty(PURPOSE, String.class);
243-
String batchId = exchange.getProperty(BATCH_ID, String.class);
244-
String callbackUrl = exchange.getProperty(CALLBACK, String.class);
245-
String note = null;
246-
247-
if (purpose == null || purpose.isEmpty()) {
248-
purpose = "test payment";
249-
}
250-
251-
logger.debug("\n\n Filename: {}", fileName);
252-
logger.debug("\n\n BatchId: {} ", batchId);
253-
254-
File file = new File(fileName);
255-
file.setWritable(true);
256-
file.setReadable(true);
257-
258-
logger.debug("File absolute path: {}", file.getAbsolutePath());
259-
260-
boolean verifyData = verifyData(file);
261-
logger.debug("Data verification result {}", verifyData);
262-
if (!verifyData) {
263-
note = "Invalid data in file data processing stopped";
264-
}
265-
266-
String nm = fileTransferService.uploadFile(file, bucketName);
267-
268-
logger.debug("File uploaded {}", nm);
269-
270-
// extracting and setting callback Url
271-
exchange.setProperty(CALLBACK_URL, callbackUrl);
272-
273-
List<Integer> phases = phaseUtils.getValues();
274-
logger.debug(phases.toString());
275-
Map<String, Object> variables = new HashMap<>();
276-
variables.put(BATCH_ID, batchId);
277-
variables.put(FILE_NAME, fileName);
278-
variables.put(REQUEST_ID, requestId);
279-
variables.put(PURPOSE, purpose);
280-
variables.put(TENANT_ID, exchange.getProperty(TENANT_NAME));
281-
variables.put(CALLBACK, callbackUrl);
282-
variables.put(PHASES, phases);
283-
variables.put(PHASE_COUNT, phases.size());
284-
variables.put(NOTE, note);
285-
variables.put(CLIENT_CORRELATION_ID, exchange.getProperty(CLIENT_CORRELATION_ID));
286-
variables.put(PROGRAM_NAME, exchange.getProperty(PROGRAM_NAME));
287-
variables.put(PAYER_IDENTIFIER_TYPE, exchange.getProperty(PAYER_IDENTIFIER_TYPE));
288-
variables.put(PAYER_IDENTIFIER_VALUE, exchange.getProperty(PAYER_IDENTIFIER_VALUE));
289-
variables.put(REGISTERING_INSTITUTE_ID, exchange.getProperty(REGISTERING_INSTITUTE_ID));
290-
variables.put(IS_FILE_VALID, true);
291-
setConfigProperties(variables);
292-
293-
logger.debug("Zeebe variables published: {}", variables);
294-
log.debug("Variables published to zeebe: {}", variables);
295-
296-
JSONObject response = new JSONObject();
297-
String bpmn = getWorkflowForTenant(exchange.getProperty(TENANT_NAME).toString(), "batch-transactions");
298-
299-
try {
300-
String tenantSpecificWorkflowId = bpmn.replace("{dfspid}", exchange.getProperty(TENANT_NAME).toString());
301-
String txnId = zeebeProcessStarter.startZeebeWorkflow(tenantSpecificWorkflowId, "", variables);
302-
if (txnId == null || txnId.isEmpty()) {
303-
response.put("errorCode", 500);
304-
response.put("errorDescription", "Unable to start zeebe workflow");
305-
response.put("developerMessage", "Issue in starting the zeebe workflow, check the zeebe configuration");
306-
} else {
307-
response.put("batch_id", batchId);
308-
response.put("request_id", requestId);
309-
response.put("status", "queued");
310-
}
311-
} catch (ClientStatusException c) {
312-
log.error("Got ClientStatusException : {}", c.getMessage());
313-
throw c;
314-
} catch (Exception e) {
315-
response.put("errorCode", 500);
316-
response.put("errorDescription", "Unable to start zeebe workflow");
317-
response.put("developerMessage", e.getLocalizedMessage());
318-
}
319-
320-
exchange.getIn().setBody(response.toString());
321-
322-
}).log("Completed route direct:start-batch-process-csv").bean(ProcessorStartRouteService.class, "pollingOutput");
167+
.to("direct:update-incoming-data").bean(ProcessorStartRouteService.class, "startBatchProcessCsv")
168+
.log("Completed route direct:start-batch-process-csv").bean(ProcessorStartRouteService.class, "pollingOutput");
323169

324170
from("direct:start-batch-process-raw").id("direct:start-batch-process-raw").log("Starting route direct:start-batch-process-raw")
325171
.process(exchange -> {
@@ -331,26 +177,9 @@ private void setup() {
331177
}).log("Completed route direct:start-batch-process-raw");
332178

333179
from("direct:executeBatch").id("direct:executeBatch").log("Starting route direct:executeBatch")
334-
.bean(ProcessorStartRouteService.class, "validateTenant").process(exchange -> {
335-
String filename = exchange.getIn().getHeader("filename", String.class);
336-
String requestId = exchange.getIn().getHeader("X-CorrelationID", String.class);
337-
String purpose = exchange.getIn().getHeader("Purpose", String.class);
338-
String type = exchange.getIn().getHeader("Type", String.class);
339-
String clientCorrelationId = exchange.getIn().getHeader(HEADER_CLIENT_CORRELATION_ID, String.class);
340-
String registeringInstitutionId = exchange.getIn().getHeader(HEADER_REGISTERING_INSTITUTE_ID, String.class);
341-
logger.info("registeringInstitutionId {}", registeringInstitutionId);
342-
String programId = exchange.getIn().getHeader(HEADER_PROGRAM_ID, String.class);
343-
String callbackUrl = exchange.getIn().getHeader("X-CallbackURL", String.class);
344-
exchange.setProperty(FILE_NAME, filename);
345-
exchange.setProperty(REQUEST_ID, requestId);
346-
exchange.setProperty(PURPOSE, purpose);
347-
exchange.setProperty(BATCH_REQUEST_TYPE, type);
348-
exchange.setProperty(CLIENT_CORRELATION_ID, clientCorrelationId);
349-
exchange.setProperty(REGISTERING_INSTITUTE_ID, registeringInstitutionId);
350-
exchange.setProperty(PROGRAM_ID, programId);
351-
exchange.setProperty(CALLBACK, callbackUrl);
352-
}).choice().when(exchange -> exchange.getProperty(BATCH_REQUEST_TYPE, String.class).equalsIgnoreCase("raw"))
353-
.to("direct:start-batch-process-raw")
180+
.bean(ProcessorStartRouteService.class, "validateTenant").bean(ProcessorStartRouteService.class, "executeBatch").choice()
181+
.when(exchange -> exchange.getProperty(BATCH_REQUEST_TYPE, String.class).equalsIgnoreCase("raw"))
182+
.bean(ProcessorStartRouteService.class, "startBatchProcessRaw")
354183
.when(exchange -> exchange.getProperty(BATCH_REQUEST_TYPE, String.class).equalsIgnoreCase("csv"))
355184
.to("direct:start-batch-process-csv").otherwise()
356185
.setBody(exchange -> getUnsupportedTypeJson(exchange.getProperty(BATCH_REQUEST_TYPE, String.class)).toString())
@@ -405,7 +234,7 @@ private void setup() {
405234

406235
}
407236

408-
private boolean verifyData(File file) throws IOException {
237+
public boolean verifyData(File file) throws IOException {
409238
InputStream ips = new FileInputStream(file);
410239
InputStreamReader ipsr = new InputStreamReader(ips);
411240
BufferedReader br = new BufferedReader(ipsr);
@@ -493,7 +322,7 @@ public void setResponse(Exchange exchange, int responseCode) {
493322
exchange.getIn().setHeader(Exchange.HTTP_RESPONSE_CODE, responseCode);
494323
}
495324

496-
private Map<String, Object> setConfigProperties(Map<String, Object> variables) {
325+
public Map<String, Object> setConfigProperties(Map<String, Object> variables) {
497326
variables.put(BATCH_AGGREGATE_ENABLED, workerConfig.isBatchAggregateEnabled);
498327
variables.put(PARTY_LOOKUP_ENABLED, workerConfig.isPartyLookUpWorkerEnabled);
499328
variables.put(AUTHORIZATION_ENABLED, workerConfig.isAuthorizationWorkerEnabled);

0 commit comments

Comments
 (0)