diff --git a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/config/PipelineConfiguration.java b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/config/PipelineConfiguration.java deleted file mode 100644 index 2984a1225..000000000 --- a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/config/PipelineConfiguration.java +++ /dev/null @@ -1,30 +0,0 @@ -package ru.spcex.clearing.gatewayapi.config; - -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import ru.spcex.clearing.gatewayapi.logic.Stage; -import ru.spcex.clearing.gatewayapi.logic.listings_fond.*; -import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender; -import ru.spcex.platform.imdg.api.ImdgProvider; - -import java.util.ArrayList; -import java.util.List; - -@Configuration -public class PipelineConfiguration { - - @Bean - public List> listingsRequestPipeline(ImdgProvider imdgProvider, KafkaSender kafkaSender) { - List> pipeline = new ArrayList<>(); - pipeline.add(new PrepareSecurities()); - pipeline.add(new PrepareIssuerCompanies()); - pipeline.add(new ValidateIncomeSecurities()); - pipeline.add(new ValidateIssuerCompanies()); - pipeline.add(new CheckFondSecurityExist(imdgProvider)); - pipeline.add(new CheckIssuerCompanyExist(imdgProvider)); - pipeline.add(new SendMessageToCompanyServiceWithIssuerCompanies(kafkaSender, imdgProvider)); - pipeline.add(new SendMessageToSecurityServiceWithFondSecurities(kafkaSender, imdgProvider)); - return pipeline; - } - -} diff --git a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/config/ProcessorConfiguration.java b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/config/ProcessorConfiguration.java new file mode 100644 index 000000000..98dfb4f85 --- /dev/null +++ b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/config/ProcessorConfiguration.java @@ -0,0 +1,63 @@ +package ru.spcex.clearing.gatewayapi.config; + +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import ru.spcex.clearing.gatewayapi.logic.Processor; +import ru.spcex.clearing.gatewayapi.logic.Stage; +import ru.spcex.clearing.gatewayapi.logic.companies.*; +import ru.spcex.clearing.gatewayapi.logic.listings_fond.*; +import ru.spcex.clearing.gatewayapi.logic.listings_mm.*; +import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender; +import ru.spcex.platform.imdg.api.ImdgProvider; + +import java.util.ArrayList; +import java.util.List; + +@Configuration +public class ProcessorConfiguration { + + @Qualifier("fondListingsRequestProcessor") + @Bean + public Processor fondListingsRequestProcessor(ImdgProvider imdgProvider, KafkaSender kafkaSender) { + Processor processor = new Processor<>(); + List> pipeline = new ArrayList<>(); + pipeline.add(new PrepareSecurities()); + pipeline.add(new PrepareIssuerCompanies()); + pipeline.add(new ValidateIncomeSecurities()); + pipeline.add(new ValidateIssuerCompanies()); + pipeline.add(new CheckFondSecurityExist(imdgProvider)); + pipeline.add(new CheckIssuerCompanyExist(imdgProvider)); + pipeline.add(new SendMessageToCompanyServiceWithIssuerCompanies(kafkaSender, imdgProvider)); + pipeline.add(new SendMessageToSecurityServiceWithFondSecurities(kafkaSender, imdgProvider)); + processor.setPipeline(pipeline); + return processor; + } + + @Qualifier("mmListingsRequestProcessor") + @Bean + public Processor mmListingsRequestProcessor(ImdgProvider imdgProvider, KafkaSender kafkaSender) { + Processor processor = new Processor<>(); + List> pipeline = new ArrayList<>(); + pipeline.add(new PrepareExchangeInstruments()); + pipeline.add(new ValidateExchangeInstruments(imdgProvider)); + pipeline.add(new CheckMoneyMarketSecurityExist(imdgProvider)); + pipeline.add(new SendMessageToSecurityServiceWithMMSecurities(kafkaSender)); + processor.setPipeline(pipeline); + return processor; + } + + @Qualifier("companiesRequestProcessor") + @Bean + public Processor companiesRequestProcessor(ImdgProvider imdgProvider, KafkaSender kafkaSender) { + Processor processor = new Processor<>(); + List> pipeline = new ArrayList<>(); + pipeline.add(new PrepareMemberCompanies()); + pipeline.add(new ValidateMemberCompanies()); + pipeline.add(new CheckMemberCompanyExist(imdgProvider)); + pipeline.add(new SendMessageToCompanyServiceWithMemberCompanies(kafkaSender, imdgProvider)); + processor.setPipeline(pipeline); + return processor; + } + +} diff --git a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/logic/ProcessResult.java b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/logic/ProcessResult.java index cf569bb07..663d2db83 100644 --- a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/logic/ProcessResult.java +++ b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/logic/ProcessResult.java @@ -1,4 +1,13 @@ package ru.spcex.clearing.gatewayapi.logic; public class ProcessResult { + private boolean ok = true; + + public boolean isOk() { + return ok; + } + + public void setOk(boolean ok) { + this.ok = ok; + } } diff --git a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/logic/Processor.java b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/logic/Processor.java index 23f9b5863..9618db44f 100644 --- a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/logic/Processor.java +++ b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/logic/Processor.java @@ -2,7 +2,7 @@ package ru.spcex.clearing.gatewayapi.logic; import java.util.List; -public abstract class Processor { +public class Processor { private List> pipeline; public ProcessResult process(T param) { diff --git a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/logic/listings_mm/CheckMoneyMarketSecurityExist.java b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/logic/listings_mm/CheckMoneyMarketSecurityExist.java index 9e2af8c98..53c522f01 100644 --- a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/logic/listings_mm/CheckMoneyMarketSecurityExist.java +++ b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/logic/listings_mm/CheckMoneyMarketSecurityExist.java @@ -29,7 +29,11 @@ public class CheckMoneyMarketSecurityExist extends Stage List exchangeInstrumentList = param.getExchangeInstrumentsList(); for (ExchangeInstrument exchangeInstrument : exchangeInstrumentList) { + if (exchangeInstrument.isInvalidData()) continue; + String securitySymbol = extractSecuritySymbol(exchangeInstrument.getCode()); + exchangeInstrument.setSecuritySymbol(securitySymbol); + Collection securitiesFromImdg = moneyMarketSecurityImdg.getCollectionObjectsByFieldValues( Map.of( "securitySymbol", securitySymbol, diff --git a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/logic/listings_mm/SendMessageToSecurityServiceWithMMSecurities.java b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/logic/listings_mm/SendMessageToSecurityServiceWithMMSecurities.java new file mode 100644 index 000000000..2bf591fc5 --- /dev/null +++ b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/logic/listings_mm/SendMessageToSecurityServiceWithMMSecurities.java @@ -0,0 +1,54 @@ +package ru.spcex.clearing.gatewayapi.logic.listings_mm; + +import ru.spcex.clearing.gatewayapi.logic.ProcessResult; +import ru.spcex.clearing.gatewayapi.logic.Stage; +import ru.spcex.clearing.gatewayapi.request.objects.ExchangeInstrument; +import ru.spcex.clearing.platform.messaging.domain.Consts; +import ru.spcex.clearing.platform.messaging.domain.cud.securitites.MoneyMarketSecurityNewRequest; +import ru.spcex.clearing.platform.messaging.domain.cud.securitites.MoneyMarketSecurityUpdateRequest; +import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender; +import ru.spcex.platform.enumeration.InstrumentType; + +import java.util.List; + +public class SendMessageToSecurityServiceWithMMSecurities extends Stage { + private final KafkaSender kafkaSender; + + public SendMessageToSecurityServiceWithMMSecurities(KafkaSender kafkaSender) { + this.kafkaSender = kafkaSender; + } + + @Override + public ProcessResult process(MMListingsRequestParam param) { + List exchangeInstrumentList = param.getExchangeInstrumentsList(); + for (ExchangeInstrument exchangeInstrument : exchangeInstrumentList) { + if (exchangeInstrument.isInvalidData()) continue; + + if (!exchangeInstrument.isAlreadyExist()) { + MoneyMarketSecurityNewRequest moneyMarketSecurityNewRequest = new MoneyMarketSecurityNewRequest(); + moneyMarketSecurityNewRequest.setNominalCurrency(exchangeInstrument.getSettlementCurrencyLetterCode()); + moneyMarketSecurityNewRequest.setInstrumentType(InstrumentType.RATE.getKey()); + moneyMarketSecurityNewRequest.setShortName(exchangeInstrument.getName()); + moneyMarketSecurityNewRequest.setSecuritySymbol(exchangeInstrument.getSecuritySymbol()); + moneyMarketSecurityNewRequest.setLotSize(exchangeInstrument.getLot()); + moneyMarketSecurityNewRequest.setTermType(exchangeInstrument.getBankDepositAgreementType()); + moneyMarketSecurityNewRequest.setIssuerId(exchangeInstrument.getIssuerId()); + moneyMarketSecurityNewRequest.setWorkflowStatus(exchangeInstrument.getWorkflowStatus()); + kafkaSender.sendRequestToQueue(Consts.DESTINATION_MONEY_MARKET_SECURITY_NEW, moneyMarketSecurityNewRequest); + } else { + MoneyMarketSecurityUpdateRequest moneyMarketSecurityUpdateRequest = new MoneyMarketSecurityUpdateRequest(); + moneyMarketSecurityUpdateRequest.setId(exchangeInstrument.getMapId()); + moneyMarketSecurityUpdateRequest.setNominalCurrency(exchangeInstrument.getSettlementCurrencyLetterCode()); + moneyMarketSecurityUpdateRequest.setInstrumentType(InstrumentType.RATE.getKey()); + moneyMarketSecurityUpdateRequest.setShortName(exchangeInstrument.getName()); + moneyMarketSecurityUpdateRequest.setSecuritySymbol(exchangeInstrument.getSecuritySymbol()); + moneyMarketSecurityUpdateRequest.setLotSize(exchangeInstrument.getLot()); + moneyMarketSecurityUpdateRequest.setTermType(exchangeInstrument.getBankDepositAgreementType()); + moneyMarketSecurityUpdateRequest.setIssuerId(exchangeInstrument.getIssuerId()); + moneyMarketSecurityUpdateRequest.setWorkflowStatus(exchangeInstrument.getWorkflowStatus()); + kafkaSender.sendRequestToQueue(Consts.DESTINATION_MONEY_MARKET_SECURITY_NEW, moneyMarketSecurityUpdateRequest); + } + } + return null; + } +} diff --git a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/logic/listings_mm/ValidateExchangeInstruments.java b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/logic/listings_mm/ValidateExchangeInstruments.java index 79191bc30..2446aab6b 100644 --- a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/logic/listings_mm/ValidateExchangeInstruments.java +++ b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/logic/listings_mm/ValidateExchangeInstruments.java @@ -10,6 +10,7 @@ import ru.spcex.clearing.gatewayapi.logic.Stage; import ru.spcex.clearing.gatewayapi.request.objects.ExchangeInstrument; import ru.spcex.clearing.imdg.IMDGDistributedNames; import ru.spcex.platform.enumeration.CompanySymbol; +import ru.spcex.platform.enumeration.TermType; import ru.spcex.platform.enumeration.WorkflowStatus; import ru.spcex.platform.imdg.api.Imdg; import ru.spcex.platform.imdg.api.ImdgProvider; @@ -89,6 +90,13 @@ public class ValidateExchangeInstruments extends Stage { exchangeInstrument.setInvalidData(true); } + String bankDepositAgreementTypeStr = exchangeInstrument.getBankDepositAgreementType(); + TermType termType = IEnumKey.getEnumByKey(TermType.class, bankDepositAgreementTypeStr); + if (termType == null) { + log.warn("Invalid bank_deposit_agreement_type {} for exchange_instrument (id = {})", bankDepositAgreementTypeStr, exchangeInstrument.getId()); + exchangeInstrument.setInvalidData(true); + } + } List filteredExchangeInstrument = exchangeInstrumentList.stream() diff --git a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/request/objects/ExchangeInstrument.java b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/request/objects/ExchangeInstrument.java index 4132b7372..1caa47703 100644 --- a/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/request/objects/ExchangeInstrument.java +++ b/clearing-parent/gateway-api/src/main/java/ru/spcex/clearing/gatewayapi/request/objects/ExchangeInstrument.java @@ -6,6 +6,7 @@ import com.fasterxml.jackson.databind.annotation.JsonDeserialize; import io.swagger.annotations.ApiModelProperty; import ru.spcex.clearing.gatewayapi.config.deserializers.LocalDateDeserializer; +import java.math.BigDecimal; import java.time.LocalDate; import java.util.UUID; @@ -130,7 +131,7 @@ public class ExchangeInstrument extends WithMapId { value = "Лот", example = "10" ) - private Long lot; + private BigDecimal lot; @JsonProperty("order_execution_type_by_price") @ApiModelProperty( @@ -213,6 +214,10 @@ public class ExchangeInstrument extends WithMapId { @ApiModelProperty(hidden = true) private Long issuerId; + @JsonProperty + @ApiModelProperty(hidden = true) + private String securitySymbol; + public UUID getId() { return id; @@ -390,11 +395,11 @@ public class ExchangeInstrument extends WithMapId { this.precision = precision; } - public Long getLot() { + public BigDecimal getLot() { return lot; } - public void setLot(Long lot) { + public void setLot(BigDecimal lot) { this.lot = lot; } @@ -405,4 +410,12 @@ public class ExchangeInstrument extends WithMapId { public void setIssuerId(Long issuerId) { this.issuerId = issuerId; } + + public String getSecuritySymbol() { + return securitySymbol; + } + + public void setSecuritySymbol(String securitySymbol) { + this.securitySymbol = securitySymbol; + } } \ No newline at end of file diff --git a/platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/TermType.java b/platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/TermType.java new file mode 100644 index 000000000..a3cbb46e7 --- /dev/null +++ b/platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/TermType.java @@ -0,0 +1,21 @@ +package ru.spcex.platform.enumeration; + +import ru.spcex.platform.utils.enumeration.IEnumKey; + +public enum TermType implements IEnumKey { + S("S"), // Срочный + V("V"), // До востребования + K("K"), // Комбинированный + ; + + private final String key; + + TermType(String key) { + this.key = key; + } + + @Override + public String getKey() { + return key; + } +}