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

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

Browse source 
Добавил отправку сообщений о выгрузке лимитов в торговую систему для utility-service через kafka.
This commit is contained in:
psemenkov 2024-04-24 11:22:00 +03:00
parent 97bdf20975
commit 8315e9814b
13 changed files with 287 additions and 51 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
clearing-parent/gateway-api/pom.xml
View file

@ -72,17 +72,7 @@
<!-- TEST -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-test-autoconfigure</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>

18
clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/config/RestTemplateErrorHandler.java
View file

@ -4,6 +4,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpStatus;
import org.springframework.http.client.ClientHttpResponse;
import org.springframework.stereotype.Component;
import org.springframework.util.StreamUtils;
@ -11,6 +12,7 @@ import org.springframework.web.client.ResponseErrorHandler;
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.gatewayapi.service.NotificationService;
import ru.spcex.clearing.platform.messaging.domain.Consts;
import ru.spcex.clearing.platform.messaging.domain.cud.utilities.NotificationNewRequest;
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
@ -30,13 +32,16 @@ public class RestTemplateErrorHandler implements ResponseErrorHandler {
private final KafkaSender kafaSender;
private final ObjectMapper objectMapper;
private final Map<UUID, OutboundRequest> outboundRequestByUuid;
private final NotificationService notificationService;
public RestTemplateErrorHandler(KafkaSender kafaSender,
ObjectMapper objectMapper,
Map<UUID, OutboundRequest> outboundRequestByUuid) {
Map<UUID, OutboundRequest> outboundRequestByUuid,
NotificationService notificationService) {
this.kafaSender = kafaSender;
this.objectMapper = objectMapper;
this.outboundRequestByUuid = outboundRequestByUuid;
this.notificationService = notificationService;
}
@Override
@ -51,11 +56,12 @@ public class RestTemplateErrorHandler implements ResponseErrorHandler {
log.info("Response error body: " + responseErrorBody);
ErrorResponse errorResponse = objectMapper.readValue(responseErrorBody, ErrorResponse.class);
String errorMessage;
OutboundRequest outboundRequest = null;
if (StringUtils.isEmpty(errorResponse.getMessage())) {
String error = (String) errorResponse.getUnknownProperties().get("error");
errorMessage = String.format("Error: %s", error);
} else {
OutboundRequest outboundRequest = outboundRequestByUuid.remove(errorResponse.getId());
outboundRequest = outboundRequestByUuid.remove(errorResponse.getId());
if (outboundRequest != null) {
errorMessage = makeMessage(outboundRequest, errorResponse);
} else {
@ -63,8 +69,12 @@ public class RestTemplateErrorHandler implements ResponseErrorHandler {
errorMessage = errorResponse.getMessage();
}
}
NotificationNewRequest notificationNewRequest = createNotification(errorMessage);
kafaSender.sendRequestToQueue(Consts.NOTIFICATION_NEW, notificationNewRequest);
if (HttpStatus.BAD_REQUEST.equals(response.getStatusCode()) && outboundRequest != null && notificationService.canProcess(outboundRequest.getType())) {
notificationService.sendErrorNotification(errorResponse, outboundRequest);
} else {
NotificationNewRequest notificationNewRequest = createNotification(errorMessage);
kafaSender.sendRequestToQueue(Consts.NOTIFICATION_NEW, notificationNewRequest);
}
}
private NotificationNewRequest createNotification(String message) {

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

@ -19,6 +19,7 @@ import ru.spcex.clearing.gatewayapi.controller.inbound.request.listing.mkr.MMLis
import ru.spcex.clearing.gatewayapi.controller.inbound.request.operations.OperationsRequest;
import ru.spcex.clearing.gatewayapi.controller.inbound.response.CommonResponse;
import ru.spcex.clearing.gatewayapi.exception.GatewayException;
import ru.spcex.clearing.gatewayapi.service.NotificationService;
import ru.spcex.clearing.gatewayapi.service.OperationService;
import ru.spcex.clearing.gatewayapi.service.processor.CompanyProcessor;
import ru.spcex.clearing.gatewayapi.service.processor.ListingCurrProcessor;
@ -45,6 +46,7 @@ public class GatewayController {
private final ListingCurrProcessor listingCurrProcessor;
private final ExecutorService executor;
private final OperationService operationService;
private final NotificationService notificationService;
private List<String> validTypes = Arrays.asList("DAY_START", "ON_DEMAND");
@ -54,7 +56,8 @@ public class GatewayController {
ListingMMProcessor listingMMProcessor,
ListingCurrProcessor listingCurrProcessor,
@Qualifier("gatewayExecutor") ExecutorService executor,
OperationService operationService) {
OperationService operationService,
NotificationService notificationService) {
this.messageResolver = messageResolver;
this.companiesProcessor = companiesProcessor;
this.listingFondProcessor = listingFondProcessor;
@ -62,6 +65,7 @@ public class GatewayController {
this.listingCurrProcessor = listingCurrProcessor;
this.executor = executor;
this.operationService = operationService;
this.notificationService = notificationService;
}
@ -154,6 +158,7 @@ public class GatewayController {
@ResponseBody
public CommonResponse limits(@RequestBody LimitRequest request) {
log.debug("Received message: {}", request);
notificationService.sendLimitNotification(request);
return createCommonResponse(request.getId());
}

6
clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/controller/inbound/request/limit/LimitRequest.java
View file

@ -6,7 +6,7 @@ import java.util.UUID;
public class LimitRequest {
@JsonProperty("result")
private String result;
private Boolean result;
@JsonProperty("description")
private String description;
@JsonProperty("id")
@ -16,11 +16,11 @@ public class LimitRequest {
@JsonProperty("error_file")
private String errorFile;
public String getResult() {
public Boolean getResult() {
return result;
}
public void setResult(String result) {
public void setResult(Boolean result) {
this.result = result;
}

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

@ -51,7 +51,7 @@ public class GatewayService extends QueueConsumer implements InitializingBean {
public GatewayService(Consumer<String, Object> kafkaQueue,
Producer<String, Object> kafkaProducer,
RestTemplate restTemplate,
@Qualifier("rest-template") RestTemplate restTemplate,
UserRoleVerification userRoleVerification,
GatewayApiSettings gatewayApiSettings,
@Qualifier("sentAssets") Map<Long, List<SentAsset>> sentAssets,
@ -274,7 +274,7 @@ public class GatewayService extends QueueConsumer implements InitializingBean {
}
}
private String formingInboundUrl(String path) {
public String formingInboundUrl(String path) {
return "%s://%s:%s/%s".formatted(
inboundServerSettings.getEnableSsl() ? "https" : "http",
inboundServerSettings.getHost(),

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

@ -0,0 +1,70 @@
package ru.spcex.clearing.gatewayapi.service;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import ru.spcex.clearing.gatewayapi.controller.inbound.request.limit.LimitRequest;
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.platform.messaging.domain.Consts;
import ru.spcex.clearing.platform.messaging.domain.cud.utilities.NotificationNewRequest;
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
import ru.spcex.platform.enumeration.ObjectType;
import ru.spcex.platform.enumeration.Priority;
import java.util.List;
import java.util.Map;
import java.util.UUID;
@Service
public class NotificationService {
private final Logger log = LoggerFactory.getLogger(getClass());
private final KafkaSender kafkaSender;
private final Map<UUID, OutboundRequest> outboundRequestByUuid;
private final List<String> supportedTypes = List.of(OutboundRequestType.FILL_LIMITS.getKey());
public NotificationService(KafkaSender kafkaSender,
Map<UUID, OutboundRequest> outboundRequestByUuid) {
this.kafkaSender = kafkaSender;
this.outboundRequestByUuid = outboundRequestByUuid;
}
public void sendLimitNotification(LimitRequest request) {
NotificationNewRequest newRequest = new NotificationNewRequest();
OutboundRequest outboundReq = outboundRequestByUuid.get(request.getId());
if (outboundReq == null) {
log.error("Unknown uuid = {} in LimitRequest", request.getId());
return;
}
log.info("Receive request: uuid = {}, file = {} with result = {} and sending section = {}.",
request.getId(), request.getLogFile(), request.getResult(), outboundReq.getSection());
newRequest.setObjectType(ObjectType.rgst.getKey());
newRequest.setComment(request.getDescription());
if (request.getResult() != null && request.getResult()) {
newRequest.setPriority(Priority.LOW.getKey());
} else if (request.getResult() != null && !request.getResult()) {
newRequest.setPriority(Priority.HIGH.getKey());
} else {
log.error("Unknown status in LimitRequest result = {}, uuid = {}", request.getResult(), request.getId());
}
log.info("Sending notification to kafka: {}", newRequest);
kafkaSender.sendRequestToQueue(Consts.NOTIFICATION_NEW, newRequest);
}
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());
newRequest.setObjectType(ObjectType.rgst.getKey());
newRequest.setComment(response.getMessage());
newRequest.setPriority(Priority.HIGH.getKey());
log.info("Sending notification to kafka: {}", newRequest);
kafkaSender.sendRequestToQueue(Consts.NOTIFICATION_NEW, newRequest);
}
public boolean canProcess(String type) {
return supportedTypes.contains(type);
}
}

4
clearing-parent/gateway-api/src/main/resources/application.properties
View file

@ -28,5 +28,5 @@ gateway-api.kafka-consumer.buffer-memory=33554432
gateway-api.inbound-server.enable-ssl=false
gateway-api.inbound-server.host=10.200.200.183
gateway-api.inbound-server.port=8084
gateway-api.inbound-server.pathLOCM=/test_rest/inbound_request
gateway-api.inbound-server.pathLOSC=/test_rest/inbound_request
gateway-api.inbound-server.pathLOCM=test_rest/inbound_request
gateway-api.inbound-server.pathLOSC=test_rest/inbound_request

6
clearing-parent/gateway-api/src/test/java/ru/spcex/clearing/gatewayapi/config/TestConfiguration.java → clearing-parent/gateway-api/src/test/java/ru/spcex/clearing/gatewayapi/config/MapperConf.java
View file

@ -6,12 +6,12 @@ import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import org.springframework.boot.test.context.TestConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import ru.spcex.clearing.test.json.JsonUtil;
@Configuration
public class TestConfiguration {
@TestConfiguration
public class MapperConf {
@Bean("testObjectMapper")
public ObjectMapper objectMapper() {

15
clearing-parent/gateway-api/src/test/java/ru/spcex/clearing/gatewayapi/config/RestTemplateTestConfig.java
View file

@ -1,15 +0,0 @@
package ru.spcex.clearing.gatewayapi.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.client.RestTemplate;
@Configuration
public class RestTemplateTestConfig {
@Bean("rest-template")
public RestTemplate restTemplate(RestTemplateErrorHandler restTemplateErrorHandler) {
RestTemplate restTemplate = new RestTemplate();
return restTemplate;
}
}

20
clearing-parent/gateway-api/src/test/java/ru/spcex/clearing/gatewayapi/config/TestConfig.java
View file

@ -1,18 +1,22 @@
package ru.spcex.clearing.gatewayapi.config;
import org.springframework.boot.test.context.TestConfiguration;
import org.springframework.boot.web.client.RestTemplateBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.FilterType;
@Configuration
@ComponentScan(basePackages = {"ru.spcex.clearing.gatewayapi.controller", "ru.spcex.clearing.gatewayapi.service"})
@TestConfiguration
@ComponentScan(basePackages = {"ru.spcex.clearing.gatewayapi.config",
"ru.spcex.clearing.gatewayapi.controller",
"ru.spcex.clearing.gatewayapi.service"},
excludeFilters = {@ComponentScan.Filter(type = FilterType.ASSIGNABLE_TYPE, value = KafkaConfig.class),
@ComponentScan.Filter(type = FilterType.ASSIGNABLE_TYPE, value = GatewayApiImdgConfig.class)})
//@EnableConfigurationProperties(value = GatewayApiSettings.class) нужна только если не @SpringBootTest
public class TestConfig {
@Bean
public GatewayApiSettings getGatewayApiSettings() {
GatewayApiSettings gatewayApiSettings = new GatewayApiSettings();
InboundServerSettings inboundServerSettings = new InboundServerSettings();
gatewayApiSettings.setInboundServer(inboundServerSettings);
return gatewayApiSettings;
public RestTemplateBuilder restTemplateBuilder() {
return new RestTemplateBuilder();
}
}

2
clearing-parent/gateway-api/src/test/java/ru/spcex/clearing/gatewayapi/controller/AbstractControllerTest.java
View file

@ -14,7 +14,6 @@ import org.springframework.web.context.WebApplicationContext;
import org.springframework.web.filter.CharacterEncodingFilter;
import ru.spcex.clearing.gatewayapi.config.BeanConfiguration;
import ru.spcex.clearing.gatewayapi.config.RestTemplateErrorHandler;
import ru.spcex.clearing.gatewayapi.config.RestTemplateTestConfig;
import ru.spcex.clearing.gatewayapi.config.TestConfig;
import ru.spcex.clearing.gatewayapi.controller.inbound.request.CommonRequest;
import ru.spcex.clearing.gatewayapi.controller.inbound.response.CommonResponse;
@ -35,7 +34,6 @@ import static ru.spcex.clearing.test.json.MatcherFactoryWithJson.usingIgnoringFi
@ContextConfiguration(classes = {
BeanConfiguration.class,
RestTemplateTestConfig.class,
RestTemplateErrorHandler.class,
TestConfig.class,
ImdgTestConfig.class,

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

@ -1,6 +1,168 @@
package ru.spcex.clearing.gatewayapi.service;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.test.autoconfigure.web.client.AutoConfigureWebClient;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.mock.http.client.MockClientHttpResponse;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit.jupiter.SpringExtension;
import org.springframework.test.web.client.MockRestServiceServer;
import org.springframework.test.web.client.ResponseCreator;
import org.springframework.test.web.servlet.MockMvc;
import org.springframework.test.web.servlet.MvcResult;
import org.springframework.test.web.servlet.ResultActions;
import org.springframework.test.web.servlet.request.MockHttpServletRequestBuilder;
import org.springframework.test.web.servlet.request.MockMvcRequestBuilders;
import org.springframework.test.web.servlet.setup.MockMvcBuilders;
import org.springframework.util.StreamUtils;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.context.WebApplicationContext;
import org.springframework.web.filter.CharacterEncodingFilter;
import ru.spcex.clearing.gatewayapi.config.GatewayApiSettings;
import ru.spcex.clearing.gatewayapi.config.TestConfig;
import ru.spcex.clearing.gatewayapi.controller.inbound.request.limit.LimitRequest;
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.utilities.LimExportedRequest;
import ru.spcex.clearing.test.config.ImdgTestConfig;
import ru.spcex.clearing.test.config.KafkaTestConfig;
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.nio.charset.Charset;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicReference;
import static org.springframework.test.web.client.match.MockRestRequestMatchers.requestTo;
import static org.springframework.test.web.servlet.result.MockMvcResultHandlers.print;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
@ExtendWith(SpringExtension.class)
@ContextConfiguration(classes = {TestConfig.class,
ImdgTestConfig.class,
KafkaTestConfig.class})
@SpringBootTest
@AutoConfigureWebClient(registerRestTemplate = true)
public class GatewayServiceTest {
private static final CharacterEncodingFilter CHARACTER_ENCODING_FILTER = new CharacterEncodingFilter();
public static final String LIMITS_URL = "/limits/";
private MockRestServiceServer mockServer;
@Autowired
@Qualifier("testObjectMapper")
private ObjectMapper mapper;
@Autowired
@Qualifier("rest-template")
private RestTemplate restTemplate;
@Autowired
private GatewayService service;
@Autowired
private GatewayApiSettings settings;
@Autowired
private WebApplicationContext webApplicationContext;
@Autowired
@Qualifier("hazelcastServiceTest")
protected ImdgProvider imdgProvider;
private MockMvc mockMvc;
static {
Path path = Paths.get("src", "main", "resources");
String currentPath = path.toAbsolutePath().toString();
System.setProperty("spring.config.location", currentPath + File.separator);
CHARACTER_ENCODING_FILTER.setEncoding("UTF-8");
CHARACTER_ENCODING_FILTER.setForceEncoding(true);
}
@PostConstruct
private void postConstruct() {
mockServer = MockRestServiceServer.bindTo(restTemplate).build();
mockMvc = MockMvcBuilders
.webAppContextSetup(webApplicationContext)
.addFilter(CHARACTER_ENCODING_FILTER)
// .apply(springSecurity())
.build();
imdgProvider.waitAvailable();
}
protected ResultActions perform(MockHttpServletRequestBuilder builder) throws Exception {
return mockMvc.perform(builder);
}
@Disabled//Для отладки
@Test
void requestOnLimit() throws Exception {
String message = "test response";
BaseRequest<LimExportedRequest> req = new BaseRequest<>();
LimExportedRequest reqP = new LimExportedRequest();
req.setRequestPayload(reqP);
reqP.setSection(Section.FOND.getKey());
reqP.setLimFileName("limits_security_20240108134638.lim");
AtomicReference<UUID> uuid = prepareMockServerResponse(HttpStatus.OK, message);
service.requestOnLimit(req);
LimitRequest request = new LimitRequest();
request.setId(uuid.get());
request.setResult(true);
request.setDescription(message);
MvcResult mvcResult = perform(MockMvcRequestBuilders.post(LIMITS_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.requestOnLimit(req);
}
public AtomicReference<UUID> prepareMockServerResponse(HttpStatus responseStatus, String message) {
mockServer.reset();
AtomicReference<String> body = new AtomicReference<>();
AtomicReference<UUID> uuid = new AtomicReference<>();
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)){
ErrorResponse errorResponse = new ErrorResponse();
errorResponse.setId(request.getId());
errorResponse.setMessage(message);
body.set(mapper.writeValueAsString(errorResponse));
} else {
SuccessResponse errorResponse = new SuccessResponse();
errorResponse.setId(request.getId());
body.set(mapper.writeValueAsString(errorResponse));
}
uuid.set(request.getId());
MockClientHttpResponse response = new MockClientHttpResponse(body.get().getBytes(Charset.defaultCharset()), responseStatus);
response.getHeaders().setContentType(MediaType.APPLICATION_JSON);
return response;
};
mockServer.expect(requestTo(service.formingInboundUrl(settings.getInboundServer().getPathLOCM()))).andRespond(responseCreator);
return uuid;
}
// @Test
// public void makeContentByReportRequestTest() {

12
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/utilities/NotificationNewRequest.java
View file

@ -63,4 +63,16 @@ public class NotificationNewRequest {
public void setPriority(String priority) {
this.priority = priority;
}
@Override
public String toString() {
return "NotificationNewRequest{" +
"senderId=" + senderId +
", addresseeId=" + addresseeId +
", objectType='" + objectType + '\'' +
", objectId=" + objectId +
", comment='" + comment + '\'' +
", priority='" + priority + '\'' +
'}';
}
}