scheduler-service fix NullPointerException в backend-api.RequestInfoAccepter: улучшил редирект сообщения (временный статус)
This commit is contained in:
parent
e9e93ece7e
commit
64e6f932bc
1 changed files with 21 additions and 5 deletions
|
|
@ -19,6 +19,7 @@ import ru.spcex.clearing.platform.messaging.domain.cud.payment.PIClearingOutbond
|
||||||
import ru.spcex.clearing.platform.messaging.domain.cud.schedule.LauncherCommandRequest;
|
import ru.spcex.clearing.platform.messaging.domain.cud.schedule.LauncherCommandRequest;
|
||||||
import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
|
import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
|
||||||
import ru.spcex.clearing.platform.messaging.service.RequestInfoUpdate;
|
import ru.spcex.clearing.platform.messaging.service.RequestInfoUpdate;
|
||||||
|
import ru.spcex.clearing.platform.messaging.service.Status;
|
||||||
import ru.spcex.clearing.scheduler.error.ValidationError;
|
import ru.spcex.clearing.scheduler.error.ValidationError;
|
||||||
import ru.spcex.clearing.util.security.UserRoleVerification;
|
import ru.spcex.clearing.util.security.UserRoleVerification;
|
||||||
import ru.spcex.platform.enumeration.Task;
|
import ru.spcex.platform.enumeration.Task;
|
||||||
|
|
@ -53,12 +54,12 @@ public class LauncherService extends QueueConsumer implements InitializingBean {
|
||||||
@Override
|
@Override
|
||||||
public void afterPropertiesSet() {
|
public void afterPropertiesSet() {
|
||||||
callback(LauncherCommandRequest.class)
|
callback(LauncherCommandRequest.class)
|
||||||
.setConsumer(this::newLauncher)
|
.setFunction(this::newLauncher)
|
||||||
.forDestination(Consts.LAUNCHER_NEW, callbacks::put);
|
.forDestination(Consts.LAUNCHER_NEW, callbacks::put);
|
||||||
init();
|
init();
|
||||||
}
|
}
|
||||||
|
|
||||||
private void newLauncher(BaseRequest<LauncherCommandRequest> userRequest) {
|
private RequestInfoUpdate newLauncher(BaseRequest<LauncherCommandRequest> userRequest) {
|
||||||
LauncherCommandRequest req = userRequest.getRequestPayload();
|
LauncherCommandRequest req = userRequest.getRequestPayload();
|
||||||
log.debug("LauncherCommandRequest received");
|
log.debug("LauncherCommandRequest received");
|
||||||
|
|
||||||
|
|
@ -81,13 +82,14 @@ public class LauncherService extends QueueConsumer implements InitializingBean {
|
||||||
launcherMap.insert(launcher);
|
launcherMap.insert(launcher);
|
||||||
log.debug("successfully processed, new id {}", launcher.getId());
|
log.debug("successfully processed, new id {}", launcher.getId());
|
||||||
if (Task.dbfExport_OUTV.equalsByKey(req.getTaskName())) {
|
if (Task.dbfExport_OUTV.equalsByKey(req.getTaskName())) {
|
||||||
processRequestOUTV(userRequest, req);
|
return processRequestOUTV(userRequest, req);
|
||||||
} else {
|
} else {
|
||||||
kafkaProducer.send(new ProducerRecord<>("launcher-" + launcher.getTask(), userRequest));
|
kafkaProducer.send(new ProducerRecord<>("launcher-" + launcher.getTask(), userRequest));
|
||||||
}
|
}
|
||||||
|
return null;
|
||||||
}
|
}
|
||||||
|
|
||||||
private void processRequestOUTV(BaseRequest<LauncherCommandRequest> userRequest, LauncherCommandRequest launcherNew) {
|
private RequestInfoUpdate processRequestOUTV(BaseRequest<LauncherCommandRequest> userRequest, LauncherCommandRequest launcherNew) {
|
||||||
log.debug("Redirect {} (userRequest.id={}) to queue {}", launcherNew.getTaskName(),
|
log.debug("Redirect {} (userRequest.id={}) to queue {}", launcherNew.getTaskName(),
|
||||||
userRequest.getId(), Consts.PAYMENT_INSTRUCTION_CLEARING_OUTBOUND_ACTION);
|
userRequest.getId(), Consts.PAYMENT_INSTRUCTION_CLEARING_OUTBOUND_ACTION);
|
||||||
var req = new PIClearingOutbondActionNewRequest();
|
var req = new PIClearingOutbondActionNewRequest();
|
||||||
|
|
@ -109,12 +111,26 @@ public class LauncherService extends QueueConsumer implements InitializingBean {
|
||||||
try {
|
try {
|
||||||
send.get();
|
send.get();
|
||||||
log.debug("Request id={} by user {} send to queue \"{}\". Action: {}", request.getId(), request.getUserId(), destination, request.getActionType());
|
log.debug("Request id={} by user {} send to queue \"{}\". Action: {}", request.getId(), request.getUserId(), destination, request.getActionType());
|
||||||
} catch (InterruptedException e) {
|
return makeTempRoutingResponse(userRequest);
|
||||||
|
} catch (InterruptedException ei) {
|
||||||
log.error("Request id={} by user {} send to queue \"{}\". Action: {}; error send: thread interrupted", request.getId(), request.getUserId(), destination, request.getActionType());
|
log.error("Request id={} by user {} send to queue \"{}\". Action: {}; error send: thread interrupted", request.getId(), request.getUserId(), destination, request.getActionType());
|
||||||
Thread.currentThread().interrupt();
|
Thread.currentThread().interrupt();
|
||||||
|
throw new RuntimeException("Thread interrupted", ei);
|
||||||
} catch (ExecutionException e) {
|
} catch (ExecutionException e) {
|
||||||
log.error("Request id={} by user {} send to queue \"{}\". Action: {}; error send: {}", request.getId(), request.getUserId(), destination, request.getActionType(), ExceptionUtils.getStackTrace(e.getCause()));
|
log.error("Request id={} by user {} send to queue \"{}\". Action: {}; error send: {}", request.getId(), request.getUserId(), destination, request.getActionType(), ExceptionUtils.getStackTrace(e.getCause()));
|
||||||
|
throw new RuntimeException("Can not redirect to destination " + destination, e);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Требуется вернуть статус Processing, т.к. окончательный Success/Error будет потом
|
||||||
|
*/
|
||||||
|
RequestInfoUpdate makeTempRoutingResponse(BaseRequest<?> about) {
|
||||||
|
RequestInfoUpdate successRouting = new RequestInfoUpdate();
|
||||||
|
successRouting.setId(about.getId());
|
||||||
|
successRouting.setStatus(Status.Processing);
|
||||||
|
successRouting.setMessage(null);
|
||||||
|
return successRouting;
|
||||||
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue