http://jira.mfd.msk:8088/browse/CLS-151

clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/config/KafkaConfig.java

@@ -8,6 +8,7 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Scope;
 import ru.spcex.clearing.account.config.settings.AccountServiceSettings;
+import ru.spcex.clearing.imdg.IMDGDistributedNames;
 import ru.spcex.clearing.platform.messaging.config.KafkaConsumerFactory;
 import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory;
 import ru.spcex.clearing.platform.messaging.service.RequestInfo;
@@ -41,7 +42,7 @@ public class KafkaConfig {
                 .producer(kafkaProducer)
                 .idGenerator(imdgIdGenerator::nextId)
                 .imdgProvider(s -> {
-                    Imdg<RequestInfo> imdg = imdgProvider.getImdg(s, RequestInfo.class);
+                    Imdg<RequestInfo> imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class);
                     return imdg::insert;
                 })
                 .build();

clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/queue/QueueExceptionHandler.java

@@ -39,7 +39,7 @@ public class QueueExceptionHandler extends ResponseEntityExceptionHandler {
     @ResponseStatus(HttpStatus.BAD_REQUEST)
     @ResponseBody
     public BasicSpcexResponse handleValidationException(ActionValidationException e) {
-        log.error("Default scoring exception handler: {}", ExceptionUtils.getStackTrace(e));
+        log.error("Default exception handler: {}", ExceptionUtils.getStackTrace(e));
         EnumMessage firstError = e.getErrors().iterator().next();
         BasicSpcexResponse errorResponse = new BasicSpcexResponse();
         errorResponse.setCode(firstError.getSubject().getId());
@@ -68,7 +68,7 @@ public class QueueExceptionHandler extends ResponseEntityExceptionHandler {
     @ResponseStatus(HttpStatus.INTERNAL_SERVER_ERROR)
     @ResponseBody
     public BasicSpcexResponse handleDefault(Throwable e) {
-        log.error("Default scoring exception handler: {}", ExceptionUtils.getStackTrace(e));
+        log.error("Default exception handler: {}", ExceptionUtils.getStackTrace(e));
         BasicSpcexResponse errorResponse = new BasicSpcexResponse();
         errorResponse.setCode(-1);
         errorResponse.setMessage("internal server error");

clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/response/system/RequestInfoResponse.java

@@ -0,0 +1,51 @@
+package ru.spcex.clearing.backendapi.controller.response.system;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import ru.spcex.clearing.backendapi.controller.response.BasicSpcexResponse;
+import ru.spcex.clearing.platform.messaging.domain.json.serialize.EnumSerializer;
+import ru.spcex.clearing.platform.messaging.service.Status;
+
+
+/**
+ * Банковские реквизиты для перечисления денежных средств
+ * @ApiModel(description = "Ответ в результате отправки операции в топик Kafka.")
+ * @ApiModelProperty(value = "Информация о принятом запросе")
+ **/
+@ApiModel(description = "Ответ при получении статуса запроса.")
+public class RequestInfoResponse extends BasicSpcexResponse {
+
+    @JsonProperty
+    @ApiModelProperty(value = "Поля объекта")
+    private RequestInfoPayload payload;
+
+    public RequestInfoPayload getPayload() {
+        return payload;
+    }
+
+    public void setStatus(Status status) {
+        this.payload = new RequestInfoPayload();
+        payload.setStatus(status);
+    }
+
+    public void setPayload(RequestInfoPayload payload) {
+        this.payload = payload;
+    }
+
+    private static class RequestInfoPayload {
+        @JsonProperty
+        @JsonSerialize(using = EnumSerializer.class)
+        private Status status;
+
+        public Status getStatus() {
+            return status;
+        }
+
+        public void setStatus(Status status) {
+            this.status = status;
+        }
+    }
+
+}

clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/system/RequestStatusController.java

@@ -0,0 +1,48 @@
+package ru.spcex.clearing.backendapi.controller.system;
+
+import io.swagger.annotations.ApiOperation;
+import io.swagger.annotations.ApiParam;
+import io.swagger.annotations.ApiResponse;
+import io.swagger.annotations.ApiResponses;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Controller;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.bind.annotation.ResponseBody;
+import ru.spcex.clearing.backendapi.controller.response.BasicSpcexResponse;
+import ru.spcex.clearing.backendapi.controller.response.system.RequestInfoResponse;
+import ru.spcex.clearing.imdg.IMDGDistributedNames;
+import ru.spcex.clearing.platform.messaging.service.RequestInfo;
+import ru.spcex.platform.imdg.api.Imdg;
+import ru.spcex.platform.imdg.api.ImdgProvider;
+
+@Controller
+@RequestMapping("/requests")
+public class RequestStatusController {
+
+    private final Imdg<RequestInfo> imdg;
+
+    @Autowired
+    public RequestStatusController(ImdgProvider imdgProvider) {
+        this.imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class);
+    }
+
+    @ApiOperation(value = "get request status.")
+    @ApiResponses(value = {@ApiResponse(code = 200, message = "OK", response = RequestInfoResponse.class), @ApiResponse(code = 400, message = "Ошибка валидации", response = BasicSpcexResponse.class)})
+    @RequestMapping(value = "/{id}", method = RequestMethod.GET)
+    @ResponseBody
+    public RequestInfoResponse getById(@ApiParam(value = "Идентификатор объекта", required = true, example = "1234")
+                              @PathVariable("id") Long id) {
+        RequestInfo singleObjectByID;
+        if (id == null || (singleObjectByID = imdg.getSingleObjectByID(id)) == null) {
+            RequestInfoResponse response = new RequestInfoResponse();
+            response.setCode(-1L);
+            response.setMessage("cannot find request with id='" + id + "'.");
+            return response;
+        }
+        RequestInfoResponse response = new RequestInfoResponse();
+        response.setStatus(singleObjectByID.getStatus());
+        return response;
+    }
+}

clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/service/impl/OperatorImpl.java

@@ -8,6 +8,7 @@ import ru.spcex.clearing.backendapi.controller.response.cud.QueueSuccessResponse
 import ru.spcex.clearing.backendapi.domain.actions.IAction;
 import ru.spcex.clearing.backendapi.errors.ActionValidationException;
 import ru.spcex.clearing.backendapi.service.IOperator;
+import ru.spcex.clearing.imdg.IMDGDistributedNames;
 import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
 import ru.spcex.clearing.platform.messaging.service.RequestInfo;
 import ru.spcex.platform.imdg.api.Imdg;
@@ -46,7 +47,7 @@ public class OperatorImpl implements IOperator {
     }
 
     private void saveRequestToStorage(String destination, BaseRequest<Object> request) {
-        Imdg<RequestInfo> requestStorage = imdgProvider.getImdg(destination, RequestInfo.class);
+        Imdg<RequestInfo> requestStorage = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class);
         RequestInfo requestInfo = RequestInfo.create(request.getId());
         requestStorage.insert(requestInfo);
     }

clearing-parent/balance-service/src/main/java/ru/spcex/clearing/balance/config/KafkaSenderConfig.java

@@ -4,6 +4,7 @@ import org.apache.kafka.clients.producer.Producer;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import ru.spcex.clearing.imdg.IMDGDistributedNames;
 import ru.spcex.clearing.platform.messaging.service.RequestInfo;
 import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
 import ru.spcex.platform.imdg.api.Imdg;
@@ -22,7 +23,7 @@ public class KafkaSenderConfig {
                 .producer(kafkaProducer)
                 .idGenerator(imdgIdGenerator::nextId)
                 .imdgProvider(s -> {
-                    Imdg<RequestInfo> imdg = imdgProvider.getImdg(s, RequestInfo.class);
+                    Imdg<RequestInfo> imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class);
                     return imdg::insert;
                 })
                 .build();

clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/config/KafkaConfig.java

@@ -5,6 +5,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import ru.spcex.clearing.dbf.importer.config.settings.ImportDBFServiceSettings;
+import ru.spcex.clearing.imdg.IMDGDistributedNames;
 import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory;
 import ru.spcex.clearing.platform.messaging.service.RequestInfo;
 import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
@@ -28,7 +29,7 @@ public class KafkaConfig {
                     .producer(kafkaProducer)
                     .idGenerator(imdgIdGenerator::nextId)
                     .imdgProvider(s -> {
-                        Imdg<RequestInfo> imdg = imdgProvider.getImdg(s, RequestInfo.class);
+                        Imdg<RequestInfo> imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class);
                         return imdg::insert;
                     })
                     .build();

platform-parent/platform-imdg-api/src/main/java/ru/spcex/clearing/imdg/IMDGDistributedNames.java

@@ -101,6 +101,7 @@ public final class IMDGDistributedNames {
     public static final String Map_AccountStatusDictionary = "Map_AccountStatusDictionary";
     public static final String Map_KeyRate = "Map_KeyRate";
     public static final String Map_MoneyMarketSecurity = "Map_MoneyMarketSecurity";
+    public static final String Map_RequestInfo = "Map_RequestInfo";
     public static final String MAP_SEQUENCE_NAME = "MAP_SEQUENCE_NAME"; // todo вынести idGenerator отдельно и завернуть в метод, чтобы не напрямую обращаться.
 
     private IMDGDistributedNames() {

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java

@@ -33,4 +33,6 @@ public interface Consts {
     String ACCOUNT_NEW = "account-new";
     String BALANCE_ACCOUNT_NEW = "balance-account-new";
 
+    String REQUEST_INFO_UPDATE = "request-info-update";
+
 }

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/json/serialize/EnumSerializer.java

@@ -0,0 +1,19 @@
+package ru.spcex.clearing.platform.messaging.domain.json.serialize;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.SerializerProvider;
+
+import java.io.IOException;
+import java.time.format.DateTimeFormatter;
+
+public class EnumSerializer extends JsonSerializer<Enum<?>> {
+    private static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");
+
+
+    @Override
+    public void serialize(Enum<?> value, JsonGenerator gen, SerializerProvider serializers) throws IOException {
+        if (value == null) return;
+        gen.writeString(value.name());
+    }
+}

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/logic/functional/BuilderConsumerStep.java

@@ -3,7 +3,9 @@ package ru.spcex.clearing.platform.messaging.logic.functional;
 import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
 
 import java.util.function.Consumer;
+import java.util.function.Function;
 
 public interface BuilderConsumerStep<T1> {
     BuilderDestinationStep setConsumer(Consumer<BaseRequest<T1>> consumer);
+    BuilderDestinationStep setFunction(Function<BaseRequest<T1>, Object> function);
 }

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/logic/functional/ConsumerSpecificClass.java

@@ -4,6 +4,7 @@ import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
 
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
+import java.util.function.Function;
 
 /**
  * вспомогательный класс
@@ -18,6 +19,7 @@ public class ConsumerSpecificClass<T> implements BuilderConsumerStep<T>, Builder
      * действие-обработчик реквеста (метод из конкретного менеджера)
      */
     private java.util.function.Consumer<BaseRequest<T>> consumer;
+    private java.util.function.Function<BaseRequest<T>, Object> function;
 
     private ConsumerSpecificClass(Class<T> clazz) {
         this.clazz = clazz;
@@ -29,7 +31,17 @@ public class ConsumerSpecificClass<T> implements BuilderConsumerStep<T>, Builder
 
     @Override
     public BuilderDestinationStep setConsumer(Consumer<BaseRequest<T>> consumer) {
-        this.consumer = consumer;
+        if (this.consumer == null) {
+            this.consumer = consumer;
+        } else {
+            this.consumer = this.consumer.andThen(consumer);
+        }
+        return this;
+    }
+
+    @Override
+    public BuilderDestinationStep setFunction(Function<BaseRequest<T>, Object> function) {
+        this.function = function;
         return this;
     }
 
@@ -38,10 +50,18 @@ public class ConsumerSpecificClass<T> implements BuilderConsumerStep<T>, Builder
     }
 
     @SuppressWarnings("unchecked")
-    public void acceptRaw(Object obj) {
+    public Object acceptRaw(Object obj) {
-        this.consumer.accept((BaseRequest<T>) obj);
+        if (this.consumer != null) {
+            this.consumer.accept((BaseRequest<T>) obj);
+            return null;
+        } else {
+            return this.function.apply((BaseRequest<T>) obj);
+        }
     }
 
+    public Consumer<BaseRequest<T>> getConsumer() {
+        return consumer;
+    }
     public Class<T> getClazz() {
         return clazz;
     }

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/QueueConsumer.java

@@ -5,10 +5,14 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.errors.WakeupException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
+import ru.spcex.clearing.platform.messaging.domain.Consts;
 import ru.spcex.clearing.platform.messaging.logic.functional.BuilderConsumerStep;
 import ru.spcex.clearing.platform.messaging.logic.functional.ConsumerSpecificClass;
 import ru.spcex.platform.utils.log.ExceptionUtils;
@@ -19,6 +23,7 @@ import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
@@ -28,33 +33,40 @@ public class QueueConsumer implements AutoCloseable {
     private final Logger log = LoggerFactory.getLogger(getClass());
     private final AtomicBoolean closed = new AtomicBoolean(false);
     private final Consumer<String, Object> consumer;
-    private final ExecutorService executor;
+    private Producer<String, Object> producer;
+    private final ExecutorService inputExecutor;
+    private final ExecutorService outputExecutor;
     protected final Map<String, ConsumerSpecificClass<?>> callbacks;
     private final ObjectMapper json;
+    protected boolean supportStartOffsetTimeWindow;
 
     public QueueConsumer(Consumer<String, Object> kafkaQueue) {
         this.consumer = kafkaQueue;
         this.callbacks = new HashMap<>();
-        this.executor = Executors.newSingleThreadExecutor();
+        this.inputExecutor = Executors.newSingleThreadExecutor();
+        this.outputExecutor = Executors.newSingleThreadExecutor();
         this.json = new ObjectMapper();
+        this.supportStartOffsetTimeWindow = false;
+    }
+
+    public QueueConsumer(Consumer<String, Object> kafkaQueue, Producer<String, Object> kafkaResponseQueue) {
+        this(kafkaQueue);
+        this.producer = kafkaResponseQueue;
     }
 
     protected boolean needsProcessing(String topicName, BaseRequest<?> request) {
         return true;
     }
 
-    protected boolean supportStartOffsetTimeWindow() {
-        return false;
-    }
-
     public void init() {
-        executor.submit(() -> {
+        inputExecutor.submit(() -> {
             try {
-                if (supportStartOffsetTimeWindow()) {
+                if (supportStartOffsetTimeWindow) {
                     consumer.subscribe(callbacks.keySet(), new OffsetChanger(consumer, callbacks.keySet()));
                 } else {
                     consumer.subscribe(callbacks.keySet());
                 }
+                Object o = null;
                 while (!closed.get()) {
                     try {
                         ConsumerRecords<String, Object> records = consumer.poll(Duration.of(10, ChronoUnit.SECONDS));
@@ -62,13 +74,19 @@ public class QueueConsumer implements AutoCloseable {
                             ConsumerSpecificClass<?> callback = callbacks.get(next.topic());
                             Class<?> clazz = callback.getClazz();
                             JavaType payloadType = json.getTypeFactory().constructParametricType(BaseRequest.class, clazz);
-                            Object o = json.readValue((String) next.value(), payloadType);
+                            o = json.readValue((String) next.value(), payloadType);
                             if (needsProcessing(next.topic(), (BaseRequest<?>) o)) {
-                                callback.acceptRaw(o);
+                                Object topicResponse = callback.acceptRaw(o);
+                                if (producer != null) {
+                                    sendResponse((BaseRequest<?>) o, topicResponse);
+                                }
                             }
                         }
                     } catch (Throwable e) {
                         log.error(ExceptionUtils.getStackTrace(e));
+                        if (producer != null) {
+                            sendErrorResponse((BaseRequest<?>) o);
+                        }
                     }
                 }
             } catch (WakeupException e) {
@@ -81,6 +99,43 @@ public class QueueConsumer implements AutoCloseable {
         });
     }
 
+    //мб перенести в другой класс
+    private void sendErrorResponse(BaseRequest<?> o) {
+        outputExecutor.submit(() -> {
+            try {
+                Future<RecordMetadata> send;
+                //default response
+                RequestInfoUpdate success = new RequestInfoUpdate();
+                success.setId(o.getId());
+                success.setStatus(Status.Error);
+                send = producer.send(new ProducerRecord<>(Consts.REQUEST_INFO_UPDATE, success));
+                send.get();
+            } catch (Exception e) {
+                log.error(ExceptionUtils.getStackTrace(e));
+            }
+        });
+    }
+
+    private void sendResponse(BaseRequest<?> o, Object response) {
+        outputExecutor.submit(() -> {
+            try {
+                Future<RecordMetadata> send;
+                if (response != null) {
+                    send = producer.send(new ProducerRecord<>(Consts.REQUEST_INFO_UPDATE, response));
+                } else {
+                    //default response
+                    RequestInfoUpdate success = new RequestInfoUpdate();
+                    success.setId(o.getId());
+                    success.setStatus(Status.Success);
+                    send = producer.send(new ProducerRecord<>(Consts.REQUEST_INFO_UPDATE, success));
+                }
+                send.get();
+            } catch (Exception e) {
+                log.error(ExceptionUtils.getStackTrace(e));
+            }
+        });
+    }
+
     protected <T> BuilderConsumerStep<T> callback(Class<T> clazz) {
         return ConsumerSpecificClass.build(clazz);
     }

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/RequestInfoUpdate.java

@@ -0,0 +1,22 @@
+package ru.spcex.clearing.platform.messaging.service;
+
+public class RequestInfoUpdate {
+    private Long id;
+    private Status status;
+
+    public Long getId() {
+        return id;
+    }
+
+    public void setId(Long id) {
+        this.id = id;
+    }
+
+    public Status getStatus() {
+        return status;
+    }
+
+    public void setStatus(Status status) {
+        this.status = status;
+    }
+}