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

http://jira.mfd.msk:8088/browse/CLS-272 ClientCodeService, доделал BiDirectionQueueExchanger. Необходимо это покрыть тестами.

Browse source
This commit is contained in:
AKurakin 2023-04-27 18:55:14 +03:00
parent aa265afd86
commit ae1c39f8b4
7 changed files with 784 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

180
clearing-parent/clearing-validation/src/main/java/ru/spcex/clearing/util/services/exchangers/BiDirectionQueueExchanger.java
View file

@ -1,25 +1,56 @@
package ru.spcex.clearing.util.services.exchangers;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import ru.spcex.clearing.platform.messaging.domain.ActionType;
import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
import ru.spcex.clearing.platform.messaging.domain.Consts;
import ru.spcex.clearing.platform.messaging.domain.cud.balance.AccountBalanceClearingRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.clearing.Sdf04Request;
import ru.spcex.clearing.platform.messaging.domain.cud.common.CommonIdRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.schedule.LauncherCommandRequest;
import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
import ru.spcex.clearing.platform.messaging.service.RequestInfoUpdate;
import ru.spcex.clearing.platform.messaging.service.Status;
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
import ru.spcex.platform.enumeration.Task;
import ru.spcex.platform.utils.log.ExceptionUtils;
import java.io.Closeable;
import java.util.Objects;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;
/**
* Синхронный обмен сообщениями с ассинхронным сервисом.
* Метод exchange() отправляет сообщение в очередь и дожидается ответа из другой очереди
* Метод exchange() отправляет сообщение в очередь и дожидается ответа из другой очереди.
* Использовать с осторожностью, т.к. нет возобновления ожидания в случае нештатного выключения модуля при ожидании ответа из kafka.
* <p>
* <p>
* Ожидает из очереди (CommonIdRequest)
*
* @param <TOut> отправляется в очередь
*/
public class BiDirectionQueueExchanger<TIn, TOut extends BaseRequest<?>> extends QueueConsumer implements InitializingBean, DisposableBean {
public class BiDirectionQueueExchanger<TOut extends BaseRequest<?>> extends QueueConsumer implements Closeable {
protected final Logger log = LoggerFactory.getLogger(getClass());
protected final Object sync = new Object();
protected String outQueue;
protected String inQueue;
protected Class<TIn> listenClass;
protected long timeout;
private Producer<String, Object> producer;
protected volatile boolean isTerminated;
protected Object syncObject;
protected Long lastSentRequestId;
protected volatile Long lastResponseId;
protected boolean ignoreOtherResponse = true; // Пропускать другие ID, пока не получит lastResponseId==lastSentRequestId
/**
* Синхронно-ассинхронный обмен сообщениями
@ -27,40 +58,153 @@ public class BiDirectionQueueExchanger<TIn, TOut extends BaseRequest<?>> extends
* @param kafkaQueue
* @param kafkaProducer
* @param outQueue отправляет в очередь
* @param inQueue слушает очередь, ожидает ответов
* @param listenClass типы объектов из inQueue
* @param timeout - максимальное ожидание ответа, в миллисекундах, 0 - неограничено
* @param inQueue слушает очередь/топик, ожидает ответов. Пример: Consts.CONTINUE_CLEARING
* @param timeout - максимальное ожидание ответа, в миллисекундах, 0 - неограничено
*/
public BiDirectionQueueExchanger(Consumer<String, Object> kafkaQueue, Producer<String, Object> kafkaProducer,
String outQueue,
String inQueue, Class<TIn> listenClass,
String inQueue,
long timeout) {
super(kafkaQueue, kafkaProducer);
super(kafkaQueue);
this.producer = kafkaProducer;
if (kafkaQueue == null) {
throw new IllegalArgumentException("kafkaQueue is empty");
}
if (kafkaProducer == null) {
throw new IllegalArgumentException("kafkaProducer is empty");
}
if (StringUtils.isEmpty(outQueue)) {
throw new IllegalArgumentException("outQueue is empty");
}
this.outQueue = outQueue;
if (StringUtils.isEmpty(inQueue)) {
throw new IllegalArgumentException("inQueue is empty");
}
this.inQueue = inQueue;
this.listenClass = listenClass;
if (timeout < 0) {
timeout = 0;
}
this.timeout = timeout;
init();
}
/**
* Отправить сообщение message в outQueue и дождаться ответа из очереди inQueue
*
* @param message
* @return
* @throws InterruptedException
* @return request/reply id
* @throws InterruptedException или когда поток прерывают, или когда BiDirectionQueueExchanger.close()
*/
public TOut exchange(TIn message) throws InterruptedException {
//todo impl BiDirectionQueueExchanger
return null;
public Long exchange(TOut message) throws InterruptedException {
if (syncObject == null)
throw new IllegalStateException("Listener queue not initialized");
if (message.getId() == null) {
throw new IllegalArgumentException("Required message " + message.getClass().getSimpleName() + ".id is null");
}
//send request to kafka, wait for a reply
try {
synchronized (syncObject) {
Long sentRequestId = sendMessage(message);
log.trace("sent request to kafka, requestId: {}", sentRequestId);
this.lastSentRequestId = sentRequestId;
long waitStart = System.currentTimeMillis();
long waitTime = timeout;
do {
try {
if (timeout > 0) {
syncObject.wait(waitTime);
} else {
syncObject.wait();
}
//если на этом месте произошла ошибка - непонятно как обрабатывать
//т.к. request на самом деле могла быть проблема с сетью/недоступностью кафки etc.
} catch (InterruptedException e) {
log.error(ExceptionUtils.getStackTrace(e));
throw e;
// Thread.currentThread().interrupt();
// throw new RuntimeException(e);
}
waitTime = timeout - (System.currentTimeMillis() - waitStart);
if (isTerminated && lastResponseId == null) {
log.warn("Terminated waiter {} kafka at thread {}", this, Thread.currentThread().getName());
throw new InterruptedException(getClass().getSimpleName() + " closed");
}
} while (ignoreOtherResponse && !sentRequestId.equals(lastResponseId) && (timeout > 0 && waitTime > 0));
if (!sentRequestId.equals(lastResponseId)) {
log.error("FATAL: last response from {} update from Kafka ID was {}, but sent ID was {}",
inQueue, lastResponseId, sentRequestId);
throw new IllegalStateException("Expected wait till requestId=" + sentRequestId + " but catch lastResponseId=" + lastResponseId);
}
return sentRequestId;
}
} finally {
this.lastSentRequestId = null;
}
}
@Override
public void destroy() throws Exception {
public void close() {
isTerminated = true;
if (lastSentRequestId != null) {
log.warn("Finished before wait, lastSentRequestId={}", lastSentRequestId);
if (syncObject != null) {
synchronized (syncObject) {
lastResponseId = null;
syncObject.notifyAll();
}
}
}
syncObject = null;
log.debug("Listener {} for async exchange {}-{} close", this, outQueue, inQueue);
}
@Override
public void afterPropertiesSet() throws Exception {
public void init() {
if (syncObject != null) {
throw new IllegalStateException("Already initialized");
}
callback(CommonIdRequest.class)
.setConsumer(this::continueWaiting)
.forDestination(inQueue, callbacks::put);
init();
syncObject = new Object();
isTerminated = false;
log.debug("Listener {} for async exchange {}-{} ready", this, outQueue, inQueue);
}
protected Long sendMessage(TOut message) throws InterruptedException {
Long sentRequestId = message.getId();
if (sentRequestId == null) {
throw new IllegalArgumentException("Message " + message.getClass().getSimpleName() + " required id.");
}
try {
Future<RecordMetadata> send = producer.send(new ProducerRecord<>(Consts.REQUEST_INFO_UPDATE, message));
send.get();
return sentRequestId;
} catch (InterruptedException e) {
log.error("Interrupt when send message, {}", ExceptionUtils.getStackTrace(e));
Thread.currentThread().interrupt();
throw e;
} catch (ExecutionException e) {
log.error("Error at send message, {} cause: {}", e.toString(), ExceptionUtils.getStackTrace(e.getCause()));
throw new RuntimeException("Send message error", e.getCause());
} catch (Exception e) {
log.error("Unexpected exception when send message: {}", ExceptionUtils.getStackTrace(e));
}
return null;
}
protected void continueWaiting(BaseRequest<CommonIdRequest> event) {
synchronized (syncObject) {
CommonIdRequest requestPayload = event.getRequestPayload();
log.debug("continueWaiting {} for id={}", inQueue, requestPayload.getId());
lastResponseId = requestPayload.getId();
syncObject.notifyAll();
}
}
}

184
clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/config/validation/ClientCodeValidationConfig.java Normal file
View file

@ -0,0 +1,184 @@
package ru.spcex.clearing.company.config.validation;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import ru.clearing.classes.statics.data.account.Account;
import ru.clearing.classes.statics.data.account.ClientCode;
import ru.clearing.classes.statics.data.company.Company;
import ru.clearing.platform.dictionary.ClearingCategoryDictionary;
import ru.clearing.platform.dictionary.WorkflowStatusDictionary;
import ru.spcex.clearing.company.error.CompanyErrors;
import ru.spcex.clearing.imdg.IMDGDistributedNames;
import ru.spcex.clearing.platform.messaging.domain.cud.account.ClientCodeNewRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.account.ClientCodeUpdateRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.common.CommonDeleteRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.company.ClearingMemberCategoryNewRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.company.CompanySymbolUpdateRequest;
import ru.spcex.clearing.validation.common.rules.DictionaryPresentRule;
import ru.spcex.clearing.validation.common.rules.FieldRequiredRule;
import ru.spcex.clearing.validation.common.rules.IdPresentRule;
import ru.spcex.platform.classes.base.SpcexObjectBase;
import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.validation.ImdgValidationContext;
import ru.spcex.platform.utils.validation.IValidator;
import ru.spcex.platform.utils.validation.ValidatorImpl;
import java.util.Map;
import java.util.Objects;
import java.util.function.Consumer;
import java.util.function.Function;
@Configuration
public class ClientCodeValidationConfig {
@Bean("clientCodeNewRequestValidator")
public Function<ClientCodeNewRequest, IValidator> clientCodeNewRequestValidator(Map<String, Imdg<? extends SpcexObjectBase>> imdgForValidation) {
return clientCodeUpdateRequest -> {
ImdgValidationContext<ClientCodeNewRequest> context = new ImdgValidationContext<>();
context.setValidatedObject(clientCodeUpdateRequest);
Consumer<String> addImdg = (s) -> context.addImdg(s, imdgForValidation.get(s));
addImdg.accept(IMDGDistributedNames.Map_Company);
addImdg.accept(IMDGDistributedNames.Map_Account);
addImdg.accept(IMDGDistributedNames.Map_TradingClearingRegistry);
addImdg.accept(IMDGDistributedNames.Map_WorkflowStatusDictionary);
return new ValidatorImpl<>(context,
FieldRequiredRule.instance("companyId", CompanySymbolUpdateRequest::getCompanyId, CompanyErrors.RequiredFieldEmpty),
IdPresentRule.instance("companyId",
ClientCodeNewRequest::getCompanyId,
IMDGDistributedNames.Map_Company,
Company.class,
CompanyErrors.RequiredFieldEmpty,
CompanyErrors.CompanyNotFound),
IdPresentRule.instance("moneyAccountId",
ClientCodeNewRequest::getMoneyAccountId,
IMDGDistributedNames.Map_Account,
Account.class,
CompanyErrors.RequiredFieldEmpty,
CompanyErrors.AccountNotFound,
false,
acc -> Objects.equals(context.getValidatedObject().getCompanyId(), acc.getCompanyId())
? null : CompanyErrors.AccountNotFound
),
IdPresentRule.instance("depoAccountId",
ClientCodeNewRequest::getDepoAccountId,
IMDGDistributedNames.Map_Account,
Account.class,
CompanyErrors.RequiredFieldEmpty,
CompanyErrors.AccountNotFound,
false,
acc -> Objects.equals(context.getValidatedObject().getCompanyId(), acc.getCompanyId())
? null : CompanyErrors.AccountNotFound
),
IdPresentRule.instance("tradingClearingRegistryId",
ClientCodeNewRequest::getTradingClearingRegistryId,
IMDGDistributedNames.Map_TradingClearingRegistry,
Account.class,
CompanyErrors.RequiredFieldEmpty,
CompanyErrors.TradingClearingRegistryNotFound,
false,
acc -> Objects.equals(context.getValidatedObject().getCompanyId(), acc.getCompanyId())
? null : CompanyErrors.TradingClearingRegistryNotFound
),
DictionaryPresentRule.instance("stauts",
ClientCodeNewRequest::getStatus,
IMDGDistributedNames.Map_WorkflowStatusDictionary,
WorkflowStatusDictionary.class,
CompanyErrors.RequiredFieldEmpty,
CompanyErrors.DictionaryNotFound,
false)
);
};
}
@Bean("clientCodeUpdateRequestValidator")
public Function<ClientCodeUpdateRequest, IValidator> clientCodeUpdateRequestValidator(Map<String, Imdg<? extends SpcexObjectBase>> imdgForValidation) {
return clientCodeUpdateRequest -> {
ImdgValidationContext<ClientCodeUpdateRequest> context = new ImdgValidationContext<>();
context.setValidatedObject(clientCodeUpdateRequest);
Consumer<String> addImdg = (s) -> context.addImdg(s, imdgForValidation.get(s));
addImdg.accept(IMDGDistributedNames.Map_ClientCode);
addImdg.accept(IMDGDistributedNames.Map_Company);
addImdg.accept(IMDGDistributedNames.Map_Account);
addImdg.accept(IMDGDistributedNames.Map_TradingClearingRegistry);
addImdg.accept(IMDGDistributedNames.Map_WorkflowStatusDictionary);
return new ValidatorImpl<ImdgValidationContext<ClientCodeUpdateRequest>>(context,
IdPresentRule.instance("id",
ClientCodeUpdateRequest::getId,
IMDGDistributedNames.Map_ClientCode,
ClientCode.class,
CompanyErrors.RequiredFieldEmpty,
CompanyErrors.RecordNotFound
),
FieldRequiredRule.instance("companyId", CompanySymbolUpdateRequest::getCompanyId, CompanyErrors.RequiredFieldEmpty),
IdPresentRule.instance("companyId",
ClientCodeUpdateRequest::getCompanyId,
IMDGDistributedNames.Map_Company,
Company.class,
CompanyErrors.RequiredFieldEmpty,
CompanyErrors.CompanyNotFound),
IdPresentRule.instance("moneyAccountId",
ClientCodeUpdateRequest::getMoneyAccountId,
IMDGDistributedNames.Map_Account,
Account.class,
CompanyErrors.RequiredFieldEmpty,
CompanyErrors.AccountNotFound,
false,
acc -> Objects.equals(context.getValidatedObject().getCompanyId(), acc.getCompanyId())
? null : CompanyErrors.AccountNotFound
),
IdPresentRule.instance("depoAccountId",
ClientCodeUpdateRequest::getDepoAccountId,
IMDGDistributedNames.Map_Account,
Account.class,
CompanyErrors.RequiredFieldEmpty,
CompanyErrors.AccountNotFound,
false,
acc -> Objects.equals(context.getValidatedObject().getCompanyId(), acc.getCompanyId())
? null : CompanyErrors.AccountNotFound
),
IdPresentRule.instance("tradingClearingRegistryId",
ClientCodeUpdateRequest::getTradingClearingRegistryId,
IMDGDistributedNames.Map_TradingClearingRegistry,
Account.class,
CompanyErrors.RequiredFieldEmpty,
CompanyErrors.TradingClearingRegistryNotFound,
false,
acc -> Objects.equals(context.getValidatedObject().getCompanyId(), acc.getCompanyId())
? null : CompanyErrors.TradingClearingRegistryNotFound
),
DictionaryPresentRule.instance("stauts",
ClientCodeUpdateRequest::getStatus,
IMDGDistributedNames.Map_WorkflowStatusDictionary,
WorkflowStatusDictionary.class,
CompanyErrors.RequiredFieldEmpty,
CompanyErrors.DictionaryNotFound,
false)
);
};
}
@Bean("clientCodeDeleteRequestValidator")
public Function<CommonDeleteRequest, IValidator> clientCodeDeleteRequestValidator(Map<String, Imdg<? extends SpcexObjectBase>> imdgForValidation) {
return companyDeleteRequest -> {
ImdgValidationContext<CommonDeleteRequest> context = new ImdgValidationContext<>();
context.setValidatedObject(companyDeleteRequest);
Consumer<String> addImdg = (s) -> context.addImdg(s, imdgForValidation.get(s));
addImdg.accept(IMDGDistributedNames.Map_Company);
return new ValidatorImpl<>(context,
// FieldRequiredRule.instance("id", CommonDeleteRequest::getId, CompanyErrors.RequiredFieldEmpty),
IdPresentRule.instance("id",
CommonDeleteRequest::getId,
IMDGDistributedNames.Map_ClientCode,
ClientCode.class,
CompanyErrors.RequiredFieldEmpty,
CompanyErrors.RecordNotFound)
);
};
}
}

7
clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/config/validation/ValidationConfig.java
View file

@ -2,11 +2,14 @@ package ru.spcex.clearing.company.config.validation;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import ru.clearing.classes.statics.data.account.Account;
import ru.clearing.classes.statics.data.account.ClientCode;
import ru.clearing.classes.statics.data.company.Company;
import ru.clearing.classes.statics.data.company.CompanySymbols;
import ru.clearing.classes.statics.data.company.ClearingMemberCategory;
import ru.clearing.classes.statics.data.profile.Contact;
import ru.clearing.classes.statics.data.profile.ProfileDocument;
import ru.clearing.classes.statics.data.registry.TradingClearingRegistry;
import ru.clearing.platform.dictionary.*;
import ru.spcex.clearing.imdg.IMDGDistributedNames;
import ru.spcex.clearing.validation.common.ValidationHelper;
@ -44,6 +47,10 @@ public class ValidationConfig {
addImdg.accept(IMDGDistributedNames.Map_LegalKindDictionary, LegalKindDictionary.class);
addImdg.accept(IMDGDistributedNames.Map_OrganizationTypeDictionary, OrganizationTypeDictionary.class);
addImdg.accept(IMDGDistributedNames.Map_ClientCode, ClientCode.class);
addImdg.accept(IMDGDistributedNames.Map_Account, Account.class);
addImdg.accept(IMDGDistributedNames.Map_TradingClearingRegistry, TradingClearingRegistry.class);
return imdg;
}

4
clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/error/CompanyErrors.java
View file

@ -22,7 +22,9 @@ public enum CompanyErrors implements IErrorEnumId {
CompanyAlreadyHasLiabilities(3018L), // У компании %s присутствуют обязательства.
CompanyWithCompanySymbolAlreadyExist(3019L), // Компания с %s = %s уже создана" (где первый %s - companySymbol, второй %s - companySymbolValue)
EditCompanySymbols(3020L), // Тип реквизита компании %s не может быть изменен.
EditContactType(3021L) // Тип контакта компании %s не может быть изменен.
EditContactType(3021L), // Тип контакта компании %s не может быть изменен.
TradingClearingRegistryNotFound(3022L), // ТКР с %s не найден
AccountNotFound(3023L), // Счет %s не найден
;
private final Long id;

416
clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/ClientCodeService.java Normal file
View file

@ -0,0 +1,416 @@
package ru.spcex.clearing.company.service;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import ru.clearing.classes.statics.data.account.ClientCode;
import ru.clearing.classes.statics.data.company.Company;
import ru.clearing.classes.statics.data.registry.TradingClearingRegistry;
import ru.spcex.clearing.company.error.CompanyErrors;
import ru.spcex.clearing.imdg.IMDGDistributedNames;
import ru.spcex.clearing.platform.messaging.domain.ActionType;
import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
import ru.spcex.clearing.platform.messaging.domain.Consts;
import ru.spcex.clearing.platform.messaging.domain.cud.account.ClientCodeNewRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.account.ClientCodeUpdateRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.common.CommonDeleteRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.registry.TradingClearingRegistryNewRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.registry.TradingClearingRegistryUpdateRequest;
import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
import ru.spcex.clearing.platform.messaging.service.RequestInfoUpdate;
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
import ru.spcex.clearing.util.security.UserRoleVerification;
import ru.spcex.clearing.util.services.exchangers.BiDirectionQueueExchanger;
import ru.spcex.clearing.validation.common.ValidationHelper;
import ru.spcex.platform.enumeration.TradingClearingRegistryType;
import ru.spcex.platform.enumeration.WorkflowStatus;
import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.ImdgId;
import ru.spcex.platform.imdg.api.ImdgProvider;
import ru.spcex.platform.imdg.iml.hazelcast.adapter.ImdgIdGeneratorHazelcast;
import ru.spcex.platform.utils.enumeration.EnumMessage;
import ru.spcex.platform.utils.enumeration.IMessageResolver;
import ru.spcex.platform.utils.error.ClearingBaseException;
import ru.spcex.platform.utils.log.ExceptionUtils;
import ru.spcex.platform.utils.validation.IValidator;
import java.time.Instant;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.function.Function;
@Service
public class ClientCodeService extends QueueConsumer implements InitializingBean, DisposableBean {
private final Logger log = LoggerFactory.getLogger(getClass());
private final Producer<String, Object> kafkaProducer;
private final ImdgId idGenerator;
private final Imdg<ClientCode> clientCodeMap;
private final Imdg<Company> companyMap;
private final Imdg<TradingClearingRegistry> tradingClearingRegistryMap;
private final Function<ClientCodeNewRequest, IValidator> clientCodeNewRequestValidator;
private final Function<ClientCodeUpdateRequest, IValidator> clientCodeUpdateRequestValidator;
private final Function<CommonDeleteRequest, IValidator> clientCodeDeleteRequestValidator;
private final ValidationHelper validationHelper;
private final UserRoleVerification userRoleVerification;
private final IMessageResolver messageResolver;
BiDirectionQueueExchanger<BaseRequest<TradingClearingRegistryNewRequest>> accountServiceExchanger;
@Autowired
public ClientCodeService(Consumer<String, Object> kafkaQueue,
Producer<String, Object> kafkaProducer,
ImdgProvider imdgProvider,
ValidationHelper validationHelper,
IMessageResolver messageResolver,
UserRoleVerification userRoleVerification,
@Qualifier("clientCodeNewRequestValidator") Function<ClientCodeNewRequest, IValidator> clientCodeNewRequestValidator,
@Qualifier("clientCodeUpdateRequestValidator") Function<ClientCodeUpdateRequest, IValidator> clientCodeUpdateRequestValidator,
@Qualifier("clientCodeDeleteRequestValidator") Function<CommonDeleteRequest, IValidator> clientCodeDeleteRequestValidator) {
super(kafkaQueue, kafkaProducer);
this.kafkaProducer = kafkaProducer;
this.idGenerator = imdgProvider.getImdgIdGenerator();
this.clientCodeMap = imdgProvider.getImdg(IMDGDistributedNames.Map_ClientCode, ClientCode.class);
this.companyMap = imdgProvider.getImdg(IMDGDistributedNames.Map_Company, Company.class);
this.tradingClearingRegistryMap = imdgProvider.getImdg(IMDGDistributedNames.Map_TradingClearingRegistry, TradingClearingRegistry.class);
this.clientCodeNewRequestValidator = clientCodeNewRequestValidator;
this.clientCodeUpdateRequestValidator = clientCodeUpdateRequestValidator;
this.clientCodeDeleteRequestValidator = clientCodeDeleteRequestValidator;
this.validationHelper = validationHelper;
this.messageResolver = messageResolver;
this.userRoleVerification = userRoleVerification;
this.accountServiceExchanger = new BiDirectionQueueExchanger<>(kafkaQueue, kafkaProducer,
Consts.DESTINATION_TRADING_CLEARING_REGISTRY_NEW,
Consts.DESTINATION_TRADING_CLEARING_REGISTRY_REPLY, // todo naming
60000
);
}
@Override
public void afterPropertiesSet() {
callback(ClientCodeNewRequest.class)
.setConsumer(this::clientCodeNew)
.forDestination(Consts.DESTINATION_CLIENT_CODE_NEW, callbacks::put);
callback(ClientCodeNewRequest.class)
.setConsumer(this::clientCodeNewFromApiUmCompany)
.forDestination(Consts.DESTINATION_CLIENT_CODE_NEW_UM_COMPANY, callbacks::put);
callback(ClientCodeUpdateRequest.class)
.setConsumer(this::clientCodeUpdate)
.forDestination(Consts.DESTINATION_CLIENT_CODE_UPDATE, callbacks::put);
callback(CommonDeleteRequest.class)
.setConsumer(this::clientCodeDelete)
.forDestination(Consts.DESTINATION_CLIENT_CODE_DELETE, callbacks::put);
init();
}
@Override
public void destroy() {
accountServiceExchanger.close();
}
protected RequestInfoUpdate clientCodeNew(BaseRequest<ClientCodeNewRequest> userRequest) {
log.debug("ClientCodeNewRequest received {}", userRequest.getId());
RequestInfoUpdate requestInfoUpdate = userRoleVerification.validateRoleAndGetResult(userRequest);
if (requestInfoUpdate != null) return requestInfoUpdate;
requestInfoUpdate = validationHelper.validateTillFirstError(userRequest, clientCodeNewRequestValidator);
if (requestInfoUpdate != null) return requestInfoUpdate;
ClientCodeNewRequest req = userRequest.getRequestPayload();
boolean doCreateTCR = checkNeedCreateTCR(req.getCompanyId(), req.getMoneyAccountId(), req.getDepoAccountId());
if (doCreateTCR) {
try {
createAndWaitTCR(userRequest.getId(), null,
req.getCompanyId(), req.getMoneyAccountId(), req.getDepoAccountId());
} catch (ClearingBaseException e) {
log.error("Can not wait creation of TCR. request id={};CompanyId={}, MoneyAccountId={}, DepoAccountId={}; {}",
userRequest.getId(), req.getCompanyId(), req.getMoneyAccountId(), req.getDepoAccountId(),
e.toString());
return makeErrorResponse(userRequest, CompanyErrors.GeneralError, "Can not create TCR at this moment");
}
}
if (req.getTradingClearingRegistryId() == null && req.getMoneyAccountId() != null) {
TradingClearingRegistry tradingClearingRegistry = selectTradingClearingRegistry(req.getCompanyId(), req.getMoneyAccountId(), req.getDepoAccountId());
if (tradingClearingRegistry == null) {
return makeErrorResponse(userRequest, CompanyErrors.TradingClearingRegistryNotFound);
} else {
req.setTradingClearingRegistryId(tradingClearingRegistry.getId());
}
}
ClientCode newClientCode = buildClientCode(req);
clientCodeMap.insert(newClientCode);
log.debug("successfully processed, new clientCode id {}", newClientCode.getId());
return null;
}
protected RequestInfoUpdate clientCodeNewFromApiUmCompany(BaseRequest<ClientCodeNewRequest> userRequest) {
log.debug("ClientCodeNewRequest received {}", userRequest.getId());
RequestInfoUpdate requestInfoUpdate = userRoleVerification.validateRoleAndGetResult(userRequest);
if (requestInfoUpdate != null) return requestInfoUpdate;
requestInfoUpdate = validationHelper.validateTillFirstError(userRequest, clientCodeNewRequestValidator);
if (requestInfoUpdate != null) return requestInfoUpdate;
ClientCodeNewRequest req = userRequest.getRequestPayload();
boolean doCreateTCR = checkNeedCreateTCR(req.getCompanyId(), req.getMoneyAccountId(), req.getDepoAccountId());
if (doCreateTCR) {
try {
createAndWaitTCR(userRequest.getId(), null,
req.getCompanyId(), req.getMoneyAccountId(), req.getDepoAccountId());
} catch (ClearingBaseException e) {
log.error("Can not wait creation of TCR. request id={};CompanyId={}, MoneyAccountId={}, DepoAccountId={}; {}",
userRequest.getId(), req.getCompanyId(), req.getMoneyAccountId(), req.getDepoAccountId(),
e.toString());
return makeErrorResponse(userRequest, CompanyErrors.GeneralError, "Can not create TCR at this moment");
}
}
if (req.getTradingClearingRegistryId() == null && req.getMoneyAccountId() != null) {
TradingClearingRegistry tradingClearingRegistry = selectTradingClearingRegistry(req.getCompanyId(), req.getMoneyAccountId(), req.getDepoAccountId());
if (tradingClearingRegistry == null) {
return makeErrorResponse(userRequest, CompanyErrors.TradingClearingRegistryNotFound);
} else {
req.setTradingClearingRegistryId(tradingClearingRegistry.getId());
}
}
ClientCode newClientCode = buildClientCode(req);
clientCodeMap.insert(newClientCode);
log.debug("successfully processed, new clientCode id {}", newClientCode.getId());
return null;
}
protected RequestInfoUpdate clientCodeUpdate(BaseRequest<ClientCodeUpdateRequest> userRequest) {
ClientCodeUpdateRequest req = userRequest.getRequestPayload();
RequestInfoUpdate requestInfoUpdate = userRoleVerification.validateRoleAndGetResult(userRequest);
if (requestInfoUpdate != null) return requestInfoUpdate;
requestInfoUpdate = validationHelper.validateTillFirstError(userRequest, clientCodeUpdateRequestValidator);
if (requestInfoUpdate != null) return requestInfoUpdate;
log.debug("ClientCodeUpdateRequest received, id={}", req.getId());
ClientCode clientCode = clientCodeMap.getSingleObjectByID(req.getId());
if (clientCode == null) {
return makeErrorResponse(userRequest, CompanyErrors.RecordNotFound, req.getId());
}
boolean doCreateTCR = checkNeedCreateTCR(req.getCompanyId(), req.getMoneyAccountId(), req.getDepoAccountId());
if (doCreateTCR) {
try {
createAndWaitTCR(userRequest.getId(), clientCode.getId(),
req.getCompanyId(), req.getMoneyAccountId(), req.getDepoAccountId());
} catch (ClearingBaseException e) {
log.error("Can not wait creation of TCR. request id={};CompanyId={}, MoneyAccountId={}, DepoAccountId={}; {}",
userRequest.getId(), req.getCompanyId(), req.getMoneyAccountId(), req.getDepoAccountId(),
e.toString());
return makeErrorResponse(userRequest, CompanyErrors.GeneralError, "Can not create TCR at this moment");
}
}
if (req.getTradingClearingRegistryId() == null && req.getMoneyAccountId() != null) {
TradingClearingRegistry tradingClearingRegistry = selectTradingClearingRegistry(req.getCompanyId(), req.getMoneyAccountId(), req.getDepoAccountId());
if (tradingClearingRegistry == null) {
return makeErrorResponse(userRequest, CompanyErrors.TradingClearingRegistryNotFound);
} else {
req.setTradingClearingRegistryId(tradingClearingRegistry.getId());
}
}
updateClientCode(clientCode, req);
clientCodeMap.update(clientCode);
log.debug("successfully processed update, id {}", clientCode.getId());
return null;
}
private RequestInfoUpdate makeErrorResponse(BaseRequest<?> req, CompanyErrors err, Object... arg) {
String errMsg = messageResolver.resolve(new EnumMessage(err, arg));
return new RequestInfoUpdate()
.setId(req.getId())
.setStatus(ru.spcex.clearing.platform.messaging.service.Status.Error)
.setMessage(errMsg);
}
protected RequestInfoUpdate clientCodeDelete(BaseRequest<CommonDeleteRequest> userRequest) {
log.debug("CommonDeleteRequest received id = {}", userRequest.getId());
CommonDeleteRequest req = userRequest.getRequestPayload();
RequestInfoUpdate requestInfoUpdate = userRoleVerification.validateRoleAndGetResult(userRequest);
if (requestInfoUpdate != null) return requestInfoUpdate;
requestInfoUpdate = validationHelper.validateTillFirstError(userRequest, clientCodeDeleteRequestValidator);
if (requestInfoUpdate != null) return requestInfoUpdate;
ClientCode clientCode = clientCodeMap.getSingleObjectByID(req.getId());
if (clientCode.getMoneyAccountId() != null) {
log.debug("Delete clientCode.id = {}: send message to account-service", clientCode.getId());
sendBlockTCR(clientCode.getTradingClearingRegistryId(), clientCode.getMoneyAccountId());
}
log.debug("Delete clientCode.id={}", clientCode.getId());
clientCodeMap.delete(clientCode);
return null;
}
TradingClearingRegistry selectTradingClearingRegistry(Long companyId, Long moneyAccountId, Long depoAccountId) {
Map<String, Comparable<?>> query = new HashMap<>();
query.put("companyId", companyId);
query.put("moneyAccountId", moneyAccountId);
query.put("tradingClearingRegistryType", TradingClearingRegistryType.Client_B.getKey());
if (depoAccountId != null) {
query.put("depoAccountId", depoAccountId);
}
TradingClearingRegistry result = tradingClearingRegistryMap.getSingleObjectByFieldValues(query);
log.trace("TradingClearingRegistry by: {}; {}found", query, result == null ? "not " : "");
return result;
}
private boolean checkNeedCreateTCR(Long companyId, Long moneyAccountId, Long depoAccountId) {
// moneyAccountId обязателен, depoAccountId опционален
if (moneyAccountId == null) {
return false;
}
TradingClearingRegistry result = selectTradingClearingRegistry(companyId, moneyAccountId, depoAccountId);
if (result == null) {
return true;
} else {
return false;
}
}
protected void createAndWaitTCR(Long reqId, Long clientCode,
Long companyId, Long moneyAccountId, Long depoAccountId) throws ClearingBaseException {
log.debug("For request {}, clientCode={} need create TCR: companyId={}, moneyAccountId={}, depoAccountId={}",
reqId, clientCode == null ? "new" : clientCode,
companyId, moneyAccountId, depoAccountId);
BaseRequest<TradingClearingRegistryNewRequest> request = new BaseRequest<>();
request.setId(idGenerator.nextId());
request.setActionType(ActionType.NEW);
TradingClearingRegistryNewRequest requestPayload = new TradingClearingRegistryNewRequest();
requestPayload.setCompanyId(companyId);
requestPayload.setMoneyAccountId(moneyAccountId);
requestPayload.setDepoAccountId(depoAccountId);
// requestPayload.setStatus(WorkflowStatus.Active.getKey());
requestPayload.setTradingClearingRegistryType(TradingClearingRegistryType.Client_B.getKey());
request.setRequestPayload(requestPayload);
try {
Long reply = accountServiceExchanger.exchange(request);
if (reply == null) {
throw new ClearingBaseException(CompanyErrors.GeneralError, "Waiting account-service timeout");
}
// примечание: ошибка и сбой (непредвиденное завершение программы) не приведёт к необратимым последствиям,
// т.к. пользователь сможет повторить запрос, а созданный на предыдущем запросе ТКР уже будет создан и найдётся.
} catch (InterruptedException e) {
throw new ClearingBaseException(CompanyErrors.GeneralError, "Waiting account-service timeout");
}
}
protected void sendBlockTCR(Long tradingClearingRegistryId, Long moneyAccountId) {
if (tradingClearingRegistryId == null) {
throw new IllegalArgumentException("tradingClearingRegistryId was null");
}
BaseRequest<TradingClearingRegistryUpdateRequest> request = new BaseRequest<>();
request.setId(idGenerator.nextId());
request.setActionType(ActionType.UPDATE);
TradingClearingRegistryUpdateRequest requestPayload = new TradingClearingRegistryUpdateRequest();
requestPayload.setId(tradingClearingRegistryId);
requestPayload.setStatus(WorkflowStatus.Blocked.getKey());
request.setRequestPayload(requestPayload);
try {
sendMessage(Consts.DESTINATION_TRADING_CLEARING_REGISTRY_UPDATE, request);
} catch (Exception e) {
log.error("Can not send block TCR message, {}", e.toString());
throw e;
}
}
/**
* @param queue Consts.*
* @param message BaseRequest
* @return
*/
private Long sendMessage(String queue, BaseRequest<?> message) {
Long sentRequestId = message.getId();
if (sentRequestId == null) {
throw new IllegalArgumentException("Message " + message.getClass().getSimpleName() + " required id.");
}
log.debug("Send to \"{}\" message {} id={}", queue, message.getActionType(), sentRequestId);
try {
Future<RecordMetadata> send = kafkaProducer.send(new ProducerRecord<>(queue, message));
send.get();
return sentRequestId;
} catch (InterruptedException e) {
log.error("Interrupt when send message, {}", ExceptionUtils.getStackTrace(e));
Thread.currentThread().interrupt();
throw new RuntimeException("Thread interrupted when send message to " + queue, e);
} catch (ExecutionException e) {
log.error("Error at send message, {} cause: {}", e.toString(), ExceptionUtils.getStackTrace(e.getCause()));
throw new RuntimeException("Send message error", e.getCause());
} catch (Exception e) {
log.error("Unexpected exception when send message: {}", ExceptionUtils.getStackTrace(e));
}
return null;
}
/**
* @param req требуется заполнить tradingClearingRegistryId по tradingClearingRegistry.code
* @return
*/
private ClientCode buildClientCode(ClientCodeNewRequest req) {
ClientCode clientCode = new ClientCode();
// clientCode.setId(idSequence.newId()); add in insert
clientCode.setCreated(Instant.now());
clientCode.setUpdated(clientCode.getCreated());
clientCode.setCompanyId(req.getCompanyId());
clientCode.setCode(req.getCode());
clientCode.setTradingClearingRegistryId(req.getTradingClearingRegistryId());
clientCode.setMoneyAccountId(req.getMoneyAccountId());
clientCode.setDepoAccountId(req.getDepoAccountId());
clientCode.setStatus(req.getStatus());
return clientCode;
}
private void updateClientCode(ClientCode clientCode, ClientCodeUpdateRequest req) {
assert clientCode.getId() != null && clientCode.getId().equals(req.getId());
clientCode.setCompanyId(req.getCompanyId());
clientCode.setCode(req.getCode());
clientCode.setTradingClearingRegistryId(req.getTradingClearingRegistryId());
clientCode.setMoneyAccountId(req.getMoneyAccountId());
clientCode.setDepoAccountId(req.getDepoAccountId());
clientCode.setStatus(req.getStatus());
clientCode.setUpdated(Instant.now());
}
}

2
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java
View file

@ -66,12 +66,14 @@ public interface Consts {
String DESTINATION_PROFILE_DOCUMENT_DELETE = "profile-document-delete";
String DESTINATION_CLIENT_CODE_NEW = "client-code-new";
String DESTINATION_CLIENT_CODE_NEW_UM_COMPANY = "client-code-new-from-api-um-company";
String DESTINATION_CLIENT_CODE_UPDATE = "client-code-update";
String DESTINATION_CLIENT_CODE_DELETE = "client-code-delete";
String DESTINATION_TRADING_CLEARING_REGISTRY_NEW = "trading-clearing-registry-new";
String DESTINATION_TRADING_CLEARING_REGISTRY_UPDATE = "trading-clearing-registry-update";
String DESTINATION_TRADING_CLEARING_REGISTRY_DELETE = "trading-clearing-registry-delete";
String DESTINATION_TRADING_CLEARING_REGISTRY_REPLY = "trading-clearing-registry";
String DESTINATION_SDF08_NEW = "s-df-08-new";
String DESTINATION_SDF02_NEW = "s-df-02-new";

10
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/registry/TradingClearingRegistryNewRequest.java
View file

@ -21,6 +21,8 @@ public class TradingClearingRegistryNewRequest {
private Long depoAccountId;
@JsonProperty
private String status;
@JsonProperty
private String tradingClearingRegistryType;
public Long getCompanyId() {
return companyId;
@ -50,6 +52,14 @@ public class TradingClearingRegistryNewRequest {
return status;
}
public String getTradingClearingRegistryType() {
return tradingClearingRegistryType;
}
public void setTradingClearingRegistryType(String tradingClearingRegistryType) {
this.tradingClearingRegistryType = tradingClearingRegistryType;
}
public void setStatus(String status) {
this.status = status;
}