statement service bugfix
This commit is contained in:
parent
99d141bf41
commit
eae1e816da
2 changed files with 2 additions and 24 deletions
|
|
@ -16,15 +16,13 @@ public class SdfExecutorsConfig {
|
||||||
Sdf57Executor sdf57Executor,
|
Sdf57Executor sdf57Executor,
|
||||||
Sdf04Executor sdf04Executor,
|
Sdf04Executor sdf04Executor,
|
||||||
Sdf13Executor sdf13Executor,
|
Sdf13Executor sdf13Executor,
|
||||||
Sdf08Executor sdf08Executor,
|
Sdf08Executor sdf08Executor) {
|
||||||
Sdf06Executor sdf06Executor) {
|
|
||||||
Map<SdfTable, AbstractExecutor<?>> executors = new HashMap<>();
|
Map<SdfTable, AbstractExecutor<?>> executors = new HashMap<>();
|
||||||
executors.put(SdfTable.SDF_01, sdf01Executor);
|
executors.put(SdfTable.SDF_01, sdf01Executor);
|
||||||
executors.put(SdfTable.SDF_57, sdf57Executor);
|
executors.put(SdfTable.SDF_57, sdf57Executor);
|
||||||
executors.put(SdfTable.SDF_04, sdf04Executor);
|
executors.put(SdfTable.SDF_04, sdf04Executor);
|
||||||
executors.put(SdfTable.SDF_13, sdf13Executor);
|
executors.put(SdfTable.SDF_13, sdf13Executor);
|
||||||
executors.put(SdfTable.SDF_08, sdf08Executor);
|
executors.put(SdfTable.SDF_08, sdf08Executor);
|
||||||
executors.put(SdfTable.SDF_06, sdf06Executor);
|
|
||||||
return executors;
|
return executors;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -62,31 +62,11 @@ public class StatementService extends QueueConsumer implements InitializingBean
|
||||||
@Override
|
@Override
|
||||||
public void afterPropertiesSet() throws Exception {
|
public void afterPropertiesSet() throws Exception {
|
||||||
callback(StatementRequest.class)
|
callback(StatementRequest.class)
|
||||||
.setConsumer(systemRequest -> {
|
.setConsumer(this::processPaired)
|
||||||
SdfTable table = systemRequest.getRequestPayload().getTable();
|
|
||||||
if (SdfTable.SDF_06.equals(table)) {
|
|
||||||
processSdf06(systemRequest);
|
|
||||||
} else {
|
|
||||||
processPaired(systemRequest);
|
|
||||||
}
|
|
||||||
})
|
|
||||||
.forDestination(Consts.STATEMENT_PROCESS, callbacks::put);
|
.forDestination(Consts.STATEMENT_PROCESS, callbacks::put);
|
||||||
init();
|
init();
|
||||||
}
|
}
|
||||||
|
|
||||||
@SuppressWarnings("unchecked")
|
|
||||||
private void processSdf06(BaseRequest<StatementRequest> systemRequest) {
|
|
||||||
Imdg<SDf06> sdfImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_SDf06, SDf06.class);
|
|
||||||
Collection<SDf06> sdf06Group = sdfImdg.getCollectionObjectsByFieldValues(Map.of(
|
|
||||||
"generationId", systemRequest.getRequestPayload().getGroupId()
|
|
||||||
));
|
|
||||||
AbstractExecutor<SDf06> service = (AbstractExecutor<SDf06>) executorsMap.get(SdfTable.SDF_06);
|
|
||||||
Result res = service.execute(sdf06Group, systemRequest.getRequestPayload());
|
|
||||||
if (service.isNeedToSendCommand()) {
|
|
||||||
service.sendCommand(kafkaSender, res);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private void processPaired(BaseRequest<StatementRequest> systemRequest) {
|
private void processPaired(BaseRequest<StatementRequest> systemRequest) {
|
||||||
StatementRequest statementRequest = systemRequest.getRequestPayload();
|
StatementRequest statementRequest = systemRequest.getRequestPayload();
|
||||||
SdfTable table = statementRequest.getTable();
|
SdfTable table = statementRequest.getTable();
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue