вводы/выводы - gateway

clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/config/BeanConfiguration.java

@@ -4,6 +4,7 @@ import org.springframework.beans.factory.config.ConfigurableBeanFactory;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Scope;
+import ru.spcex.clearing.gatewayapi.controller.request.operations.SentAsset;
 import ru.spcex.clearing.gatewayapi.errors.GatewayError;
 import ru.spcex.clearing.util.security.UserRoleVerification;
 import ru.spcex.clearing.util.services.IMDGMessageResolver;
@@ -11,6 +12,9 @@ import ru.spcex.platform.enumeration.UserRole;
 import ru.spcex.platform.imdg.api.ImdgProvider;
 import ru.spcex.platform.utils.enumeration.IMessageResolver;
 
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
@@ -32,4 +36,9 @@ public class BeanConfiguration {
     public ExecutorService executor() {
         return Executors.newFixedThreadPool(10);
     }
+
+    @Bean("sentAssets")
+    public Map<Long, List<SentAsset>> sentAssets() {
+        return new ConcurrentHashMap<>();
+    }
 }

clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/controller/GatewayController.java

@@ -15,8 +15,10 @@ import ru.spcex.clearing.gatewayapi.controller.request.company.CompaniesRequest;
 import ru.spcex.clearing.gatewayapi.controller.request.limit.LimitRequest;
 import ru.spcex.clearing.gatewayapi.controller.request.listing.fond.FondListingsRequest;
 import ru.spcex.clearing.gatewayapi.controller.request.listing.mkr.MMListingsRequest;
+import ru.spcex.clearing.gatewayapi.controller.request.operations.OperationsRequest;
 import ru.spcex.clearing.gatewayapi.controller.response.CommonResponse;
 import ru.spcex.clearing.gatewayapi.exception.GatewayException;
+import ru.spcex.clearing.gatewayapi.service.OperationService;
 import ru.spcex.clearing.gatewayapi.service.processor.CompanyProcessor;
 import ru.spcex.clearing.gatewayapi.service.processor.ListingFondProcessor;
 import ru.spcex.clearing.gatewayapi.service.processor.ListingMMProcessor;
@@ -39,6 +41,7 @@ public class GatewayController {
     private final ListingFondProcessor listingFondProcessor;
     private final ListingMMProcessor listingMMProcessor;
     private final ExecutorService executor;
+    private final OperationService operationService;
 
     private List<String> validTypes = Arrays.asList("DAY_START", "ON_DEMAND");
 
@@ -46,12 +49,14 @@ public class GatewayController {
                              CompanyProcessor companiesProcessor,
                              ListingFondProcessor listingFondProcessor,
                              ListingMMProcessor listingMMProcessor,
-                             @Qualifier("gatewayExecutor") ExecutorService executor) {
+                             @Qualifier("gatewayExecutor") ExecutorService executor,
+                             OperationService operationService) {
         this.messageResolver = messageResolver;
         this.companiesProcessor = companiesProcessor;
         this.listingFondProcessor = listingFondProcessor;
         this.listingMMProcessor = listingMMProcessor;
         this.executor = executor;
+        this.operationService = operationService;
     }
 
 
@@ -125,7 +130,24 @@ public class GatewayController {
     @ResponseBody
     public CommonResponse limits(@RequestBody LimitRequest request) {
         log.debug("Received message: {}", request);
-        return createResponseByLimit(request.getId());
+        return createCommonResponse(request.getId());
+    }
+
+    @ApiResponses(value = {
+            @ApiResponse(code = 200, message = "OK", response = CommonResponse.class),
+            @ApiResponse(code = 400, message = "Ошибка валидации", response = CommonResponse.class)
+    })
+    @RequestMapping(
+            path = "/operations",
+            method = RequestMethod.POST,
+            consumes = MediaType.APPLICATION_JSON_VALUE,
+            produces = MediaType.APPLICATION_JSON_VALUE
+    )
+    @ResponseBody
+    public CommonResponse operations(@RequestBody OperationsRequest request) {
+        log.debug("Received message: {}", request);
+        executor.submit(() -> operationService.checkAssetsAndSendMessage(request));
+        return createCommonResponse(request.getId());
     }
 
     @ResponseStatus(value = HttpStatus.BAD_REQUEST)
@@ -151,7 +173,7 @@ public class GatewayController {
         return commonResponse;
     }
 
-    private CommonResponse createResponseByLimit(UUID uuid) {
+    private CommonResponse createCommonResponse(UUID uuid) {
         CommonResponse commonResponse = new CommonResponse();
         commonResponse.setId(uuid);
 //        commonResponse.setType(request.getType());

clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/controller/request/operations/OperationsRequest.java

@@ -0,0 +1,38 @@
+package ru.spcex.clearing.gatewayapi.controller.request.operations;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.UUID;
+
+public class OperationsRequest {
+    @JsonProperty("id")
+    private UUID id;
+    @JsonProperty("result")
+    private Boolean result;
+    @JsonProperty("description")
+    private String description;
+
+    public UUID getId() {
+        return id;
+    }
+
+    public void setId(UUID id) {
+        this.id = id;
+    }
+
+    public Boolean getResult() {
+        return result;
+    }
+
+    public void setResult(Boolean result) {
+        this.result = result;
+    }
+
+    public String getDescription() {
+        return description;
+    }
+
+    public void setDescription(String description) {
+        this.description = description;
+    }
+}

clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/controller/request/operations/SentAsset.java

@@ -0,0 +1,42 @@
+package ru.spcex.clearing.gatewayapi.controller.request.operations;
+
+import java.util.UUID;
+
+public class SentAsset {
+    private boolean isReceivedResponse = false;
+    private UUID id;
+    private Long statementId;
+    private OperationsRequest operationsRequest;
+
+    public boolean isReceivedResponse() {
+        return isReceivedResponse;
+    }
+
+    public void setReceivedResponse(boolean receivedResponse) {
+        isReceivedResponse = receivedResponse;
+    }
+
+    public UUID getId() {
+        return id;
+    }
+
+    public void setId(UUID id) {
+        this.id = id;
+    }
+
+    public OperationsRequest getOperationsRequest() {
+        return operationsRequest;
+    }
+
+    public void setOperationsRequest(OperationsRequest operationsRequest) {
+        this.operationsRequest = operationsRequest;
+    }
+
+    public Long getStatementId() {
+        return statementId;
+    }
+
+    public void setStatementId(Long statementId) {
+        this.statementId = statementId;
+    }
+}

clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/service/GatewayService.java

@@ -1,10 +1,12 @@
 package ru.spcex.clearing.gatewayapi.service;
 
+import org.apache.commons.lang3.exception.ExceptionUtils;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.producer.Producer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.InitializingBean;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.http.*;
 import org.springframework.stereotype.Service;
 import org.springframework.web.client.HttpClientErrorException;
@@ -12,15 +14,14 @@ import org.springframework.web.client.RestTemplate;
 import ru.spcex.clearing.gatewayapi.config.GatewayApiSettings;
 import ru.spcex.clearing.gatewayapi.config.InboundServerSettings;
 import ru.spcex.clearing.gatewayapi.controller.request.OutboundRequest;
+import ru.spcex.clearing.gatewayapi.controller.request.operations.SentAsset;
 import ru.spcex.clearing.gatewayapi.enums.OutboundRequestType;
 import ru.spcex.clearing.gatewayapi.service.builders.OutboundRequestBuilder;
 import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
 import ru.spcex.clearing.platform.messaging.domain.Consts;
 import ru.spcex.clearing.platform.messaging.domain.cud.clearing.AssetOperationListRequest;
 import ru.spcex.clearing.platform.messaging.domain.cud.clearing.AssetOperationRequest;
-import ru.spcex.clearing.platform.messaging.domain.cud.gateway.AssetOperationApprovalRequest;
 import ru.spcex.clearing.platform.messaging.domain.cud.gateway.GatewayTaskRequest;
-import ru.spcex.clearing.platform.messaging.domain.cud.gateway.SingleAssetResponse;
 import ru.spcex.clearing.platform.messaging.domain.cud.utilities.LimExportedRequest;
 import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
 import ru.spcex.clearing.platform.messaging.service.RequestInfoUpdate;
@@ -38,19 +39,21 @@ public class GatewayService extends QueueConsumer implements InitializingBean {
     private final UserRoleVerification userRoleVerification;
     private final RestTemplate restTemplate;
     private final InboundServerSettings inboundServerSettings;
-    private final KafkaSender kafkaSender;
+
+    private final Map<Long, List<SentAsset>> sentAssets;
 
     public GatewayService(Consumer<String, Object> kafkaQueue,
                           Producer<String, Object> kafkaProducer,
                           RestTemplate restTemplate,
                           UserRoleVerification userRoleVerification,
                           GatewayApiSettings gatewayApiSettings,
-                          KafkaSender kafkaSender) {
+                          KafkaSender kafkaSender,
+                          @Qualifier("sentAssets") Map<Long, List<SentAsset>> sentAssets) {
         super(kafkaQueue, kafkaProducer);
         this.restTemplate = restTemplate;
         this.userRoleVerification = userRoleVerification;
         this.inboundServerSettings = gatewayApiSettings.getInboundServer();
-        this.kafkaSender = kafkaSender;
+        this.sentAssets = sentAssets;
     }
 
     @Override
@@ -145,9 +148,9 @@ public class GatewayService extends QueueConsumer implements InitializingBean {
     }
 
     public void requestOnAsset(BaseRequest<AssetOperationListRequest> request) {
+        Long requestId = request.getId();
         Collection<AssetOperationRequest> assetOperations = request.getRequestPayload().getAssetOperationRequests();
-        List<SingleAssetResponse> responses = new ArrayList<>();
+        List<SentAsset> assets = new ArrayList<>();
-        AssetOperationApprovalRequest assetOperationApprovalRequest = new AssetOperationApprovalRequest();
         for (AssetOperationRequest assetOperation : assetOperations) {
             Map<String, Object> content = Map.of("direction", assetOperation.getDirection(),
                     "amount", assetOperation.getAmount(),
@@ -166,19 +169,21 @@ public class GatewayService extends QueueConsumer implements InitializingBean {
                     .content(content).build();
 
             String url = formingInboundUrl(inboundServerSettings.getPathLOCM());
-
             HttpEntity<OutboundRequest> r = makeDefaultRequest(outboundRequest);
-            ResponseEntity<Map> response = restTemplate.exchange(url, HttpMethod.POST, r, Map.class);
+            try {
-            Map bodyResponse = response.getBody();
+                ResponseEntity<Map> response = restTemplate.exchange(url, HttpMethod.POST, r, Map.class);
-            SingleAssetResponse singleAssetResponse = new SingleAssetResponse();
+                Map bodyResponse = response.getBody();
-            singleAssetResponse.setApproved(Boolean.parseBoolean((String) bodyResponse.get("result")));
+                log.debug("Response from service: {}", bodyResponse != null ? bodyResponse : "empty");
-            singleAssetResponse.setStatementId(assetOperation.getStatementId());
+            } catch (Throwable err){
-            responses.add(singleAssetResponse);
+                log.warn("Exchange error: {}", ExceptionUtils.getStackTrace(err));
-            log.debug("Response from service: {}", bodyResponse);
+                SentAsset sentAsset = new SentAsset();
+                sentAsset.setId(outboundRequest.getId());
+                sentAsset.setStatementId(assetOperation.getStatementId());
+                assets.add(sentAsset);
+            }
         }
 
-        assetOperationApprovalRequest.setApprovals(responses);
+        sentAssets.put(requestId, assets);
-        kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION_APPROVAL, responses);
     }
 
     private String formingInboundUrl(String path) {

clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/service/OperationService.java

@@ -0,0 +1,61 @@
+package ru.spcex.clearing.gatewayapi.service;
+
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.stereotype.Service;
+import ru.spcex.clearing.gatewayapi.controller.request.operations.OperationsRequest;
+import ru.spcex.clearing.gatewayapi.controller.request.operations.SentAsset;
+import ru.spcex.clearing.platform.messaging.domain.Consts;
+import ru.spcex.clearing.platform.messaging.domain.cud.gateway.AssetOperationApprovalRequest;
+import ru.spcex.clearing.platform.messaging.domain.cud.gateway.SingleAssetResponse;
+import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+@Service
+public class OperationService {
+    private final Map<Long, List<SentAsset>> sentAssets;
+    private final KafkaSender kafkaSender;
+
+    public OperationService(@Qualifier("sentAssets") Map<Long, List<SentAsset>> sentAssets,
+                            KafkaSender kafkaSender) {
+        this.sentAssets = sentAssets;
+        this.kafkaSender = kafkaSender;
+    }
+
+    public void checkAssetsAndSendMessage(OperationsRequest operationsRequest) {
+        Long requestId = null;
+        for (Map.Entry<Long, List<SentAsset>> entry : sentAssets.entrySet()) {
+            List<SentAsset> value = entry.getValue();
+            Optional<SentAsset> assetById = value.stream()
+                    .filter(sentAsset -> sentAsset.getId().equals(operationsRequest.getId()))
+                    .findFirst();
+            if (assetById.isPresent()) {
+                requestId = entry.getKey();
+                SentAsset sentAsset = assetById.get();
+                sentAsset.setReceivedResponse(true);
+                sentAsset.setOperationsRequest(operationsRequest);
+                break;
+            }
+        }
+        if (requestId != null) {
+            List<SentAsset> assets = sentAssets.get(requestId);
+            boolean isAllReceivedInBatch = assets.stream().allMatch(SentAsset::isReceivedResponse);
+            if (isAllReceivedInBatch) {
+                List<SingleAssetResponse> responsesToClearing = new ArrayList<>();
+                for (SentAsset sentAsset : assets) {
+                    SingleAssetResponse singleAssetResponse = new SingleAssetResponse();
+                    singleAssetResponse.setApproved(sentAsset.getOperationsRequest().getResult());
+                    singleAssetResponse.setStatementId(sentAsset.getStatementId());
+                    responsesToClearing.add(singleAssetResponse);
+                }
+                AssetOperationApprovalRequest assetOperationApprovalRequest = new AssetOperationApprovalRequest();
+                assetOperationApprovalRequest.setApprovals(responsesToClearing);
+                kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION_APPROVAL, assetOperationApprovalRequest);
+                sentAssets.remove(requestId);
+            }
+        }
+    }
+}