treshenkov/clearing
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

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

Browse source 
Добавил формирование сообщения при отправке запроса в ТС.
This commit is contained in:
psemenkov 2024-04-26 16:22:14 +03:00
parent d93bf629af
commit 21976322c2
5 changed files with 225 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -158,7 +158,7 @@ public class GatewayController {
@ResponseBody
public CommonResponse limits(@RequestBody LimitRequest request) {
log.debug("Received message: {}", request);
notificationService.sendLimitNotification(request);
executor.submit(() -> notificationService.sendLimitNotification(request));
return createCommonResponse(request.getId());
}
@ -175,7 +175,10 @@ public class GatewayController {
@ResponseBody
public CommonResponse operations(@RequestBody OperationsRequest request) {
log.debug("Received message: {}", request);
executor.submit(() -> operationService.checkAssetsAndSendMessage(request));
executor.submit(() -> {
operationService.checkAssetsAndSendMessage(request);
notificationService.updateGatewayResultInMap(request);
});
return createCommonResponse(request.getId());
}

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

@ -10,13 +10,16 @@ import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.http.*;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;
import ru.clearing.classes.statics.data.register.GatewayResult;
import ru.spcex.clearing.gatewayapi.config.GatewayApiSettings;
import ru.spcex.clearing.gatewayapi.config.InboundServerSettings;
import ru.spcex.clearing.gatewayapi.controller.inbound.request.operations.SentAsset;
import ru.spcex.clearing.gatewayapi.controller.outbound.request.OutboundRequest;
import ru.spcex.clearing.gatewayapi.controller.outbound.response.SuccessResponse;
import ru.spcex.clearing.gatewayapi.enums.OutboundRequestType;
import ru.spcex.clearing.gatewayapi.service.builders.GatewayResultBuilder;
import ru.spcex.clearing.gatewayapi.service.builders.OutboundRequestBuilder;
import ru.spcex.clearing.imdg.IMDGDistributedNames;
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;
@ -32,6 +35,8 @@ import ru.spcex.platform.enumeration.ReportKeys;
import ru.spcex.platform.enumeration.ReportType;
import ru.spcex.platform.enumeration.Section;
import ru.spcex.platform.enumeration.Task;
import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.ImdgProvider;
import ru.spcex.platform.utils.enumeration.IEnumKey;
import java.math.BigDecimal;
@ -47,6 +52,7 @@ public class GatewayService extends QueueConsumer implements InitializingBean {
private final InboundServerSettings inboundServerSettings;
private final Map<Long, List<SentAsset>> sentAssets;
private final Map<UUID, OutboundRequest> outboundRequestByUuid;
protected final Imdg<GatewayResult> gatewayResultImdg;
private final static Pattern REPORT_FILE_NAME_PATTERN = Pattern.compile("(\\S*)_(\\d{15}).csv");
public GatewayService(Consumer<String, Object> kafkaQueue,
@ -55,13 +61,15 @@ public class GatewayService extends QueueConsumer implements InitializingBean {
UserRoleVerification userRoleVerification,
GatewayApiSettings gatewayApiSettings,
@Qualifier("sentAssets") Map<Long, List<SentAsset>> sentAssets,
Map<UUID, OutboundRequest> outboundRequestByUuid) {
Map<UUID, OutboundRequest> outboundRequestByUuid,
ImdgProvider imdgProvider) {
super(kafkaQueue, kafkaProducer);
this.restTemplate = restTemplate;
this.userRoleVerification = userRoleVerification;
this.inboundServerSettings = gatewayApiSettings.getInboundServer();
this.sentAssets = sentAssets;
this.outboundRequestByUuid = outboundRequestByUuid;
this.gatewayResultImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_GatewayResult, GatewayResult.class);
}
@Override
@ -121,6 +129,7 @@ public class GatewayService extends QueueConsumer implements InitializingBean {
HttpEntity<OutboundRequest> request = makeDefaultRequest(outboundFondRequest);
outboundRequestByUuid.put(outboundFondRequest.getId(), outboundFondRequest);
gatewayResultImdg.insert(GatewayResultBuilder.builder().outboundRequest(outboundFondRequest).build());
ResponseEntity<SuccessResponse> response = restTemplate.exchange(url, HttpMethod.POST, request, SuccessResponse.class);
SuccessResponse bodyResponse = response.getBody();
if (bodyResponse != null) {
@ -181,7 +190,9 @@ public class GatewayService extends QueueConsumer implements InitializingBean {
Long requestId = request.getId();
Collection<AssetOperationRequest> assetOperations = request.getRequestPayload().getAssetOperationRequests();
List<SentAsset> assets = new ArrayList<>();
GatewayResultBuilder gatewayBuilder = GatewayResultBuilder.builder();
for (AssetOperationRequest assetOperation : assetOperations) {
gatewayBuilder.assetOperation(assetOperation);
Map<String, Object> content = new HashMap<>() {{
put("direction", assetOperation.getDirection());
put("code", assetOperation.getCode());
@ -199,10 +210,12 @@ public class GatewayService extends QueueConsumer implements InitializingBean {
.type(OutboundRequestType.ASSET_OPERATION.getKey())
.content(content).build();
gatewayBuilder.outboundRequest(outboundRequest);
String url = formingInboundUrl(inboundServerSettings.getPathLOCM());
HttpEntity<OutboundRequest> r = makeDefaultRequest(outboundRequest);
try {
outboundRequestByUuid.put(outboundRequest.getId(), outboundRequest);
gatewayResultImdg.insert(gatewayBuilder.build());
ResponseEntity<SuccessResponse> response = restTemplate.exchange(url, HttpMethod.POST, r, SuccessResponse.class);
SuccessResponse bodyResponse = response.getBody();
if (bodyResponse != null) {

79
clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/service/NotificationService.java
View file

@ -3,16 +3,24 @@ package ru.spcex.clearing.gatewayapi.service;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import ru.clearing.classes.statics.data.register.GatewayResult;
import ru.spcex.clearing.gatewayapi.controller.inbound.request.limit.LimitRequest;
import ru.spcex.clearing.gatewayapi.controller.inbound.request.operations.OperationsRequest;
import ru.spcex.clearing.gatewayapi.controller.outbound.request.OutboundRequest;
import ru.spcex.clearing.gatewayapi.controller.outbound.response.ErrorResponse;
import ru.spcex.clearing.gatewayapi.enums.OutboundRequestType;
import ru.spcex.clearing.imdg.IMDGDistributedNames;
import ru.spcex.clearing.platform.messaging.domain.Consts;
import ru.spcex.clearing.platform.messaging.domain.cud.utilities.NotificationNewRequest;
import ru.spcex.clearing.platform.messaging.serialization.LogFormatter;
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
import ru.spcex.platform.enumeration.ObjectType;
import ru.spcex.platform.enumeration.Priority;
import ru.spcex.platform.enumeration.ResultStatuses;
import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.ImdgProvider;
import java.time.Instant;
import java.util.List;
import java.util.Map;
import java.util.UUID;
@ -23,12 +31,15 @@ public class NotificationService {
private final KafkaSender kafkaSender;
private final Map<UUID, OutboundRequest> outboundRequestByUuid;
private final List<String> supportedTypes = List.of(OutboundRequestType.FILL_LIMITS.getKey());
protected final Imdg<GatewayResult> gatewayResultImdg;
private final List<String> supportedTypes = List.of(OutboundRequestType.FILL_LIMITS.getKey(), OutboundRequestType.ASSET_OPERATION.getKey());
public NotificationService(KafkaSender kafkaSender,
Map<UUID, OutboundRequest> outboundRequestByUuid) {
Map<UUID, OutboundRequest> outboundRequestByUuid,
ImdgProvider imdgProvider) {
this.kafkaSender = kafkaSender;
this.outboundRequestByUuid = outboundRequestByUuid;
this.gatewayResultImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_GatewayResult, GatewayResult.class);
}
public void sendLimitNotification(LimitRequest request) {
@ -40,6 +51,9 @@ public class NotificationService {
}
log.info("Receive request: uuid = {}, file = {} with result = {} and sending section = {}.",
request.getId(), request.getLogFile(), request.getResult(), outboundReq.getSection());
updateGatewayResultInMap(request);
newRequest.setObjectType(ObjectType.rgst.getKey());
newRequest.setComment(request.getDescription());
if (request.getResult() != null && request.getResult()) {
@ -53,14 +67,69 @@ public class NotificationService {
kafkaSender.sendRequestToQueue(Consts.NOTIFICATION_NEW, newRequest);
}
public void updateGatewayResultInMap(LimitRequest request) {
if (request.getId() == null || request.getResult() == null) {
log.warn("GatewayResult not found because LimitRequest not valid: {}", LogFormatter.toString(request));
return;
}
GatewayResult gatewayResult = gatewayResultImdg.getFirstObjectByFieldValues(Map.of("requestId", request.getId().toString()));
if (gatewayResult == null) {
log.warn("GatewayResult by requestId = {} not found in map", request.getId().toString());
return;
}
gatewayResult.setUpdated(Instant.now());
gatewayResult.setGatewayResultStatus(request.getResult() ? ResultStatuses.success.getKey() : ResultStatuses.notSuccess.getKey());
gatewayResult.setResult(LogFormatter.toString(request));
gatewayResultImdg.update(gatewayResult);
log.info("Updated gatewayResult: {} in map", LogFormatter.toString(gatewayResult));
}
public void updateGatewayResultInMap(OperationsRequest request) {
if (request.getId() == null || request.getResult() == null) {
log.warn("GatewayResult not found because OperationsRequest not valid: {}", LogFormatter.toString(request));
return;
}
GatewayResult gatewayResult = gatewayResultImdg.getFirstObjectByFieldValues(Map.of("requestId", request.getId().toString()));
if (gatewayResult == null) {
log.warn("GatewayResult by requestId = {} not found in map", request.getId().toString());
return;
}
gatewayResult.setUpdated(Instant.now());
gatewayResult.setGatewayResultStatus(request.getResult() ? ResultStatuses.success.getKey() : ResultStatuses.notSuccess.getKey());
gatewayResult.setResult(LogFormatter.toString(request));
gatewayResultImdg.update(gatewayResult);
log.info("Updated gatewayResult: {} in map", LogFormatter.toString(gatewayResult));
}
public void updateGatewayResultInMap(ErrorResponse response) {
if (response.getId() == null) {
log.warn("GatewayResult not found because OperationsRequest not valid: {}", LogFormatter.toString(response));
return;
}
GatewayResult gatewayResult = gatewayResultImdg.getFirstObjectByFieldValues(Map.of("requestId", response.getId().toString()));
if (gatewayResult == null) {
log.warn("GatewayResult by requestId = {} not found in map", response.getId().toString());
return;
}
gatewayResult.setUpdated(Instant.now());
gatewayResult.setGatewayResultStatus(ResultStatuses.notSuccess.getKey());
gatewayResult.setResult(LogFormatter.toString(response));
gatewayResultImdg.update(gatewayResult);
log.info("Updated gatewayResult: {} in map", LogFormatter.toString(gatewayResult));
}
public void sendErrorNotification(ErrorResponse response, OutboundRequest request) {
NotificationNewRequest newRequest = new NotificationNewRequest();
log.info("Receive ErrorResponse: uuid = {}, file = {} and sending section = {}.",
response.getId(), request.getContent().get("file"), request.getSection());
updateGatewayResultInMap(response);
if (OutboundRequestType.FILL_LIMITS.equalsByKey(request.getType()) && request.getContent() != null && request.getContent().size() > 0)
log.info("Receive ErrorResponse: uuid = {}, file = {}, sending type = {} and section = {}.",
response.getId(), request.getContent().get("file"), request.getType(), request.getSection());
else
log.info("Receive ErrorResponse: uuid = {}, sending type = {} and section = {}.", response.getId(), request.getType(), request.getSection());
newRequest.setObjectType(ObjectType.rgst.getKey());
newRequest.setComment(response.getMessage());
newRequest.setPriority(Priority.HIGH.getKey());
log.info("Sending notification to kafka: {}", newRequest);
log.info("Sending notification to kafka: {}", LogFormatter.toString(newRequest));
kafkaSender.sendRequestToQueue(Consts.NOTIFICATION_NEW, newRequest);
}

95
clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/service/builders/GatewayResultBuilder.java Normal file
View file

@ -0,0 +1,95 @@
package ru.spcex.clearing.gatewayapi.service.builders;
import ru.clearing.classes.statics.data.register.GatewayResult;
import ru.spcex.clearing.gatewayapi.controller.outbound.request.OutboundRequest;
import ru.spcex.clearing.gatewayapi.enums.OutboundRequestType;
import ru.spcex.clearing.platform.messaging.domain.cud.clearing.AssetOperationRequest;
import ru.spcex.clearing.platform.messaging.serialization.LogFormatter;
import ru.spcex.platform.enumeration.InOutDirection;
import ru.spcex.platform.enumeration.Section;
import java.time.Instant;
import static ru.spcex.platform.utils.enumeration.IEnumKey.getEnumByKey;
public class GatewayResultBuilder {
private String requestId;
private String nameRequest;
private String request;
private InOutDirection direction;
private GatewayResultBuilder() {
}
public static GatewayResultBuilder builder() {
return new GatewayResultBuilder();
}
public GatewayResultBuilder assetOperation(AssetOperationRequest assetOperation) {
this.direction = getEnumByKey(InOutDirection.class, assetOperation.getDirection());
return this;
}
public GatewayResultBuilder outboundRequest(OutboundRequest outboundRequest) {
this.request = LogFormatter.toString(outboundRequest);
this.requestId = outboundRequest.getId().toString();
Section section = getEnumByKey(Section.class, outboundRequest.getSection());
if (direction != null && OutboundRequestType.ASSET_OPERATION.equalsByKey(outboundRequest.getType()))
this.nameRequest = getNameReqByAsset(section, direction);
else if (OutboundRequestType.FILL_LIMITS.equalsByKey(outboundRequest.getType()))
this.nameRequest = getNameReqByLim(section);
return this;
}
public GatewayResult build() {
GatewayResult outboundRequest = new GatewayResult();
Instant now = Instant.now();
outboundRequest.setCreated(now);
outboundRequest.setUpdated(now);
outboundRequest.setRequestId(requestId);
outboundRequest.setNameRequest(nameRequest);
outboundRequest.setRequest(request);
return outboundRequest;
}
private String getNameReqByAsset(Section section, InOutDirection direction) {
String nameRequest;
if (InOutDirection.in.equals(direction)) {
if (Section.MKR.equals(section)) {
nameRequest = "Зачисление денежных средств";
} else if (Section.FOND.equals(section)) {
nameRequest = "Зачисление ценных бумаг";
} else if (Section.CURR.equals(section)) {
nameRequest = "Зачисление валютных средств";
} else {
nameRequest = "Зачисление " + section.getKey();
}
} else {
if (Section.MKR.equals(section)) {
nameRequest = "Списание денежных средств";
} else if (Section.FOND.equals(section)) {
nameRequest = "Списание ценных бумаг";
} else if (Section.CURR.equals(section)) {
nameRequest = "Списание валютных средств";
} else {
nameRequest = "Списание " + section.getKey();
}
}
return nameRequest;
}
private String getNameReqByLim(Section section) {
String nameRequest;
if (Section.MKR.equals(section)) {
nameRequest = "Лимиты по денежным средствам";
} else if (Section.FOND.equals(section)) {
nameRequest = "Лимиты по ценным бумагам";
} else if (Section.CURR.equals(section)) {
nameRequest = "Лимиты по валютным средствам";
} else {
nameRequest = "Лимиты по " + section.getKey();
}
return nameRequest;
}
}

38
clearing-parent/gateway-api/src/test/java/ru/spcex/clearing/gatewayapi/service/GatewayServiceTest.java
View file

@ -32,18 +32,23 @@ import ru.spcex.clearing.gatewayapi.controller.outbound.request.OutboundRequest;
import ru.spcex.clearing.gatewayapi.controller.outbound.response.ErrorResponse;
import ru.spcex.clearing.gatewayapi.controller.outbound.response.SuccessResponse;
import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
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.utilities.LimExportedRequest;
import ru.spcex.clearing.test.config.ImdgTestConfig;
import ru.spcex.clearing.test.config.KafkaTestConfig;
import ru.spcex.platform.enumeration.InOutDirection;
import ru.spcex.platform.enumeration.Section;
import ru.spcex.platform.imdg.api.ImdgProvider;
import javax.annotation.PostConstruct;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.math.BigDecimal;
import java.nio.charset.Charset;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicReference;
@ -61,6 +66,7 @@ import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.
public class GatewayServiceTest {
private static final CharacterEncodingFilter CHARACTER_ENCODING_FILTER = new CharacterEncodingFilter();
public static final String LIMITS_URL = "/limits/";
public static final String OPERATIONS_URL = "/operations/";
private MockRestServiceServer mockServer;
@Autowired
@ -137,6 +143,36 @@ public class GatewayServiceTest {
service.requestOnLimit(req);
}
@Disabled//Для отладки
@Test
void requestOnAsset() throws Exception {
String message = "test response";
BaseRequest<AssetOperationListRequest> req = new BaseRequest<>();
AssetOperationListRequest reqP = new AssetOperationListRequest();
AssetOperationRequest reqI = new AssetOperationRequest();
req.setId(1L);
reqP.setAssetOperationRequests(List.of(reqI));
req.setRequestPayload(reqP);
reqI.setAmount(BigDecimal.ONE);
reqI.setDirection(InOutDirection.in.getKey());
AtomicReference<UUID> uuid = prepareMockServerResponse(HttpStatus.OK, message);
service.requestOnAsset(req);
LimitRequest request = new LimitRequest();
request.setId(uuid.get());
request.setResult(true);
request.setDescription(message);
MvcResult mvcResult = perform(MockMvcRequestBuilders.post(OPERATIONS_URL)
.contentType(MediaType.APPLICATION_JSON)
.content(mapper.writeValueAsString(request)))
.andDo(print())//output to the log request and response
.andExpect(status().isOk())
.andExpect(content().contentTypeCompatibleWith(MediaType.APPLICATION_JSON))
.andReturn();
uuid = prepareMockServerResponse(HttpStatus.BAD_REQUEST, message);
service.requestOnAsset(req);
}
public AtomicReference<UUID> prepareMockServerResponse(HttpStatus responseStatus, String message) {
mockServer.reset();
@ -145,7 +181,7 @@ public class GatewayServiceTest {
ResponseCreator responseCreator = r -> {
String reqBody = StreamUtils.copyToString((ByteArrayOutputStream) r.getBody(), Charset.defaultCharset());
OutboundRequest request = mapper.readValue(reqBody, OutboundRequest.class);
if (HttpStatus.BAD_REQUEST.equals(responseStatus)){
if (HttpStatus.BAD_REQUEST.equals(responseStatus)) {
ErrorResponse errorResponse = new ErrorResponse();
errorResponse.setId(request.getId());
errorResponse.setMessage(message);