company-service ProfileDocumentService - добавлена отправка в account-service kafka
This commit is contained in:
AKurakin
parent
e3fc42dda9
commit
2573b559c5
5 changed files with 170 additions and 15 deletions
|
|
@ -7,6 +7,7 @@ import ru.clearing.classes.statics.data.profile.ProfileDocument;
|
|||
import ru.clearing.platform.dictionary.DocumentTypeDictionary;
|
||||
import ru.spcex.clearing.company.error.CompanyErrors;
|
||||
import ru.spcex.clearing.imdg.IMDGDistributedNames;
|
||||
import ru.spcex.clearing.platform.messaging.domain.cud.common.CommonDeleteRequest;
|
||||
import ru.spcex.clearing.platform.messaging.domain.cud.company.ProfileDocumentNewRequest;
|
||||
import ru.spcex.clearing.platform.messaging.domain.cud.company.ProfileDocumentUpdateRequest;
|
||||
import ru.spcex.clearing.validation.common.rules.*;
|
||||
|
|
@ -130,4 +131,26 @@ public class ProfileDocumentValidationConfig {
|
|||
);
|
||||
};
|
||||
}
|
||||
|
||||
@Bean("profileDocumentDeleteRequestValidation")
|
||||
public Function<CommonDeleteRequest, IValidator> profileDocumentDeleteRequestValidator(
|
||||
Map<String, Imdg<? extends SpcexObjectBase>> imdgForValidation
|
||||
) {
|
||||
return profileDocumentDeleteRequest -> {
|
||||
ImdgValidationContext<CommonDeleteRequest> context = new ImdgValidationContext<>();
|
||||
context.setValidatedObject(profileDocumentDeleteRequest);
|
||||
Consumer<String> addImdg = (s) -> context.addImdg(s, imdgForValidation.get(s));
|
||||
addImdg.accept(IMDGDistributedNames.Map_Company);
|
||||
addImdg.accept(IMDGDistributedNames.Map_DocumentTypeDictionary);
|
||||
addImdg.accept(IMDGDistributedNames.Map_ProfileDocument);
|
||||
return new ValidatorImpl<>(context,
|
||||
IdPresentRule.instance("id",
|
||||
CommonDeleteRequest::getId,
|
||||
IMDGDistributedNames.Map_ProfileDocument,
|
||||
ProfileDocument.class,
|
||||
CompanyErrors.RequiredFieldEmpty,
|
||||
CompanyErrors.RecordNotFound)
|
||||
);
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -61,4 +61,24 @@ public class AccountNotificationHelper {
|
|||
r.setAccountId(account.getId());
|
||||
return r;
|
||||
}
|
||||
|
||||
|
||||
public CompanyErrors accountTerminationByDocumentNotification(Long companyId) {
|
||||
Collection<Account> accounts = accountMap.getCollectionObjectsBySQL("companyId=" + companyId);
|
||||
log.trace("Found {} accounts by company {}", accounts.size(), companyId);
|
||||
// for (Account account : accounts) {
|
||||
// Collection<LiabilitiesClaimsAssets> liabilities = liabilitiesClaimsAssetsMap.getCollectionObjectsBySQL("accountId=" + account.getId());
|
||||
// for (LiabilitiesClaimsAssets liability : liabilities) {
|
||||
// if (liability.getLiabilitiesQuantity() != null && BigDecimal.ZERO.compareTo(liability.getLiabilitiesQuantity()) == 0) {
|
||||
// return CompanyErrors.CompanyAlreadyHasLiabilities;
|
||||
// }
|
||||
// }
|
||||
// }
|
||||
log.debug("Sending messages to account-service for {} account", accounts.size());
|
||||
for (Account account : accounts) {
|
||||
Long reqId = kafkaSender.sendRequestToQueue(Consts.ACCOUNT_TERMINATION, createAccountsRequest(companyId, account));
|
||||
log.debug("For account[{}] send termination by document request id={}", account.getId(), reqId);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
|
@ -11,9 +11,12 @@ import org.springframework.lang.NonNull;
|
|||
import org.springframework.stereotype.Service;
|
||||
import ru.clearing.classes.statics.data.company.Company;
|
||||
import ru.clearing.classes.statics.data.profile.ProfileDocument;
|
||||
import ru.spcex.clearing.company.error.CompanyErrors;
|
||||
import ru.spcex.clearing.company.util.RequestHelper;
|
||||
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.common.CommonDeleteRequest;
|
||||
import ru.spcex.clearing.platform.messaging.domain.cud.company.ProfileDocumentNewRequest;
|
||||
import ru.spcex.clearing.platform.messaging.domain.cud.company.ProfileDocumentUpdateRequest;
|
||||
import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
|
||||
|
|
@ -21,10 +24,14 @@ 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.validation.common.ValidationHelper;
|
||||
import ru.spcex.platform.enumeration.DocumentTypes;
|
||||
import ru.spcex.platform.enumeration.WorkflowStatus;
|
||||
import ru.spcex.platform.imdg.api.Imdg;
|
||||
import ru.spcex.platform.imdg.api.ImdgProvider;
|
||||
import ru.spcex.platform.utils.enumeration.IMessageResolver;
|
||||
import ru.spcex.platform.utils.validation.IValidator;
|
||||
|
||||
import java.util.Objects;
|
||||
import java.util.function.Function;
|
||||
|
||||
@Service
|
||||
|
|
@ -33,29 +40,38 @@ public class ProfileDocumentService extends QueueConsumer implements Initializin
|
|||
private Imdg<ProfileDocument> profileDocumentMap;
|
||||
private Imdg<Company> companyMap;
|
||||
private ImdgProvider imdgProvider;
|
||||
private KafkaSender kafkaReqProducer;
|
||||
private AccountNotificationHelper accountNotificationHelper;
|
||||
|
||||
private final Function<ProfileDocumentNewRequest, IValidator> profileDocumentNewRequestValidator;
|
||||
private final Function<ProfileDocumentUpdateRequest, IValidator> profileDocumentUpdateRequestValidator;
|
||||
private final Function<CommonDeleteRequest, IValidator> profileDocumentDeleteRequestValidator;
|
||||
private final ValidationHelper validationHelper;
|
||||
private final UserRoleVerification userRoleVerification;
|
||||
protected IMessageResolver messageResolver;
|
||||
protected RequestHelper requestHelper;
|
||||
|
||||
@Autowired
|
||||
public ProfileDocumentService(Consumer<String, Object> kafkaQueue,
|
||||
Producer<String, Object> kafkaProducer,
|
||||
KafkaSender kafkaReqProducer,
|
||||
ImdgProvider imdgProvider,
|
||||
UserRoleVerification userRoleVerification,
|
||||
@Qualifier("profileDocumentNewRequestValidation") Function<ProfileDocumentNewRequest, IValidator> profileDocumentNewRequestValidator,
|
||||
@Qualifier("profileDocumentUpdateRequestValidation") Function<ProfileDocumentUpdateRequest, IValidator> profileDocumentUpdateRequestValidator,
|
||||
ValidationHelper validationHelper) {
|
||||
@Qualifier("profileDocumentDeleteRequestValidation") Function<CommonDeleteRequest, IValidator> profileDocumentDeleteRequestValidator,
|
||||
ValidationHelper validationHelper,
|
||||
IMessageResolver messageResolver,
|
||||
RequestHelper requestHelper,
|
||||
AccountNotificationHelper accountNotificationHelper) {
|
||||
super(kafkaQueue, kafkaProducer);
|
||||
this.imdgProvider = imdgProvider;
|
||||
this.kafkaReqProducer = kafkaReqProducer;
|
||||
this.accountNotificationHelper = accountNotificationHelper;
|
||||
this.profileDocumentNewRequestValidator = profileDocumentNewRequestValidator;
|
||||
this.profileDocumentUpdateRequestValidator = profileDocumentUpdateRequestValidator;
|
||||
this.profileDocumentDeleteRequestValidator = profileDocumentDeleteRequestValidator;
|
||||
this.validationHelper = validationHelper;
|
||||
this.userRoleVerification = userRoleVerification;
|
||||
this.messageResolver = messageResolver;
|
||||
this.requestHelper = requestHelper.setLogger(log);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -64,11 +80,14 @@ public class ProfileDocumentService extends QueueConsumer implements Initializin
|
|||
this.companyMap = imdgProvider.getImdg(IMDGDistributedNames.Map_Company, Company.class);
|
||||
|
||||
callback(ProfileDocumentNewRequest.class)
|
||||
.setConsumer(this::profileDocumentNew)
|
||||
.setFunction(this::profileDocumentNew)
|
||||
.forDestination(Consts.DESTINATION_PROFILE_DOCUMENT_NEW, callbacks::put);
|
||||
callback(ProfileDocumentUpdateRequest.class)
|
||||
.setConsumer(this::profileDocumentUpdate)
|
||||
.setFunction(this::profileDocumentUpdate)
|
||||
.forDestination(Consts.DESTINATION_PROFILE_DOCUMENT_UPDATE, callbacks::put);
|
||||
callback(CommonDeleteRequest.class)
|
||||
.setFunction(this::profileDocumentDelete)
|
||||
.forDestination(Consts.DESTINATION_PROFILE_DOCUMENT_DELETE, callbacks::put);
|
||||
init();
|
||||
}
|
||||
|
||||
|
|
@ -100,15 +119,45 @@ public class ProfileDocumentService extends QueueConsumer implements Initializin
|
|||
profileDocument.setLink(profileDocumentNewRequest.getLink());
|
||||
profileDocumentMap.insert(profileDocument);
|
||||
|
||||
// if (profileDocument.getDocumentType().equalsIgnoreCase(DocumentTypes.xcnt.getKey())) {
|
||||
// todo create request for update and check request for new account (something wrong)
|
||||
// kafkaReqProducer.sendRequestToQueue(Consts.ACCOUNT_UPDATE, )
|
||||
// }
|
||||
if (DocumentTypes.xcnt.getKey().equalsIgnoreCase(profileDocument.getDocumentType())) {
|
||||
processXCNT(profileDocument);
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
@NonNull
|
||||
private RequestInfoUpdate profileDocumentDelete(BaseRequest<CommonDeleteRequest> deleteRequest) {
|
||||
log.trace("Start processing CommonDeleteRequest!");
|
||||
|
||||
RequestInfoUpdate requestInfoUpdate = userRoleVerification.validateRoleAndGetResult(deleteRequest);
|
||||
if (requestInfoUpdate != null) return requestInfoUpdate;
|
||||
|
||||
requestInfoUpdate = validationHelper.validateTillFirstError(
|
||||
deleteRequest, profileDocumentDeleteRequestValidator
|
||||
);
|
||||
if (requestInfoUpdate != null) return requestInfoUpdate;
|
||||
|
||||
CommonDeleteRequest profileDocumentDeleteRequest = deleteRequest.getRequestPayload();
|
||||
ProfileDocument profileDocument = profileDocumentMap.getSingleObjectByID(profileDocumentDeleteRequest.getId());
|
||||
|
||||
Company existCompany = profileDocument.getCompanyId() == null ? null : companyMap.getSingleObjectByID(profileDocument.getCompanyId());
|
||||
if (existCompany == null) {
|
||||
log.warn("Company {} not found by profileDocument {}", profileDocument.getCompanyId(), profileDocument.getId());
|
||||
return requestHelper.makeErrorResponse(deleteRequest, CompanyErrors.CompanyNotFound, profileDocument.getCompanyId());
|
||||
}
|
||||
if (!WorkflowStatus.Active.equalsByKey(existCompany.getWorkflowStatus())) {
|
||||
log.debug("Company {} not active", profileDocument.getCompanyId());
|
||||
return requestHelper.makeErrorResponse(deleteRequest, CompanyErrors.CompanyNotFound, profileDocument.getCompanyId());
|
||||
}
|
||||
|
||||
profileDocumentMap.delete(profileDocument);
|
||||
|
||||
log.trace("successfully deleted, id {}", profileDocument.getId());
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
|
||||
private RequestInfoUpdate profileDocumentUpdate(BaseRequest<ProfileDocumentUpdateRequest> profileDocumentUpdateRequestBaseRequest) {
|
||||
log.trace("Start processing ProfileDocumentUpdateRequest!");
|
||||
|
||||
|
|
@ -124,6 +173,7 @@ public class ProfileDocumentService extends QueueConsumer implements Initializin
|
|||
ProfileDocumentUpdateRequest profileDocumentUpdateRequest = profileDocumentUpdateRequestBaseRequest.getRequestPayload();
|
||||
ProfileDocument profileDocument = profileDocumentMap.getSingleObjectByID(profileDocumentUpdateRequest.getId());
|
||||
|
||||
String prevDocType = profileDocument.getDocumentType();
|
||||
if (profileDocumentUpdateRequest.getCompanyId() != null)
|
||||
profileDocument.setCompanyId(profileDocumentUpdateRequest.getCompanyId());
|
||||
|
||||
|
|
@ -164,11 +214,34 @@ public class ProfileDocumentService extends QueueConsumer implements Initializin
|
|||
|
||||
log.trace("successfully processed, id {}", profileDocument.getId());
|
||||
|
||||
// if (profileDocument.getDocumentType().equalsIgnoreCase(DocumentTypes.xcnt.getKey())) {
|
||||
// todo if update profileDocument with documentType == XCNT... need send?
|
||||
// kafkaReqProducer.sendRequestToQueue(Consts.ACCOUNT_UPDATE, )
|
||||
// }
|
||||
if (!Objects.equals(prevDocType, profileDocument.getDocumentType()) &&
|
||||
DocumentTypes.xcnt.getKey().equalsIgnoreCase(profileDocument.getDocumentType())) {
|
||||
processXCNT(profileDocument);
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
protected void processXCNT(ProfileDocument profileDocument) {
|
||||
Company existCompany = profileDocument.getCompanyId() == null ? null : companyMap.getSingleObjectByID(profileDocument.getCompanyId());
|
||||
if (existCompany == null) {
|
||||
log.warn("Company {} not found by profileDocument {}", profileDocument.getCompanyId(), profileDocument.getId());
|
||||
return;
|
||||
}
|
||||
if (!WorkflowStatus.Active.equalsByKey(existCompany.getWorkflowStatus())) {
|
||||
log.warn("Current workflowStatus {} for company {}. Do not termination account.",
|
||||
existCompany.getWorkflowStatus(), existCompany.getId());
|
||||
}
|
||||
|
||||
CompanyErrors hasError = accountNotificationHelper.accountTerminationByDocumentNotification(profileDocument.getCompanyId());
|
||||
if (hasError != null) {
|
||||
log.error("Error at termination account of company {} by document {}: {}",
|
||||
profileDocument.getCompanyId(), profileDocument.getId(), hasError + "(" + hasError.getId() + ")");
|
||||
}
|
||||
/* todo processXCNT
|
||||
Дальнейшая обработка документа о расторжении, то есть обновление статусов в company и relation, возможна только после ответа об успешном обновлении статуса в account.
|
||||
1.2. Обновить статус в бизнес-объекте company [profileDocument.companyId=company.id] согласно описанию с тегом "При расторжении".
|
||||
1.3. Обновить статус в бизнес-объекте relation [profileDocument.companyId=relation.consumerId] согласно описанию с тегом "При расторжении".
|
||||
*/
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ package ru.spcex.clearing.company.service;
|
|||
import org.apache.kafka.clients.consumer.MockConsumer;
|
||||
import org.apache.kafka.clients.producer.MockProducer;
|
||||
import org.apache.kafka.clients.producer.ProducerRecord;
|
||||
import org.junit.jupiter.api.Assertions;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.extension.ExtendWith;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
|
|
@ -21,6 +22,7 @@ import ru.spcex.clearing.company.config.validation.ValidationConfig;
|
|||
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.common.CommonDeleteRequest;
|
||||
import ru.spcex.clearing.platform.messaging.domain.cud.company.ProfileDocumentNewRequest;
|
||||
import ru.spcex.clearing.platform.messaging.domain.cud.company.ProfileDocumentUpdateRequest;
|
||||
import ru.spcex.clearing.test.MatcherFactory;
|
||||
|
|
@ -47,6 +49,7 @@ import static ru.spcex.clearing.test.config.ImdgTestConfig.currentID;
|
|||
BeanConfiguration.class,
|
||||
ValidationConfig.class,
|
||||
ProfileDocumentService.class,
|
||||
AccountNotificationHelper.class,
|
||||
ProfileDocumentValidationConfig.class,
|
||||
KafkaTestConfig.class,
|
||||
ImdgTestConfig.class})
|
||||
|
|
@ -56,6 +59,7 @@ class ProfileDocumentServiceTest {
|
|||
private static final int PARTITION = 0;
|
||||
private static final String TOPIC_DESTINATION_PROFILE_DOCUMENT_NEW = Consts.DESTINATION_PROFILE_DOCUMENT_NEW;
|
||||
private static final String TOPIC_DESTINATION_PROFILE_DOCUMENT_UPDATE = Consts.DESTINATION_PROFILE_DOCUMENT_UPDATE;
|
||||
private static final String TOPIC_DESTINATION_PROFILE_DOCUMENT_DELETE = Consts.DESTINATION_PROFILE_DOCUMENT_DELETE;
|
||||
|
||||
private static final Long ID = currentID.getAndIncrement();
|
||||
@Autowired
|
||||
|
|
@ -220,4 +224,38 @@ class ProfileDocumentServiceTest {
|
|||
// predictableProfileDocument.setId(resultUpdate.getId());
|
||||
PROFILE_DOCUMENT_MATCHER.assertMatch(resultUpdate, predictableProfileDocument);
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
void deleteProfileDocument() {
|
||||
//ARRANGE
|
||||
ProfileDocument existsProfileDocument = new ProfileDocument();
|
||||
existsProfileDocument.setId(PROFILE_DOCUMENT_ID);
|
||||
existsProfileDocument.setCompanyId(COMPANY_ID);
|
||||
existsProfileDocument.setDocumentType(TEST_DOCUMENT_TYPE_ID);
|
||||
existsProfileDocument.setIssueDate(LocalDate.now());
|
||||
existsProfileDocument.setIssuePlace("place");
|
||||
existsProfileDocument.setIssuer("issuer");
|
||||
existsProfileDocument.setIssuerCode("issuerCode");
|
||||
existsProfileDocument.setName("name");
|
||||
existsProfileDocument.setNumber("number");
|
||||
existsProfileDocument.setPlace("place");
|
||||
existsProfileDocument.setValidFromDate(LocalDate.now());
|
||||
existsProfileDocument.setValidToDate(LocalDate.now().plus(1, ChronoUnit.DAYS));
|
||||
existsProfileDocument.setLink("link");
|
||||
profileDocumentMap.insert(existsProfileDocument);
|
||||
|
||||
CommonDeleteRequest profileDocumentDeleteRequest = new CommonDeleteRequest();
|
||||
profileDocumentDeleteRequest.setId(PROFILE_DOCUMENT_ID);
|
||||
|
||||
//ACT
|
||||
String jsonString = getJsonStringForUPDATE(profileDocumentDeleteRequest, ID);
|
||||
|
||||
addRecordToKafka((MockConsumer) profileDocumentService.getConsumer(), TOPIC_DESTINATION_PROFILE_DOCUMENT_DELETE, PARTITION, 0, jsonString);
|
||||
|
||||
//ASSERT
|
||||
waitingWhenAddedRecordAndCheckIt(ID, mockProducer, producerRecord);
|
||||
ProfileDocument resultUpdate = profileDocumentMap.getSingleObjectByID(PROFILE_DOCUMENT_ID);
|
||||
Assertions.assertNull(resultUpdate);
|
||||
}
|
||||
}
|
||||
|
|
@ -60,6 +60,7 @@ public interface Consts {
|
|||
String DESTINATION_RELATION_UPDATE = "relation-update";
|
||||
String DESTINATION_PROFILE_DOCUMENT_NEW = "profile-document-new";
|
||||
String DESTINATION_PROFILE_DOCUMENT_UPDATE = "profile-document-update";
|
||||
String DESTINATION_PROFILE_DOCUMENT_DELETE = "profile-document-delete";
|
||||
|
||||
String DESTINATION_SDF08_NEW = "s-df-08-new";
|
||||
String DESTINATION_SDF02_NEW = "s-df-02-new";
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue