diff --git a/clearing-parent/scheduler-service/src/main/java/ru/spcex/clearing/scheduler/service/LauncherService.java b/clearing-parent/scheduler-service/src/main/java/ru/spcex/clearing/scheduler/service/LauncherService.java index 3c08e822b..6f0185097 100644 --- a/clearing-parent/scheduler-service/src/main/java/ru/spcex/clearing/scheduler/service/LauncherService.java +++ b/clearing-parent/scheduler-service/src/main/java/ru/spcex/clearing/scheduler/service/LauncherService.java @@ -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.service.QueueConsumer; 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.util.security.UserRoleVerification; import ru.spcex.platform.enumeration.Task; @@ -53,12 +54,12 @@ public class LauncherService extends QueueConsumer implements InitializingBean { @Override public void afterPropertiesSet() { callback(LauncherCommandRequest.class) - .setConsumer(this::newLauncher) + .setFunction(this::newLauncher) .forDestination(Consts.LAUNCHER_NEW, callbacks::put); init(); } - private void newLauncher(BaseRequest userRequest) { + private RequestInfoUpdate newLauncher(BaseRequest userRequest) { LauncherCommandRequest req = userRequest.getRequestPayload(); log.debug("LauncherCommandRequest received"); @@ -81,13 +82,14 @@ public class LauncherService extends QueueConsumer implements InitializingBean { launcherMap.insert(launcher); log.debug("successfully processed, new id {}", launcher.getId()); if (Task.dbfExport_OUTV.equalsByKey(req.getTaskName())) { - processRequestOUTV(userRequest, req); + return processRequestOUTV(userRequest, req); } else { kafkaProducer.send(new ProducerRecord<>("launcher-" + launcher.getTask(), userRequest)); } + return null; } - private void processRequestOUTV(BaseRequest userRequest, LauncherCommandRequest launcherNew) { + private RequestInfoUpdate processRequestOUTV(BaseRequest userRequest, LauncherCommandRequest launcherNew) { log.debug("Redirect {} (userRequest.id={}) to queue {}", launcherNew.getTaskName(), userRequest.getId(), Consts.PAYMENT_INSTRUCTION_CLEARING_OUTBOUND_ACTION); var req = new PIClearingOutbondActionNewRequest(); @@ -109,12 +111,26 @@ public class LauncherService extends QueueConsumer implements InitializingBean { try { send.get(); 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()); Thread.currentThread().interrupt(); + throw new RuntimeException("Thread interrupted", ei); } 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())); + 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; + } + }