Skip to content

Commit 28b8ffb

Browse files
danishjamal104danishjamal104
authored
GOV-360: Enahnced the splitting logic to support subBatch data creation for individual txn (#126)
* Enahnced the splitting logic to support subBatch data creation for individual txn * Fixed the generate sub batch entity logic * Splitting log info * Update init sub batch logic to pass subbatchId in place of bathc id while calling channel api * Fixed the splittign logic * SUB_BATCH_ID variable removed for parent batch * Updated crucial data logging to debug * Checkstyle error fixed
1 parent 398050e commit 28b8ffb

6 files changed

Lines changed: 176 additions & 12 deletions

File tree

‎src/main/java/org/mifos/processor/bulk/camel/config/CamelProperties.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ private CamelProperties() {}
1919
public static final String SUB_BATCH_COUNT = "subBatchCount";
2020

2121
public static final String SUB_BATCH_CREATED = "subBatchCreated";
22+
public static final String SUB_BATCH_DETAILS = "subBatchDetails";
2223

2324
public static final String SERVER_SUB_BATCH_FILE_NAME_ARRAY = "serverSubBatchFileName";
2425

@@ -78,5 +79,5 @@ private CamelProperties() {}
7879
public static final String HEADER_PLATFORM_TENANT_ID = "Platform-TenantId";
7980
public static final String HEADER_CLIENT_CORRELATION_ID = "X-CorrelationID";
8081
public static final String CLIENT_CORRELATION_ID = "clientCorrelationId";
81-
82+
public static final String SUB_BATCH_ENTITY = "subBatchEntity";
8283
}

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

Lines changed: 10 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111
import static org.mifos.processor.bulk.camel.config.CamelProperties.PAYMENT_MODE_TYPE;
1212
import static org.mifos.processor.bulk.camel.config.CamelProperties.RESULT_TRANSACTION_LIST;
1313
import static org.mifos.processor.bulk.camel.config.CamelProperties.SERVER_FILE_NAME;
14+
import static org.mifos.processor.bulk.camel.config.CamelProperties.SUB_BATCH_ENTITY;
1415
import static org.mifos.processor.bulk.camel.config.CamelProperties.TENANT_NAME;
1516
import static org.mifos.processor.bulk.camel.config.CamelProperties.TRANSACTION_LIST;
1617
import static org.mifos.processor.bulk.camel.config.CamelProperties.TRANSACTION_LIST_ELEMENT;
@@ -27,21 +28,19 @@
2728
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.PURPOSE;
2829
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.REQUEST_ID;
2930
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.RESULT_FILE;
30-
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.SUB_BATCH_ID;
3131
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.TOTAL_AMOUNT;
3232

3333
import java.util.ArrayList;
34-
import java.util.HashMap;
3534
import java.util.List;
3635
import java.util.Map;
37-
import java.util.UUID;
3836
import java.util.function.Function;
3937
import org.apache.camel.Exchange;
4038
import org.apache.camel.LoggingLevel;
4139
import org.mifos.processor.bulk.config.ExternalApiPayloadConfig;
4240
import org.mifos.processor.bulk.config.PaymentModeConfiguration;
4341
import org.mifos.processor.bulk.config.PaymentModeMapping;
4442
import org.mifos.processor.bulk.config.PaymentModeType;
43+
import org.mifos.processor.bulk.schema.SubBatchEntity;
4544
import org.mifos.processor.bulk.schema.Transaction;
4645
import org.mifos.processor.bulk.schema.TransactionResult;
4746
import org.mifos.processor.bulk.utility.Utils;
@@ -85,9 +84,8 @@ public void configure() throws Exception {
8584
.process(exchange -> {
8685
List<Transaction> transactionList = exchange.getProperty(TRANSACTION_LIST, List.class);
8786

88-
Map<String, Object> variables = new HashMap<>();
87+
Map<String, Object> variables = exchange.getProperty(ZEEBE_VARIABLE, Map.class);
8988
variables.put(BATCH_ID, exchange.getProperty(BATCH_ID));
90-
variables.put(SUB_BATCH_ID, UUID.randomUUID().toString());
9189
variables.put(FILE_NAME, exchange.getProperty(SERVER_FILE_NAME));
9290
variables.put(REQUEST_ID, exchange.getProperty(REQUEST_ID));
9391
variables.put(PURPOSE, exchange.getProperty(PURPOSE));
@@ -190,8 +188,13 @@ public void configure() throws Exception {
190188
}
191189
}).choice().when(exchangeProperty(EXTERNAL_ENDPOINT_FAILED).isEqualTo(false))
192190
.log(LoggingLevel.DEBUG, "Making API call to endpoint ${exchangeProperty.extEndpoint} and body: ${body}")
193-
.setHeader(Exchange.CONTENT_TYPE, constant("application/json"))
194-
.setHeader(BATCH_ID_HEADER, simple("${exchangeProperty." + BATCH_ID + "}"))
191+
.setHeader(Exchange.CONTENT_TYPE, constant("application/json")).choice()
192+
.when(exchange -> exchange.getProperty(SUB_BATCH_ENTITY, SubBatchEntity.class) != null)
193+
.log("Sub batch entity is not null, hence passing subBatchId while calling channel API").process(exchange -> {
194+
SubBatchEntity subBatchEntity = exchange.getProperty(SUB_BATCH_ENTITY, SubBatchEntity.class);
195+
exchange.getIn().setHeader(BATCH_ID_HEADER, subBatchEntity.getSubBatchId());
196+
}).otherwise().log("Sub batch entity is null, hence passing batchId while calling channel API")
197+
.setHeader(BATCH_ID_HEADER, simple("${exchangeProperty." + BATCH_ID + "}")).endChoice()
195198
.setHeader(HEADER_CLIENT_CORRELATION_ID, simple("${exchangeProperty." + REQUEST_ID + "}"))
196199
.setHeader(HEADER_REGISTERING_INSTITUTE_ID, simple("${exchangeProperty." + HEADER_REGISTERING_INSTITUTE_ID + "}"))
197200
.process(exchange -> {

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

Lines changed: 75 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,18 +1,32 @@
11
package org.mifos.processor.bulk.camel.routes;
22

33
import static org.mifos.processor.bulk.camel.config.CamelProperties.LOCAL_FILE_PATH;
4+
import static org.mifos.processor.bulk.camel.config.CamelProperties.REGISTERING_INSTITUTE_ID;
45
import static org.mifos.processor.bulk.camel.config.CamelProperties.SERVER_FILE_NAME;
56
import static org.mifos.processor.bulk.camel.config.CamelProperties.SERVER_SUB_BATCH_FILE_NAME_ARRAY;
67
import static org.mifos.processor.bulk.camel.config.CamelProperties.SUB_BATCH_COUNT;
78
import static org.mifos.processor.bulk.camel.config.CamelProperties.SUB_BATCH_CREATED;
9+
import static org.mifos.processor.bulk.camel.config.CamelProperties.SUB_BATCH_DETAILS;
810
import static org.mifos.processor.bulk.camel.config.CamelProperties.SUB_BATCH_FILE_ARRAY;
11+
import static org.mifos.processor.bulk.camel.config.CamelProperties.TRANSACTION_LIST;
12+
import static org.mifos.processor.bulk.camel.config.CamelProperties.ZEEBE_VARIABLE;
13+
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.BATCH_ID;
14+
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.CLIENT_CORRELATION_ID;
15+
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.PAYER_IDENTIFIER;
16+
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.REQUEST_ID;
917
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.SPLITTING_FAILED;
1018

1119
import java.io.BufferedReader;
1220
import java.io.FileReader;
1321
import java.io.FileWriter;
1422
import java.util.ArrayList;
23+
import java.util.Date;
1524
import java.util.List;
25+
import java.util.Map;
26+
import java.util.UUID;
27+
import org.apache.camel.LoggingLevel;
28+
import org.mifos.processor.bulk.schema.SubBatchEntity;
29+
import org.mifos.processor.bulk.schema.Transaction;
1630
import org.springframework.beans.factory.annotation.Value;
1731
import org.springframework.stereotype.Component;
1832

@@ -56,7 +70,7 @@ public void configure() throws Exception {
5670
List<String> subBatchFile = new ArrayList<>();
5771
int subBatchCount = 1;
5872
for (int i = 0; i < lines.size(); i += subBatchSize) {
59-
String filename = System.currentTimeMillis() + "_" + "sub-batch-" + subBatchCount + ".csv";
73+
String filename = UUID.randomUUID() + "_" + "sub-batch-" + subBatchCount + ".csv";
6074
FileWriter writer = new FileWriter(filename);
6175
writer.write(header);
6276
for (int j = i; j < Math.min(i + subBatchSize, lines.size()); j++) {
@@ -77,13 +91,71 @@ public void configure() throws Exception {
7791
from("direct:upload-sub-batch-file").id("direct:upload-sub-batch-file").log("Starting upload of sub-batch file")
7892
.loopDoWhile(exchange -> exchange.getProperty(SUB_BATCH_FILE_ARRAY, List.class).size() > 0).process(exchange -> {
7993
List<String> subBatchFile = exchange.getProperty(SUB_BATCH_FILE_ARRAY, List.class);
80-
exchange.setProperty(LOCAL_FILE_PATH, subBatchFile.remove(0));
94+
String localFilePath = subBatchFile.remove(0);
95+
exchange.setProperty(LOCAL_FILE_PATH, localFilePath);
8196
exchange.setProperty(SUB_BATCH_FILE_ARRAY, subBatchFile);
82-
}).to("direct:upload-file").process(exchange -> {
97+
logger.debug("Local file path: {}", localFilePath);
98+
logger.debug("Sub batch file array: {}, ", subBatchFile);
99+
}).log(LoggingLevel.DEBUG, "LOCAL_FILE_PATH: ${exchangeProperty." + LOCAL_FILE_PATH + "}")
100+
.to("direct:generate-sub-batch-entity").log("direct:generate-sub-batch-entity completed").to("direct:upload-file")
101+
.process(exchange -> {
83102
String serverFilename = exchange.getProperty(SERVER_FILE_NAME, String.class);
84103
List<String> serverSubBatchFile = exchange.getProperty(SERVER_SUB_BATCH_FILE_NAME_ARRAY, List.class);
85104
serverSubBatchFile.add(serverFilename);
86105
exchange.setProperty(SERVER_SUB_BATCH_FILE_NAME_ARRAY, serverSubBatchFile);
106+
logger.debug("Server subbatch filename array: {}", serverSubBatchFile);
87107
});
108+
109+
// generate subBatchEntityDetails, make sure [LOCAL_FILE_PATH] has the absolute sub batch file path
110+
from("direct:generate-sub-batch-entity").id("direct:generate-sub-batch-entity").log("Generating sub batch entity")
111+
.to("direct:get-transaction-array").process(exchange -> {
112+
List<Transaction> transactionList = exchange.getProperty(TRANSACTION_LIST, List.class);
113+
Map<String, Object> zeebeVariables = exchange.getProperty(ZEEBE_VARIABLE, Map.class);
114+
String serverFileName = exchange.getProperty(LOCAL_FILE_PATH, String.class);
115+
116+
logger.info("Generating sub batch entity for file {}", serverFileName);
117+
if (transactionList.isEmpty()) {
118+
logger.info("Transaction list is empty");
119+
return;
120+
}
121+
122+
Long totalAmount = getTotalAmount(transactionList);
123+
124+
SubBatchEntity subBatchEntity = getDefaultSubBatchEntity();
125+
subBatchEntity.setBatchId((String) zeebeVariables.get(BATCH_ID));
126+
subBatchEntity.setSubBatchId(UUID.randomUUID().toString());
127+
subBatchEntity.setRequestId((String) zeebeVariables.get(REQUEST_ID));
128+
subBatchEntity.setCorrelationId((String) zeebeVariables.get(CLIENT_CORRELATION_ID));
129+
subBatchEntity.setPayerFsp((String) zeebeVariables.get(PAYER_IDENTIFIER));
130+
subBatchEntity.setRegisteringInstitutionId((String) zeebeVariables.get(REGISTERING_INSTITUTE_ID));
131+
subBatchEntity.setPaymentMode(transactionList.get(0).getPaymentMode());
132+
subBatchEntity.setRequestFile(serverFileName);
133+
subBatchEntity.setTotalTransactions((long) transactionList.size());
134+
subBatchEntity.setOngoing((long) transactionList.size());
135+
subBatchEntity.setTotalAmount(totalAmount);
136+
subBatchEntity.setOngoingAmount(totalAmount);
137+
subBatchEntity.setStartedAt(new Date(System.currentTimeMillis()));
138+
139+
logger.debug("SubBatchEntity: {}", objectMapper.writeValueAsString(subBatchEntity));
140+
// update the sub batch details array
141+
List<SubBatchEntity> subBatchEntityList = exchange.getProperty(SUB_BATCH_DETAILS, List.class);
142+
subBatchEntityList.add(subBatchEntity);
143+
exchange.setProperty(SUB_BATCH_DETAILS, subBatchEntityList);
144+
logger.debug("generate-sub-batch-entity route end: {}", objectMapper.writeValueAsString(subBatchEntityList));
145+
});
146+
}
147+
148+
private SubBatchEntity getDefaultSubBatchEntity() {
149+
SubBatchEntity subBatchEntity = new SubBatchEntity();
150+
subBatchEntity.setAllEmptyAmount();
151+
return subBatchEntity;
152+
}
153+
154+
private long getTotalAmount(List<Transaction> transactionList) {
155+
long totalAmount = 0L;
156+
for (Transaction transaction : transactionList) {
157+
totalAmount += Long.parseLong(transaction.getAmount());
158+
}
159+
return totalAmount;
88160
}
89161
}

‎src/main/java/org/mifos/processor/bulk/schema/SubBatchEntity.java‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
1+
package org.mifos.processor.bulk.schema;
2+
3+
import com.fasterxml.jackson.annotation.JsonIgnore;
4+
import java.util.Date;
5+
import lombok.Getter;
6+
import lombok.NoArgsConstructor;
7+
import lombok.Setter;
8+
9+
@Getter
10+
@Setter
11+
@NoArgsConstructor
12+
public class SubBatchEntity {
13+
14+
private String batchId;
15+
private String subBatchId;
16+
private String requestId;
17+
private String requestFile;
18+
private String resultFile;
19+
private String note;
20+
private String paymentMode;
21+
private String registeringInstitutionId;
22+
private String payerFsp;
23+
private String correlationId;
24+
25+
private Long totalTransactions;
26+
private Long ongoing;
27+
private Long failed;
28+
private Long completed;
29+
private Long totalAmount;
30+
private Long ongoingAmount;
31+
private Long failedAmount;
32+
private Long completedAmount;
33+
private Long workflowKey;
34+
private Long workflowInstanceKey;
35+
private Long approvedAmount;
36+
private Long approvedCount;
37+
38+
private Date resultGeneratedAt;
39+
private Date startedAt;
40+
private Date completedAt;
41+
42+
@JsonIgnore
43+
public void setAllEmptyAmount() {
44+
setTotalTransactions(0L);
45+
setOngoing(0L);
46+
setFailed(0L);
47+
setCompleted(0L);
48+
setTotalAmount(0L);
49+
setOngoingAmount(0L);
50+
setFailedAmount(0L);
51+
setCompletedAmount(0L);
52+
setApprovedAmount(0L);
53+
setApprovedCount(0L);
54+
}
55+
}

‎src/main/java/org/mifos/processor/bulk/zeebe/worker/InitSubBatchWorker.java‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,10 @@
11
package org.mifos.processor.bulk.zeebe.worker;
22

33
import static org.mifos.processor.bulk.camel.config.CamelProperties.SERVER_FILE_NAME;
4+
import static org.mifos.processor.bulk.camel.config.CamelProperties.SUB_BATCH_DETAILS;
5+
import static org.mifos.processor.bulk.camel.config.CamelProperties.SUB_BATCH_ENTITY;
46
import static org.mifos.processor.bulk.camel.config.CamelProperties.TENANT_NAME;
7+
import static org.mifos.processor.bulk.camel.config.CamelProperties.ZEEBE_VARIABLE;
58
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.BATCH_ID;
69
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.FILE_NAME;
710
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.INIT_FAILURE_SUB_BATCHES;
@@ -14,17 +17,24 @@
1417
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.SUB_BATCHES;
1518
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.TENANT_ID;
1619

20+
import com.fasterxml.jackson.core.type.TypeReference;
21+
import com.fasterxml.jackson.databind.ObjectMapper;
1722
import java.util.ArrayList;
1823
import java.util.List;
1924
import java.util.Map;
2025
import org.apache.camel.Exchange;
2126
import org.apache.camel.support.DefaultExchange;
2227
import org.mifos.processor.bulk.camel.routes.RouteId;
28+
import org.mifos.processor.bulk.schema.SubBatchEntity;
29+
import org.springframework.beans.factory.annotation.Autowired;
2330
import org.springframework.stereotype.Component;
2431

2532
@Component
2633
public class InitSubBatchWorker extends BaseWorker {
2734

35+
@Autowired
36+
private ObjectMapper objectMapper;
37+
2838
@Override
2939
public void setup() {
3040

@@ -55,14 +65,30 @@ public void setup() {
5565
subBatches.add((String) variables.get(FILE_NAME));
5666
}
5767

68+
List<Object> subBatchObjectList = (List<Object>) variables.get(SUB_BATCH_DETAILS);
69+
logger.debug("Subbatch entity list in init sub batch worker: {}", subBatchObjectList);
70+
71+
List<SubBatchEntity> subBatchEntityList = objectMapper.convertValue(subBatchObjectList, new TypeReference<>() {});
72+
5873
String fileName = subBatches.remove(0);
74+
SubBatchEntity subBatchEntity = null;
75+
76+
for (SubBatchEntity subBatch : subBatchEntityList) {
77+
if (subBatch.getRequestFile().contains(fileName)) {
78+
subBatchEntity = subBatch;
79+
logger.info("SubBatchEntity found");
80+
}
81+
}
82+
logger.debug("BatchEntity for this subbatch is {}", objectMapper.writeValueAsString(subBatchEntity));
5983

6084
Exchange exchange = new DefaultExchange(camelContext);
6185
exchange.setProperty(TENANT_NAME, variables.get(TENANT_ID));
6286
exchange.setProperty(SERVER_FILE_NAME, fileName);
6387
exchange.setProperty(BATCH_ID, variables.get(BATCH_ID));
6488
exchange.setProperty(REQUEST_ID, variables.get(REQUEST_ID));
6589
exchange.setProperty(PURPOSE, variables.get(PURPOSE));
90+
exchange.setProperty(ZEEBE_VARIABLE, variables);
91+
exchange.setProperty(SUB_BATCH_ENTITY, subBatchEntity);
6692

6793
sendToCamelRoute(RouteId.INIT_SUB_BATCH, exchange);
6894

‎src/main/java/org/mifos/processor/bulk/zeebe/worker/SplittingWorker.java‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,8 @@
33
import static org.mifos.processor.bulk.camel.config.CamelProperties.SERVER_FILE_NAME;
44
import static org.mifos.processor.bulk.camel.config.CamelProperties.SERVER_SUB_BATCH_FILE_NAME_ARRAY;
55
import static org.mifos.processor.bulk.camel.config.CamelProperties.SUB_BATCH_CREATED;
6+
import static org.mifos.processor.bulk.camel.config.CamelProperties.SUB_BATCH_DETAILS;
7+
import static org.mifos.processor.bulk.camel.config.CamelProperties.ZEEBE_VARIABLE;
68
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.FILE_NAME;
79
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.INIT_FAILURE_SUB_BATCHES;
810
import static org.mifos.processor.bulk.zeebe.ZeebeVariables.INIT_SUCCESS_SUB_BATCHES;
@@ -15,6 +17,7 @@
1517
import org.apache.camel.Exchange;
1618
import org.apache.camel.support.DefaultExchange;
1719
import org.mifos.processor.bulk.camel.routes.RouteId;
20+
import org.mifos.processor.bulk.schema.SubBatchEntity;
1821
import org.springframework.stereotype.Component;
1922

2023
@Component
@@ -29,7 +32,7 @@ public void setup() {
2932
* zeebeVariable [SPLITTING_FAILED, SUB_BATCHES, SUB_BATCH_CREATED]
3033
*/
3134
newWorker(Worker.SPLITTING, (client, job) -> {
32-
logger.debug("Job '{}' started from process '{}' with key {}", job.getType(), job.getBpmnProcessId(), job.getKey());
35+
logger.info("Job '{}' started from process '{}' with key {}", job.getType(), job.getBpmnProcessId(), job.getKey());
3336
Map<String, Object> variables = job.getVariablesAsMap();
3437
if (workerConfig.isSplittingWorkerEnabled) {
3538
variables.put(SPLITTING_FAILED, false);
@@ -38,6 +41,8 @@ public void setup() {
3841
String filename = (String) variables.get(FILE_NAME);
3942
Exchange exchange = new DefaultExchange(camelContext);
4043
exchange.setProperty(SERVER_FILE_NAME, filename);
44+
exchange.setProperty(ZEEBE_VARIABLE, variables);
45+
exchange.setProperty(SUB_BATCH_DETAILS, new ArrayList<SubBatchEntity>());
4146

4247
try {
4348
sendToCamelRoute(RouteId.SPLITTING, exchange);
@@ -56,11 +61,13 @@ public void setup() {
5661

5762
variables.put(SPLITTING_FAILED, false);
5863
variables.put(SUB_BATCHES, serverSubBatchFileList);
64+
variables.put(SUB_BATCH_DETAILS, exchange.getProperty(SUB_BATCH_DETAILS, ArrayList.class));
5965
variables.put(INIT_SUCCESS_SUB_BATCHES, new ArrayList<String>());
6066
variables.put(INIT_FAILURE_SUB_BATCHES, new ArrayList<String>());
6167
variables.put(SUB_BATCH_CREATED, subBatchCreated);
6268

6369
client.newCompleteCommand(job.getKey()).variables(variables).send();
70+
logger.info("Splitting worker completed");
6471
});
6572
}
6673

0 commit comments

Comments
 (0)