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

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

Browse source
This commit is contained in:
akulikov 2023-06-04 14:24:02 +03:00
parent 46c6d73cd9
commit 501a160f14
6 changed files with 68 additions and 49 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/logic/companies/SendMessageToCompanyServiceWithMemberCompanies.java
View file

@ -22,6 +22,10 @@ import ru.spcex.platform.utils.enumeration.IEnumKey;
import java.util.List;
import java.util.Map;
/**
* Отправка сообщения к company-service на создание/обновление company
* todo Пока убрал совсем не рабочую логику с sendToQueueWaitForAnswer
*/
public class SendMessageToCompanyServiceWithMemberCompanies extends Stage<CompaniesRequestParam> {
private final Logger log = LoggerFactory.getLogger(getClass());
@ -61,7 +65,7 @@ public class SendMessageToCompanyServiceWithMemberCompanies extends Stage<Compan
if (company.isAlreadyExist()) {
kafkaSender.sendRequestToQueue(Consts.DESTINATION_COMPANY_UPDATE, companyNewRequest);
} else {
kafkaSender.sendToQueueWaitForAnswer(Consts.DESTINATION_COMPANY_NEW, companyNewRequest);
kafkaSender.sendRequestToQueue(Consts.DESTINATION_COMPANY_NEW, companyNewRequest);
Company companyFromImdg = companyImdg.getSingleObjectByFieldValues(
Map.of(
"shortName", company.getShortName(),

20
clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/logic/listings_fond/SendMessageToCompanyServiceWithIssuerCompanies.java
View file

@ -20,6 +20,10 @@ import ru.spcex.platform.imdg.api.ImdgProvider;
import java.util.List;
import java.util.Map;
/**
* Отправка сообщения к company-service на создание/обновление company
* todo Пока убрал совсем не рабочую логику с sendToQueueWaitForAnswer
*/
public class SendMessageToCompanyServiceWithIssuerCompanies extends Stage<FondListingsRequestParam> {
private final Logger log = LoggerFactory.getLogger(getClass());
@ -51,7 +55,7 @@ public class SendMessageToCompanyServiceWithIssuerCompanies extends Stage<FondLi
if (company.isAlreadyExist()) {
kafkaSender.sendRequestToQueue(Consts.DESTINATION_COMPANY_UPDATE, companyNewRequest);
} else {
kafkaSender.sendToQueueWaitForAnswer(Consts.DESTINATION_COMPANY_NEW, companyNewRequest);
kafkaSender.sendRequestToQueue(Consts.DESTINATION_COMPANY_NEW, companyNewRequest);
Company companyFromImdg = companyImdg.getSingleObjectByFieldValues(
Map.of(
"shortName", company.getShortName(),
@ -71,12 +75,14 @@ public class SendMessageToCompanyServiceWithIssuerCompanies extends Stage<FondLi
CompanyInfoUpdateRequest companyInfoUpdateRequest = new CompanyInfoUpdateRequest();
companyInfoUpdateRequest.setId(companyId);
companyInfoUpdateRequest.setCountryCode(companyInfo.getCountryCode());
companyInfoUpdateRequest.setLegalKind(companyInfo.getLegalKind());
companyInfoUpdateRequest.setOrganizationType(companyInfo.getOrganizationType());
companyInfoUpdateRequest.setResidence(companyInfo.getResidence());
companyInfoUpdateRequest.setShortNameEng(companyInfo.getShortNameEng());
companyInfoUpdateRequest.setFullNameEng(companyInfo.getFullNameEng());
if (companyInfo != null) {
companyInfoUpdateRequest.setCountryCode(companyInfo.getCountryCode());
companyInfoUpdateRequest.setLegalKind(companyInfo.getLegalKind());
companyInfoUpdateRequest.setOrganizationType(companyInfo.getOrganizationType());
companyInfoUpdateRequest.setResidence(companyInfo.getResidence());
companyInfoUpdateRequest.setShortNameEng(companyInfo.getShortNameEng());
companyInfoUpdateRequest.setFullNameEng(companyInfo.getFullNameEng());
}
kafkaSender.sendRequestToQueue(Consts.DESTINATION_COMPANY_INFO_UPDATE, companyInfoUpdateRequest);
for (IssuerCompanySymbols companySymbols : company.getIssuerCompanySymbolsList()) {

8
clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/logic/listings_fond/SendMessageToSecurityServiceWithFondSecurities.java
View file

@ -26,6 +26,10 @@ import java.util.List;
import java.util.Map;
import java.util.UUID;
/**
* Отправка сообщения к security-service на создание/обновление security
* todo Пока убрал совсем не рабочую логику с sendToQueueWaitForAnswer
*/
public class SendMessageToSecurityServiceWithFondSecurities extends Stage<FondListingsRequestParam> {
private final Logger log = LoggerFactory.getLogger(getClass());
@ -96,7 +100,7 @@ public class SendMessageToSecurityServiceWithFondSecurities extends Stage<FondLi
fixedIncomeSecurityNewRequest.setFullNameEng(security.getFullNameEng());
fixedIncomeSecurityNewRequest.setWorkflowStatus(security.getWorkflowStatus());
fixedIncomeSecurityNewRequest.setInstrumentType(security.getInstrumentType());
kafkaSender.sendToQueueWaitForAnswer(Consts.DESTINATION_FIXED_INCOME_SECURITY_NEW, fixedIncomeSecurityNewRequest);
kafkaSender.sendRequestToQueue(Consts.DESTINATION_FIXED_INCOME_SECURITY_NEW, fixedIncomeSecurityNewRequest);
} else {
FixedIncomeSecurityUpdateRequest fixedIncomeSecurityUpdateRequest = new FixedIncomeSecurityUpdateRequest();
fixedIncomeSecurityUpdateRequest.setId(security.getMapId());
@ -129,7 +133,7 @@ public class SendMessageToSecurityServiceWithFondSecurities extends Stage<FondLi
equitySecurityNewRequest.setFullNameEng(security.getFullNameEng());
equitySecurityNewRequest.setWorkflowStatus(security.getWorkflowStatus());
equitySecurityNewRequest.setInstrumentType(security.getInstrumentType());
kafkaSender.sendToQueueWaitForAnswer(Consts.DESTINATION_EQUITY_SECURITY_NEW, equitySecurityNewRequest);
kafkaSender.sendRequestToQueue(Consts.DESTINATION_EQUITY_SECURITY_NEW, equitySecurityNewRequest);
} else {
EquitySecurityUpdateRequest equitySecurityUpdateRequest = new EquitySecurityUpdateRequest();
equitySecurityUpdateRequest.setId(security.getMapId());

4
clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/logic/listings_mm/SendMessageToSecurityServiceWithMMSecurities.java
View file

@ -11,6 +11,10 @@ import ru.spcex.platform.enumeration.InstrumentType;
import java.util.List;
/**
* Отправка сообщения к company-service на создание/обновление company
* todo Пока убрал совсем не рабочую логику с sendToQueueWaitForAnswer
*/
public class SendMessageToSecurityServiceWithMMSecurities extends Stage<MMListingsRequestParam> {
private final KafkaSender kafkaSender;

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

@ -5,18 +5,25 @@ import org.apache.kafka.clients.producer.Producer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Service;
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.request.InboundRequest;
import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.gateway.GatewayTaskRequest;
import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
import ru.spcex.clearing.platform.messaging.service.RequestInfoUpdate;
import ru.spcex.clearing.util.security.UserRoleVerification;
import ru.spcex.platform.enumeration.Section;
import ru.spcex.platform.enumeration.Task;
import java.time.Instant;
import java.util.Map;
import java.util.UUID;
@Service
public class GatewayService extends QueueConsumer implements InitializingBean {
@ -57,7 +64,15 @@ public class GatewayService extends QueueConsumer implements InitializingBean {
logUnknownProperties(userRequest);
String url = formingInboundUrl(inboundServerSettings.getPathLOCM());
String response = restTemplate.getForObject(url, String.class);
HttpHeaders httpHeaders = new HttpHeaders();
httpHeaders.setContentType(MediaType.APPLICATION_JSON);
InboundRequest inboundRequest = new InboundRequest();
inboundRequest.setId(UUID.randomUUID());
inboundRequest.setType("MEMBER_ON_DEMAND");
inboundRequest.setDatetime(Instant.now());
HttpEntity<InboundRequest> request = new HttpEntity<>(inboundRequest, httpHeaders);
String response = restTemplate.postForObject(url, request, String.class);
log.debug("LOCM task complete, response: {}", response);
return null;
@ -71,10 +86,21 @@ public class GatewayService extends QueueConsumer implements InitializingBean {
logUnknownProperties(userRequest);
String url = formingInboundUrl(inboundServerSettings.getPathLOSC());
String response = restTemplate.getForObject(url, String.class);
String url = formingInboundUrl(inboundServerSettings.getPathLOCM());
HttpHeaders httpHeaders = new HttpHeaders();
httpHeaders.setContentType(MediaType.APPLICATION_JSON);
InboundRequest inboundRequest = new InboundRequest();
inboundRequest.setId(UUID.randomUUID());
inboundRequest.setType("ON_DEMAND");
// todo здесь возникла проблема: место обработки LOCM одно, но в этом месте могут быть сформированы запросы с разными
// section, нужно уточнить как определять какой именно section отправлять в запросе
inboundRequest.setSection(Section.FOND.getKey());
inboundRequest.setDatetime(Instant.now());
HttpEntity<InboundRequest> request = new HttpEntity<>(inboundRequest, httpHeaders);
log.debug("LOSC task complete, response: {}", response);
String response = restTemplate.postForObject(url, request, String.class);
log.debug("LOCM task complete, response: {}", response);
return null;
}

45
clearing-parent/gateway-api/src/test/java/ru/spcex/clearing/gatewayapi/logic/listings_fond/ListingFondRequestTest.java
View file

@ -76,17 +76,18 @@ class ListingFondRequestTest implements InitializingBean {
ProcessResult result = processor.process(param);
// Проверяем только количество запросов, поскольку выполнение этих запросов зависит от security и company модулей
// todo upd: поскольку будет производиться переход к полностью асинхронному запросам, этот тест возможно стоит тоже поправить
Assertions.assertNull(result.getError());
Assertions.assertEquals(sentMessages.get(Consts.DESTINATION_COMPANY_NEW).size(), 2);
Assertions.assertEquals(sentMessages.get(Consts.DESTINATION_COMPANY_UPDATE).size(), 2);
Assertions.assertEquals(sentMessages.get(Consts.DESTINATION_COMPANY_NEW).size(), 1);
Assertions.assertEquals(sentMessages.get(Consts.DESTINATION_COMPANY_INFO_UPDATE).size(), 2);
Assertions.assertEquals(sentMessages.get(Consts.DESTINATION_COMPANY_UPDATE).size(), 1);
Assertions.assertEquals(sentMessages.get(Consts.LISTING_NEW).size(), 4);
Assertions.assertEquals(sentMessages.get(Consts.DESTINATION_COUPON_PERIOD_NEW).size(), 1);
Assertions.assertEquals(sentMessages.get(Consts.DESTINATION_COMPANY_SYMBOL_NEW).size(), 4);
Assertions.assertEquals(sentMessages.get(Consts.DESTINATION_COMPANY_SYMBOL_NEW).size(), 2);
Assertions.assertEquals(sentMessages.get(Consts.DESTINATION_COMPANY_SYMBOL_UPDATE).size(), 2);
Assertions.assertEquals(sentMessages.get(Consts.DESTINATION_CONTACT_NEW).size(), 2);
Assertions.assertEquals(sentMessages.get(Consts.DESTINATION_FIXED_INCOME_SECURITY_NEW).size(), 1);
Assertions.assertEquals(sentMessages.get(Consts.DESTINATION_FIXED_INCOME_SECURITY_UPDATE).size(), 1);
Assertions.assertEquals(sentMessages.get(Consts.DESTINATION_EQUITY_SECURITY_NEW).size(), 1);
Assertions.assertEquals(sentMessages.get(Consts.DESTINATION_EQUITY_SECURITY_UPDATE).size(), 1);
Assertions.assertEquals(sentMessages.get(Consts.DESTINATION_FIXED_INCOME_SECURITY_UPDATE).size(), 2);
Assertions.assertEquals(sentMessages.get(Consts.DESTINATION_EQUITY_SECURITY_UPDATE).size(), 2);
}
@Override
@ -112,13 +113,6 @@ class ListingFondRequestTest implements InitializingBean {
existCompany2Symbols.setCompanyId(company2Id);
companySymbolsImdg.insert(existCompany2Symbols);
/*
"instrumentType", security.getInstrumentType(),
"securitySymbol", security.getSecuritySymbol(),
"workflowStatus", security.getWorkflowStatus(),
"shortName", security.getShortName()
*/
FixedIncomeSecurity fixedIncomeSecurity_1 = new FixedIncomeSecurity();
fixedIncomeSecurity_1.setInstrumentType(InstrumentType.BOND.getKey());
fixedIncomeSecurity_1.setSecuritySymbol("VTB_001P-06 TEST 3");
@ -136,26 +130,6 @@ class ListingFondRequestTest implements InitializingBean {
// мок kafkaSender, чтобы он лишь сохранял в мапу переданные аргументы
Mockito
.when(mockSender.sendRequestToQueue(Mockito.any(), Mockito.any()))
.thenAnswer(
invocation -> {
String destination = invocation.getArgument(0);
Object message = invocation.getArgument(1);
List<Object> currentMessageList = sentMessages.computeIfAbsent(destination, k -> new ArrayList<>());
currentMessageList.add(message);
if (message instanceof CompanyNewRequest companyNewRequest) {
Company company = new Company();
company.setFullName(companyNewRequest.getFullName());
company.setShortName(companyNewRequest.getShortName());
company.setWorkflowStatus(companyNewRequest.getWorkflowStatus());
Long companyId = companyImdg.insert(company);
return companyId;
}
return 0L;
}
);
Mockito
.when(mockSender.sendToQueueWaitForAnswer(Mockito.any(), Mockito.any()))
.thenAnswer(
invocation -> {
String destination = invocation.getArgument(0);
@ -186,7 +160,8 @@ class ListingFondRequestTest implements InitializingBean {
List<Object> currentMessageList = sentMessages.computeIfAbsent(destination, k -> new ArrayList<>());
currentMessageList.add(invocationArgument);
return "ok";
return 0L;
}
);