diff --git a/clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/config/KafkaConfig.java b/clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/config/KafkaConfig.java index f1cac72b8..58a707200 100644 --- a/clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/config/KafkaConfig.java +++ b/clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/config/KafkaConfig.java @@ -8,6 +8,7 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Scope; import ru.spcex.clearing.account.config.settings.AccountServiceSettings; +import ru.spcex.clearing.imdg.IMDGDistributedNames; import ru.spcex.clearing.platform.messaging.config.KafkaConsumerFactory; import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory; import ru.spcex.clearing.platform.messaging.service.RequestInfo; @@ -41,7 +42,7 @@ public class KafkaConfig { .producer(kafkaProducer) .idGenerator(imdgIdGenerator::nextId) .imdgProvider(s -> { - Imdg imdg = imdgProvider.getImdg(s, RequestInfo.class); + Imdg imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class); return imdg::insert; }) .build(); diff --git a/clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/service/BankAccountService.java b/clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/service/BankAccountService.java index b95313fd0..1293f4d9d 100644 --- a/clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/service/BankAccountService.java +++ b/clearing-parent/account-service/src/main/java/ru/spcex/clearing/account/service/BankAccountService.java @@ -1,6 +1,7 @@ package ru.spcex.clearing.account.service; import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.producer.Producer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.InitializingBean; @@ -23,8 +24,8 @@ public class BankAccountService extends QueueConsumer implements InitializingBea private final Imdg bankAccountMap; @Autowired - public BankAccountService(Consumer kafkaQueue, ImdgProvider imdgProvider) { - super(kafkaQueue); + public BankAccountService(Consumer kafkaQueue, Producer kafkaProducer, ImdgProvider imdgProvider) { + super(kafkaQueue, kafkaProducer); this.bankAccountMap = imdgProvider.getImdg(IMDGDistributedNames.Map_BankAccount, BankAccount.class); } diff --git a/clearing-parent/backend-api/pom.xml b/clearing-parent/backend-api/pom.xml index 18b29f7d1..c87effd50 100644 --- a/clearing-parent/backend-api/pom.xml +++ b/clearing-parent/backend-api/pom.xml @@ -53,6 +53,10 @@ ru.spcex.clearing classes + + ru.spcex.clearing + dictionary + @@ -94,6 +98,37 @@ ${project.artifactId} + + + org.codehaus.mojo + xml-maven-plugin + + + + transform + + + + + + + src/data/meta + + src/data/meta/meta.server.html + src/data/meta/meta.server.json + src/data/meta/meta.server.xslt + + src/data/meta/meta.server.xslt + + + .json + + + + + + \ No newline at end of file diff --git a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/config/KafkaConfig.java b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/config/KafkaConfig.java index f64045cb9..9046c9bc8 100644 --- a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/config/KafkaConfig.java +++ b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/config/KafkaConfig.java @@ -1,5 +1,8 @@ package ru.spcex.clearing.backendapi.config; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.MockConsumer; +import org.apache.kafka.clients.consumer.OffsetResetStrategy; import org.apache.kafka.clients.producer.MockProducer; import org.apache.kafka.clients.producer.Producer; import org.springframework.beans.factory.annotation.Autowired; @@ -7,6 +10,7 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Profile; import ru.spcex.clearing.backendapi.config.element.BackendApiSettings; +import ru.spcex.clearing.platform.messaging.config.KafkaConsumerFactory; import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory; @Configuration @@ -16,7 +20,7 @@ public class KafkaConfig { @Autowired @Bean public Producer createProducer(BackendApiSettings settings) { - return KafkaProducerFactory.producer(settings.getKafka()); + return KafkaProducerFactory.producer(settings.getKafkaProducer()); } @Profile("kafkaDisabled") @@ -25,4 +29,17 @@ public class KafkaConfig { return new MockProducer<>(true, (topic, data) -> data != null ? data.getBytes() : new byte[0], (topic, data) -> data.toString().getBytes()); } + @Profile("!kafkaDisabled") + @Autowired + @Bean + public Consumer createConsumer(BackendApiSettings settings) { + return KafkaConsumerFactory.consumer(settings.getKafkaConsumer()); + } + + @Profile("kafkaDisabled") + @Bean + public Consumer createConsumerMock() { + return new MockConsumer<>(OffsetResetStrategy.NONE); + } + } diff --git a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/config/element/BackendApiSettings.java b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/config/element/BackendApiSettings.java index 422d18323..79ba99f2d 100644 --- a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/config/element/BackendApiSettings.java +++ b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/config/element/BackendApiSettings.java @@ -4,6 +4,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.PropertySource; import org.springframework.stereotype.Component; import ru.spcex.clearing.backendapi.security.element.SecuritySettings; +import ru.spcex.clearing.platform.messaging.config.element.KafkaConsumerSettings; import ru.spcex.clearing.platform.messaging.config.element.KafkaProducerSettings; import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams; @@ -12,7 +13,8 @@ import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams; @ConfigurationProperties("backend-api") public class BackendApiSettings { private HazelcastClientParams hazelcast; - private KafkaProducerSettings kafka; + private KafkaProducerSettings kafkaProducer; + private KafkaConsumerSettings kafkaConsumer; private SecuritySettings security; private String exampleSetting; @@ -24,12 +26,20 @@ public class BackendApiSettings { this.hazelcast = hazelcast; } - public KafkaProducerSettings getKafka() { - return kafka; + public KafkaProducerSettings getKafkaProducer() { + return kafkaProducer; } - public void setKafka(KafkaProducerSettings kafka) { - this.kafka = kafka; + public void setKafkaProducer(KafkaProducerSettings kafkaProducer) { + this.kafkaProducer = kafkaProducer; + } + + public KafkaConsumerSettings getKafkaConsumer() { + return kafkaConsumer; + } + + public void setKafkaConsumer(KafkaConsumerSettings kafkaConsumer) { + this.kafkaConsumer = kafkaConsumer; } public String getExampleSetting() { diff --git a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/queue/QueueExceptionHandler.java b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/queue/QueueExceptionHandler.java index 5b6503559..70dafb130 100644 --- a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/queue/QueueExceptionHandler.java +++ b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/queue/QueueExceptionHandler.java @@ -39,7 +39,7 @@ public class QueueExceptionHandler extends ResponseEntityExceptionHandler { @ResponseStatus(HttpStatus.BAD_REQUEST) @ResponseBody public BasicSpcexResponse handleValidationException(ActionValidationException e) { - log.error("Default scoring exception handler: {}", ExceptionUtils.getStackTrace(e)); + log.error("Default exception handler: {}", ExceptionUtils.getStackTrace(e)); EnumMessage firstError = e.getErrors().iterator().next(); BasicSpcexResponse errorResponse = new BasicSpcexResponse(); errorResponse.setCode(firstError.getSubject().getId()); @@ -68,7 +68,7 @@ public class QueueExceptionHandler extends ResponseEntityExceptionHandler { @ResponseStatus(HttpStatus.INTERNAL_SERVER_ERROR) @ResponseBody public BasicSpcexResponse handleDefault(Throwable e) { - log.error("Default scoring exception handler: {}", ExceptionUtils.getStackTrace(e)); + log.error("Default exception handler: {}", ExceptionUtils.getStackTrace(e)); BasicSpcexResponse errorResponse = new BasicSpcexResponse(); errorResponse.setCode(-1); errorResponse.setMessage("internal server error"); diff --git a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/response/entity/dictionary/DictionaryBackendGetAll.java b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/response/entity/dictionary/DictionaryBackendGetAll.java new file mode 100644 index 000000000..3d6c11aeb --- /dev/null +++ b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/response/entity/dictionary/DictionaryBackendGetAll.java @@ -0,0 +1,49 @@ +package ru.spcex.clearing.backendapi.controller.response.entity.dictionary; + +import com.fasterxml.jackson.annotation.JsonProperty; +import io.swagger.annotations.ApiModel; +import io.swagger.annotations.ApiModelProperty; +import ru.clearing.platform.dictionary.AbstractDictionary; +import ru.spcex.clearing.backendapi.controller.response.BasicSpcexResponse; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; + +@ApiModel(description = "Ответ при получении объектов словаря.") +public class DictionaryBackendGetAll extends BasicSpcexResponse { + + @JsonProperty + @ApiModelProperty(value = "Полезная нагрузка") + private DictionaryBackendPayload payload = new DictionaryBackendPayload(); + + public void fromEntity(Collection dictionaryValues) { + var payload = this.getPayload(); + for (AbstractDictionary dictionaryValue : dictionaryValues) { + var singleItem = new DictionaryBackendGetFields(); + singleItem.fromEntity(dictionaryValue); + payload.getItems().add(singleItem); + } + } + + + private static class DictionaryBackendPayload { + private List items = new ArrayList<>(); + + public List getItems() { + return items; + } + + public void setItems(List items) { + this.items = items; + } + } + + public DictionaryBackendPayload getPayload() { + return payload; + } + + public void setPayload(DictionaryBackendPayload payload) { + this.payload = payload; + } +} \ No newline at end of file diff --git a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/response/entity/dictionary/DictionaryBackendGetFields.java b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/response/entity/dictionary/DictionaryBackendGetFields.java new file mode 100644 index 000000000..fa27898c9 --- /dev/null +++ b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/response/entity/dictionary/DictionaryBackendGetFields.java @@ -0,0 +1,48 @@ +package ru.spcex.clearing.backendapi.controller.response.entity.dictionary; + +import com.fasterxml.jackson.annotation.JsonProperty; +import io.swagger.annotations.ApiModelProperty; +import ru.clearing.platform.dictionary.AbstractDictionary; + +public class DictionaryBackendGetFields { + @JsonProperty + @ApiModelProperty(value = "Идентификатор", example = "1234") + private Long id; + @JsonProperty + @ApiModelProperty(value = "", example = "") + private String code; + @JsonProperty + @ApiModelProperty(value = "Имя банка", example = "Сбербанк") + private String name; + + public void fromEntity(AbstractDictionary dictionaryValue) { + this.id = dictionaryValue.getId(); + this.code = dictionaryValue.getCode(); + this.name = dictionaryValue.getName(); + } + + + public Long getId() { + return id; + } + + public void setId(Long id) { + this.id = id; + } + + public String getCode() { + return code; + } + + public void setCode(String code) { + this.code = code; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } +} diff --git a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/response/entity/dictionary/DictionaryBackendGetSingleValue.java b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/response/entity/dictionary/DictionaryBackendGetSingleValue.java new file mode 100644 index 000000000..b58ccd807 --- /dev/null +++ b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/response/entity/dictionary/DictionaryBackendGetSingleValue.java @@ -0,0 +1,30 @@ +package ru.spcex.clearing.backendapi.controller.response.entity.dictionary; + +import com.fasterxml.jackson.annotation.JsonProperty; +import io.swagger.annotations.ApiModel; +import io.swagger.annotations.ApiModelProperty; +import ru.clearing.platform.dictionary.AbstractDictionary; +import ru.spcex.clearing.backendapi.controller.response.BasicSpcexResponse; + +@ApiModel(description = "Ответ при получении объекта BankAccount.") +public class DictionaryBackendGetSingleValue extends BasicSpcexResponse { + + @JsonProperty + @ApiModelProperty(value = "Поля объекта") + private DictionaryBackendGetFields payload; + + public DictionaryBackendGetFields getPayload() { + return payload; + } + + public void fromEntity(AbstractDictionary dictionaryValue) { + var payload = new DictionaryBackendGetFields(); + this.setPayload(payload); + payload.fromEntity(dictionaryValue); + } + + + public void setPayload(DictionaryBackendGetFields payload) { + this.payload = payload; + } +} \ No newline at end of file diff --git a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/response/system/RequestInfoResponse.java b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/response/system/RequestInfoResponse.java new file mode 100644 index 000000000..ed21ea44d --- /dev/null +++ b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/response/system/RequestInfoResponse.java @@ -0,0 +1,51 @@ +package ru.spcex.clearing.backendapi.controller.response.system; + +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.databind.annotation.JsonSerialize; +import io.swagger.annotations.ApiModel; +import io.swagger.annotations.ApiModelProperty; +import ru.spcex.clearing.backendapi.controller.response.BasicSpcexResponse; +import ru.spcex.clearing.platform.messaging.domain.json.serialize.EnumSerializer; +import ru.spcex.clearing.platform.messaging.service.Status; + + +/** + * Банковские реквизиты для перечисления денежных средств + * @ApiModel(description = "Ответ в результате отправки операции в топик Kafka.") + * @ApiModelProperty(value = "Информация о принятом запросе") + **/ +@ApiModel(description = "Ответ при получении статуса запроса.") +public class RequestInfoResponse extends BasicSpcexResponse { + + @JsonProperty + @ApiModelProperty(value = "Поля объекта") + private RequestInfoPayload payload; + + public RequestInfoPayload getPayload() { + return payload; + } + + public void setStatus(Status status) { + this.payload = new RequestInfoPayload(); + payload.setStatus(status); + } + + public void setPayload(RequestInfoPayload payload) { + this.payload = payload; + } + + private static class RequestInfoPayload { + @JsonProperty + @JsonSerialize(using = EnumSerializer.class) + private Status status; + + public Status getStatus() { + return status; + } + + public void setStatus(Status status) { + this.status = status; + } + } + +} \ No newline at end of file diff --git a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/system/DictionaryController.java b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/system/DictionaryController.java new file mode 100644 index 000000000..425bd1876 --- /dev/null +++ b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/system/DictionaryController.java @@ -0,0 +1,139 @@ +package ru.spcex.clearing.backendapi.controller.system; + +import io.swagger.annotations.ApiOperation; +import io.swagger.annotations.ApiParam; +import io.swagger.annotations.ApiResponse; +import io.swagger.annotations.ApiResponses; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.http.HttpStatus; +import org.springframework.stereotype.Controller; +import org.springframework.web.bind.annotation.*; +import ru.clearing.platform.dictionary.AbstractDictionary; +import ru.spcex.clearing.backendapi.controller.response.BasicSpcexResponse; +import ru.spcex.clearing.backendapi.controller.response.entity.dictionary.DictionaryBackendGetAll; +import ru.spcex.clearing.backendapi.controller.response.entity.dictionary.DictionaryBackendGetSingleValue; +import ru.spcex.clearing.backendapi.errors.ActionValidationException; +import ru.spcex.clearing.backendapi.errors.BackEndError; +import ru.spcex.clearing.backendapi.errors.NotFound404Exception; +import ru.spcex.platform.classes.base.SpcexObjectBase; +import ru.spcex.platform.imdg.api.Imdg; +import ru.spcex.platform.imdg.api.ImdgProvider; +import ru.spcex.platform.utils.enumeration.EnumMessage; +import ru.spcex.platform.utils.enumeration.IMessageResolver; + +import java.util.Collection; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +@Controller +@RequestMapping("/dictionaries") +public class DictionaryController { + private final Logger log = LoggerFactory.getLogger(getClass()); + private final Map> allDictionaryMaps; + private final ImdgProvider imdgProvider; + + @Autowired + public DictionaryController(ImdgProvider imdgProvider) { + this.imdgProvider = imdgProvider; + this.allDictionaryMaps = new ConcurrentHashMap<>(); + } + + @ApiOperation(value = "get all dictionary value.") + @ApiResponses(value = {@ApiResponse(code = 200, message = "OK", response = DictionaryBackendGetAll.class)}) + @RequestMapping(value = "/{dictionary-name}", method = RequestMethod.GET) + @ResponseBody + public DictionaryBackendGetAll getAll(@ApiParam(value = "Название словаря", required = true, example = "moneyFlowSide") + @PathVariable("dictionary-name") String dictionaryName) { + Imdg dictionary = extractDictionaryImdgFromUrlParameter(dictionaryName, AbstractDictionary.class); + Collection allValues = dictionary.getAllValues(); + DictionaryBackendGetAll response = new DictionaryBackendGetAll(); + response.fromEntity(allValues); + return response; + } + + @ApiOperation(value = "get dictionary value by id.") + @ApiResponses(value = {@ApiResponse(code = 200, message = "OK", response = DictionaryBackendGetSingleValue.class), @ApiResponse(code = 400, message = "Ошибка валидации", response = BasicSpcexResponse.class)}) + @RequestMapping(value = "/{dictionary-name}/{id}", method = RequestMethod.GET) + @ResponseBody + public DictionaryBackendGetSingleValue getById(@ApiParam(value = "Название словаря", required = true, example = "moneyFlowSide") + @PathVariable("dictionary-name") String dictionaryName, + @ApiParam(value = "Идентификатор объекта", required = true, example = "1234") + @PathVariable("id") Long id) { + Imdg dictionary = extractDictionaryImdgFromUrlParameter(dictionaryName, AbstractDictionary.class); + AbstractDictionary value = dictionary.getSingleObjectByID(id); + if (value == null) throw new NotFound404Exception(dictionaryName + " id='" + id + "'"); + DictionaryBackendGetSingleValue response = new DictionaryBackendGetSingleValue(); + response.fromEntity(value); + return response; + } + + @ApiOperation(value = "get dictionary value by code.") + @ApiResponses(value = {@ApiResponse(code = 200, message = "OK", response = DictionaryBackendGetSingleValue.class), @ApiResponse(code = 400, message = "Ошибка валидации", response = BasicSpcexResponse.class)}) + @RequestMapping(value = "/{dictionary-name}/code/{code}", method = RequestMethod.GET) + @ResponseBody + public DictionaryBackendGetSingleValue getByCode(@ApiParam(value = "Название словаря", required = true, example = "moneyFlowSide") + @PathVariable("dictionary-name") String dictionaryName, + @ApiParam(value = "Код словаря", required = true, example = "1234") + @PathVariable("code") String code) { + Imdg dictionary = extractDictionaryImdgFromUrlParameter(dictionaryName, AbstractDictionary.class); + AbstractDictionary value = dictionary.getSingleObjectByFieldValues(Map.of("code", code)); + if (value == null) throw new NotFound404Exception(dictionaryName + " code='" + code + "'"); + DictionaryBackendGetSingleValue response = new DictionaryBackendGetSingleValue(); + response.fromEntity(value); + return response; + } + + @SuppressWarnings("unchecked") + private Imdg extractDictionaryImdgFromUrlParameter(String dictionaryNameFromUrl, Class clazz) { + if (dictionaryNameFromUrl == null || dictionaryNameFromUrl.length() < 1) { + throw new ActionValidationException(BackEndError.ValidationError, "dictionary-name"); + } + + String actualDictionaryName = "Map_" + + dictionaryNameFromUrl.substring(0, 1).toUpperCase() + + dictionaryNameFromUrl.substring(1) + + "Dictionary"; + + Imdg dictionary = (Imdg) allDictionaryMaps.computeIfAbsent(actualDictionaryName, (mapName1) -> { + Imdg potentialImdg = imdgProvider.getImdg(actualDictionaryName, clazz); + if (potentialImdg.size() == 0) { + log.warn("coudln't find dictionary {}", actualDictionaryName); + return null; + } + return potentialImdg; + }); + + if (dictionary == null) { + throw new ActionValidationException(BackEndError.DictionaryNotFound, dictionaryNameFromUrl); + } + + return dictionary; + } + + @Autowired + private IMessageResolver errorResolver; + + @ResponseStatus(value = HttpStatus.BAD_REQUEST) + @ResponseBody + @ExceptionHandler(ActionValidationException.class) + public BasicSpcexResponse conflict(ActionValidationException ex) { + BasicSpcexResponse response = new BasicSpcexResponse(); + EnumMessage message = ex.getErrors().stream().findFirst().orElseThrow(); + response.setCode(message.getSubject().getId()); + response.setMessage(errorResolver.resolve(message)); + return response; + } + + @ResponseStatus(value = HttpStatus.NOT_FOUND) + @ResponseBody + @ExceptionHandler(NotFound404Exception.class) + public BasicSpcexResponse conflict(NotFound404Exception ex) { + BasicSpcexResponse response = new BasicSpcexResponse(); + EnumMessage message = ex.getError(); + response.setCode(message.getSubject().getId()); + response.setMessage(errorResolver.resolve(message)); + return response; + } +} diff --git a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/system/RequestStatusController.java b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/system/RequestStatusController.java new file mode 100644 index 000000000..cebfc2d57 --- /dev/null +++ b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/system/RequestStatusController.java @@ -0,0 +1,48 @@ +package ru.spcex.clearing.backendapi.controller.system; + +import io.swagger.annotations.ApiOperation; +import io.swagger.annotations.ApiParam; +import io.swagger.annotations.ApiResponse; +import io.swagger.annotations.ApiResponses; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Controller; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestMethod; +import org.springframework.web.bind.annotation.ResponseBody; +import ru.spcex.clearing.backendapi.controller.response.BasicSpcexResponse; +import ru.spcex.clearing.backendapi.controller.response.system.RequestInfoResponse; +import ru.spcex.clearing.imdg.IMDGDistributedNames; +import ru.spcex.clearing.platform.messaging.service.RequestInfo; +import ru.spcex.platform.imdg.api.Imdg; +import ru.spcex.platform.imdg.api.ImdgProvider; + +@Controller +@RequestMapping("/requests") +public class RequestStatusController { + + private final Imdg imdg; + + @Autowired + public RequestStatusController(ImdgProvider imdgProvider) { + this.imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class); + } + + @ApiOperation(value = "get request status.") + @ApiResponses(value = {@ApiResponse(code = 200, message = "OK", response = RequestInfoResponse.class), @ApiResponse(code = 400, message = "Ошибка валидации", response = BasicSpcexResponse.class)}) + @RequestMapping(value = "/{id}", method = RequestMethod.GET) + @ResponseBody + public RequestInfoResponse getById(@ApiParam(value = "Идентификатор объекта", required = true, example = "1234") + @PathVariable("id") Long id) { + RequestInfo singleObjectByID; + if (id == null || (singleObjectByID = imdg.getSingleObjectByID(id)) == null) { + RequestInfoResponse response = new RequestInfoResponse(); + response.setCode(-1L); + response.setMessage("cannot find request with id='" + id + "'."); + return response; + } + RequestInfoResponse response = new RequestInfoResponse(); + response.setStatus(singleObjectByID.getStatus()); + return response; + } +} diff --git a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/errors/ActionValidationException.java b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/errors/ActionValidationException.java index 251695991..b117f4c11 100644 --- a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/errors/ActionValidationException.java +++ b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/errors/ActionValidationException.java @@ -1,6 +1,7 @@ package ru.spcex.clearing.backendapi.errors; import ru.spcex.platform.utils.enumeration.EnumMessage; +import ru.spcex.platform.utils.enumeration.IEnumId; import java.util.Collection; import java.util.List; @@ -16,6 +17,10 @@ public class ActionValidationException extends RuntimeException { this.errors = List.of(error); } + public ActionValidationException(IEnumId subject, Object... args) { + this.errors = List.of(new EnumMessage(subject, args)); + } + public Collection getErrors() { return errors; } diff --git a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/errors/BackEndError.java b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/errors/BackEndError.java index 5808d3e7e..e84aeba69 100644 --- a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/errors/BackEndError.java +++ b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/errors/BackEndError.java @@ -6,7 +6,10 @@ public enum BackEndError implements IEnumId { ValidationError(3000L), UnknownJsonProperty(3001L), FailedToReadHttpMessage(3002L), - KeycloakRepeatedRoles(3003L); + KeycloakRepeatedRoles(3003L), + DictionaryNotFound(3004L), + ResourceNotFound(404L) + ; private final Long id; BackEndError(Long id) { diff --git a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/errors/NotFound404Exception.java b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/errors/NotFound404Exception.java new file mode 100644 index 000000000..1483749e5 --- /dev/null +++ b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/errors/NotFound404Exception.java @@ -0,0 +1,16 @@ +package ru.spcex.clearing.backendapi.errors; + +import ru.spcex.platform.utils.enumeration.EnumMessage; + +public class NotFound404Exception extends RuntimeException { + + private final EnumMessage error; + + public NotFound404Exception(String comment) { + this.error = new EnumMessage(BackEndError.ResourceNotFound, comment); + } + + public EnumMessage getError() { + return error; + } +} diff --git a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/service/RequestInfoAccepter.java b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/service/RequestInfoAccepter.java new file mode 100644 index 000000000..a1eaff115 --- /dev/null +++ b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/service/RequestInfoAccepter.java @@ -0,0 +1,48 @@ +package ru.spcex.clearing.backendapi.service; + +import org.apache.kafka.clients.consumer.Consumer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.stereotype.Service; +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.service.QueueConsumer; +import ru.spcex.clearing.platform.messaging.service.RequestInfo; +import ru.spcex.clearing.platform.messaging.service.RequestInfoUpdate; +import ru.spcex.platform.imdg.api.Imdg; +import ru.spcex.platform.imdg.api.ImdgProvider; + +@Service +public class RequestInfoAccepter extends QueueConsumer implements InitializingBean { + private final Logger log = LoggerFactory.getLogger(getClass()); + private final Imdg requestInfoImdg; + + + public RequestInfoAccepter(Consumer kafkaQueue, ImdgProvider imdgProvider) { + super(kafkaQueue); + this.requestInfoImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class); + } + + + @Override + public void afterPropertiesSet() throws Exception { + callback(RequestInfoUpdate.class) + .setConsumer(this::updateRequestInfo) + .forDestination(Consts.REQUEST_INFO_UPDATE, callbacks::put); + init(); + } + + private void updateRequestInfo(BaseRequest requestInfoUpdateBaseRequest) { + //will throw exception for any class other than RequestInfoUpdate + RequestInfoUpdate statusInfo = requestInfoUpdateBaseRequest.getRequestPayload(); + RequestInfo requestInfo = requestInfoImdg.getSingleObjectByID(statusInfo.getId()); + if (requestInfo == null) { + log.warn("unknown requestInfo id={}", statusInfo.getId()); + return; + } + requestInfo.setStatus(statusInfo.getStatus()); + requestInfoImdg.update(requestInfo); + } +} diff --git a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/service/impl/OperatorImpl.java b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/service/impl/OperatorImpl.java index dabfee7e6..216b25f58 100644 --- a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/service/impl/OperatorImpl.java +++ b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/service/impl/OperatorImpl.java @@ -8,6 +8,7 @@ import ru.spcex.clearing.backendapi.controller.response.cud.QueueSuccessResponse import ru.spcex.clearing.backendapi.domain.actions.IAction; import ru.spcex.clearing.backendapi.errors.ActionValidationException; import ru.spcex.clearing.backendapi.service.IOperator; +import ru.spcex.clearing.imdg.IMDGDistributedNames; import ru.spcex.clearing.platform.messaging.domain.BaseRequest; import ru.spcex.clearing.platform.messaging.service.RequestInfo; import ru.spcex.platform.imdg.api.Imdg; @@ -46,7 +47,7 @@ public class OperatorImpl implements IOperator { } private void saveRequestToStorage(String destination, BaseRequest request) { - Imdg requestStorage = imdgProvider.getImdg(destination, RequestInfo.class); + Imdg requestStorage = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class); RequestInfo requestInfo = RequestInfo.create(request.getId()); requestStorage.insert(requestInfo); } diff --git a/clearing-parent/backend-api/src/main/resources/application.properties b/clearing-parent/backend-api/src/main/resources/application.properties index 32e6e67c6..c375b65ed 100644 --- a/clearing-parent/backend-api/src/main/resources/application.properties +++ b/clearing-parent/backend-api/src/main/resources/application.properties @@ -12,14 +12,22 @@ backend-api.hazelcast.cluster-members=127.0.0.1:5701 backend-api.hazelcast.login=dev backend-api.hazelcast.password=dev-pass -backend-api.kafka.bootstrap-servers=localhost:9092 -backend-api.kafka.acks=all -backend-api.kafka.retries=0 -backend-api.kafka.batch-size=16384 -backend-api.kafka.linger-ms=1 -backend-api.kafka.buffer-memory=33554432 +backend-api.kafka-producer.bootstrap-servers=localhost:9092 +backend-api.kafka-producer.acks=all +backend-api.kafka-producer.retries=0 +backend-api.kafka-producer.batch-size=16384 +backend-api.kafka-producer.linger-ms=1 +backend-api.kafka-producer.buffer-memory=33554432 -backend-api.security.authorization-disabled=true +backend-api.kafka-consumer.bootstrap-servers=localhost:9092 +backend-api.kafka-consumer.group-id=dev-group-backend-api +backend-api.kafka-consumer.enable-auto-commit=true +backend-api.kafka-consumer.session-timeout-ms=30000 +backend-api.kafka-consumer.auto-offset-reset=latest +backend-api.kafka-consumer.linger-ms=1 +backend-api.kafka-consumer.buffer-memory=33554432 + +backend-api.security.authorization-disabled=false ##keycloak diff --git a/clearing-parent/backend-api/src/main/resources/messages/error.properties b/clearing-parent/backend-api/src/main/resources/messages/error.properties index bd5831476..52320a63a 100644 --- a/clearing-parent/backend-api/src/main/resources/messages/error.properties +++ b/clearing-parent/backend-api/src/main/resources/messages/error.properties @@ -1,4 +1,5 @@ 1=This is error example 1. 3000=validation error field '%s' 3001=unrecognized json property '%s' -3002=failed to read http message \ No newline at end of file +3002=failed to read http message +3004=couldn't find dictionary '%s' \ No newline at end of file diff --git a/clearing-parent/balance-service/src/main/java/ru/spcex/clearing/balance/config/KafkaSenderConfig.java b/clearing-parent/balance-service/src/main/java/ru/spcex/clearing/balance/config/KafkaSenderConfig.java index fa621992c..a32e3bc5e 100644 --- a/clearing-parent/balance-service/src/main/java/ru/spcex/clearing/balance/config/KafkaSenderConfig.java +++ b/clearing-parent/balance-service/src/main/java/ru/spcex/clearing/balance/config/KafkaSenderConfig.java @@ -4,6 +4,7 @@ import org.apache.kafka.clients.producer.Producer; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import ru.spcex.clearing.imdg.IMDGDistributedNames; import ru.spcex.clearing.platform.messaging.service.RequestInfo; import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender; import ru.spcex.platform.imdg.api.Imdg; @@ -22,7 +23,7 @@ public class KafkaSenderConfig { .producer(kafkaProducer) .idGenerator(imdgIdGenerator::nextId) .imdgProvider(s -> { - Imdg imdg = imdgProvider.getImdg(s, RequestInfo.class); + Imdg imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class); return imdg::insert; }) .build(); diff --git a/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/config/KafkaConfig.java b/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/config/KafkaConfig.java index 6ce4bca40..0be729887 100644 --- a/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/config/KafkaConfig.java +++ b/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/config/KafkaConfig.java @@ -1,6 +1,7 @@ package ru.spcex.clearing.company.config; import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.producer.Producer; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.config.ConfigurableBeanFactory; import org.springframework.context.annotation.Bean; @@ -8,6 +9,7 @@ import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Scope; import ru.spcex.clearing.company.config.settings.CompanyServiceSettings; import ru.spcex.clearing.platform.messaging.config.KafkaConsumerFactory; +import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory; @Configuration public class KafkaConfig { @@ -15,6 +17,13 @@ public class KafkaConfig { @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE) @Bean public Consumer createConsumer(CompanyServiceSettings settings) { - return KafkaConsumerFactory.consumer(settings.getKafka()); + return KafkaConsumerFactory.consumer(settings.getKafkaConsumer()); } + + @Autowired + @Bean + public Producer createProducer(CompanyServiceSettings settings) { + return KafkaProducerFactory.producer(settings.getKafkaProducer()); + } + } diff --git a/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/config/settings/CompanyServiceSettings.java b/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/config/settings/CompanyServiceSettings.java index e67bb2f8c..c27eae7f5 100644 --- a/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/config/settings/CompanyServiceSettings.java +++ b/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/config/settings/CompanyServiceSettings.java @@ -4,6 +4,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.PropertySource; import org.springframework.stereotype.Component; import ru.spcex.clearing.platform.messaging.config.element.KafkaConsumerSettings; +import ru.spcex.clearing.platform.messaging.config.element.KafkaProducerSettings; import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams; @Component @@ -11,7 +12,8 @@ import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams; @ConfigurationProperties("company-service") public class CompanyServiceSettings { private HazelcastClientParams hazelcast; - private KafkaConsumerSettings kafka; + private KafkaConsumerSettings kafkaConsumer; + private KafkaProducerSettings kafkaProducer; public HazelcastClientParams getHazelcast() { return hazelcast; @@ -21,11 +23,19 @@ public class CompanyServiceSettings { this.hazelcast = hazelcast; } - public KafkaConsumerSettings getKafka() { - return kafka; + public KafkaConsumerSettings getKafkaConsumer() { + return kafkaConsumer; } - public void setKafka(KafkaConsumerSettings kafka) { - this.kafka = kafka; + public void setKafkaConsumer(KafkaConsumerSettings kafkaConsumer) { + this.kafkaConsumer = kafkaConsumer; + } + + public KafkaProducerSettings getKafkaProducer() { + return kafkaProducer; + } + + public void setKafkaProducer(KafkaProducerSettings kafkaProducer) { + this.kafkaProducer = kafkaProducer; } } diff --git a/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/ClearingMemberCategoryService.java b/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/ClearingMemberCategoryService.java index 7ed87f27b..2cfab579f 100644 --- a/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/ClearingMemberCategoryService.java +++ b/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/ClearingMemberCategoryService.java @@ -1,6 +1,7 @@ package ru.spcex.clearing.company.service; import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.producer.Producer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.InitializingBean; @@ -20,9 +21,9 @@ public class ClearingMemberCategoryService extends QueueConsumer implements Init private final Imdg clearingMemberCategoryMap; @Autowired - public ClearingMemberCategoryService(Consumer kafkaQueue, + public ClearingMemberCategoryService(Consumer kafkaQueue, Producer kafkaProducer, ImdgProvider imdgProvider) { - super(kafkaQueue); + super(kafkaQueue, kafkaProducer); this.clearingMemberCategoryMap = imdgProvider.getImdg(IMDGDistributedNames.Map_ClearingMemberCategory, ClearingMemberCategory.class); } diff --git a/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/CompanyInfoService.java b/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/CompanyInfoService.java index 85775c9bb..953bd35b8 100644 --- a/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/CompanyInfoService.java +++ b/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/CompanyInfoService.java @@ -1,6 +1,7 @@ package ru.spcex.clearing.company.service; import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.producer.Producer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.InitializingBean; @@ -23,9 +24,9 @@ public class CompanyInfoService extends QueueConsumer implements InitializingBea private final Imdg companyMap; @Autowired - public CompanyInfoService(Consumer kafkaQueue, + public CompanyInfoService(Consumer kafkaQueue, Producer kafkaProducer, ImdgProvider imdgProvider) { - super(kafkaQueue); + super(kafkaQueue, kafkaProducer); // this.companyInfoMap = imdgProvider.getImdg(IMDGDistributedNames.Map_CompanyInfo, CompanyInfo.class); this.companyMap = imdgProvider.getImdg(IMDGDistributedNames.Map_Company, Company.class); } diff --git a/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/CompanyService.java b/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/CompanyService.java index 5a96460ac..e9008c33d 100644 --- a/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/CompanyService.java +++ b/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/CompanyService.java @@ -1,6 +1,7 @@ package ru.spcex.clearing.company.service; import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.producer.Producer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.InitializingBean; @@ -21,9 +22,9 @@ public class CompanyService extends QueueConsumer implements InitializingBean { private final Imdg companyMap; @Autowired - public CompanyService(Consumer kafkaQueue, + public CompanyService(Consumer kafkaQueue, Producer kafkaProducer, ImdgProvider imdgProvider) { - super(kafkaQueue); + super(kafkaQueue, kafkaProducer); this.companyMap = imdgProvider.getImdg(IMDGDistributedNames.Map_Company, Company.class); } diff --git a/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/CompanySymbolService.java b/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/CompanySymbolService.java index 4a6f110af..152fac925 100644 --- a/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/CompanySymbolService.java +++ b/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/CompanySymbolService.java @@ -2,6 +2,7 @@ package ru.spcex.clearing.company.service; import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.producer.Producer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.InitializingBean; @@ -22,9 +23,9 @@ public class CompanySymbolService extends QueueConsumer implements InitializingB private final Imdg companySymbolsMap; @Autowired - public CompanySymbolService(Consumer kafkaQueue, + public CompanySymbolService(Consumer kafkaQueue, Producer kafkaProducer, ImdgProvider imdgProvider) { - super(kafkaQueue); + super(kafkaQueue, kafkaProducer); this.companySymbolsMap = imdgProvider.getImdg(IMDGDistributedNames.Map_CompanySymbols, CompanySymbols.class); } diff --git a/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/ContactService.java b/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/ContactService.java index 0decc85f3..b38bf9ab2 100644 --- a/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/ContactService.java +++ b/clearing-parent/company-service/src/main/java/ru/spcex/clearing/company/service/ContactService.java @@ -1,6 +1,7 @@ package ru.spcex.clearing.company.service; import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.producer.Producer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.InitializingBean; @@ -19,9 +20,9 @@ public class ContactService extends QueueConsumer implements InitializingBean { private final Imdg contactMap; @Autowired - public ContactService(Consumer kafkaQueue, + public ContactService(Consumer kafkaQueue, Producer kafkaProducer, ImdgProvider imdgProvider) { - super(kafkaQueue); + super(kafkaQueue, kafkaProducer); this.contactMap = imdgProvider.getImdg(IMDGDistributedNames.Map_Contact, Contact.class); } diff --git a/clearing-parent/company-service/src/main/resources/application.properties b/clearing-parent/company-service/src/main/resources/application.properties index 292578a7e..6317c6c42 100644 --- a/clearing-parent/company-service/src/main/resources/application.properties +++ b/clearing-parent/company-service/src/main/resources/application.properties @@ -2,10 +2,17 @@ spring.main.web-application-type=none company-service.hazelcast.cluster-members=127.0.0.1:5701 company-service.hazelcast.login=dev company-service.hazelcast.password=dev-pass -company-service.kafka.bootstrap-servers=localhost:9092 -company-service.kafka.group-id=dev-group-company-service -company-service.kafka.enable-auto-commit=true -company-service.kafka.session-timeout-ms=30000 -company-service.kafka.auto-offset-reset=latest -company-service.kafka.linger-ms=1 -company-service.kafka.buffer-memory=33554432 \ No newline at end of file +company-service.kafka-consumer.bootstrap-servers=localhost:9092 +company-service.kafka-consumer.group-id=dev-group-company-service +company-service.kafka-consumer.enable-auto-commit=true +company-service.kafka-consumer.session-timeout-ms=30000 +company-service.kafka-consumer.auto-offset-reset=latest +company-service.kafka-consumer.linger-ms=1 +company-service.kafka-consumer.buffer-memory=33554432 + +company-service.kafka-producer.bootstrap-servers=localhost:9092 +company-service.kafka-producer.acks=all +company-service.kafka-producer.retries=0 +company-service.kafka-producer.batch-size=16384 +company-service.kafka-producer.linger-ms=1 +company-service.kafka-producer.buffer-memory=33554432 \ No newline at end of file diff --git a/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/config/KafkaConfig.java b/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/config/KafkaConfig.java index 86358d7c1..bf533336b 100644 --- a/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/config/KafkaConfig.java +++ b/clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/config/KafkaConfig.java @@ -5,6 +5,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import ru.spcex.clearing.dbf.importer.config.settings.ImportDBFServiceSettings; +import ru.spcex.clearing.imdg.IMDGDistributedNames; import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory; import ru.spcex.clearing.platform.messaging.service.RequestInfo; import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender; @@ -28,7 +29,7 @@ public class KafkaConfig { .producer(kafkaProducer) .idGenerator(imdgIdGenerator::nextId) .imdgProvider(s -> { - Imdg imdg = imdgProvider.getImdg(s, RequestInfo.class); + Imdg imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class); return imdg::insert; }) .build(); diff --git a/clearing-parent/dictionary/pom.xml b/clearing-parent/dictionary/pom.xml index cf6ecd958..323d95a0f 100644 --- a/clearing-parent/dictionary/pom.xml +++ b/clearing-parent/dictionary/pom.xml @@ -55,6 +55,10 @@ junit-jupiter test + + ru.spcex.platform + platform-classes-base + diff --git a/clearing-parent/dictionary/src/main/java/ru/clearing/platform/dictionary/AbstractDictionary.java b/clearing-parent/dictionary/src/main/java/ru/clearing/platform/dictionary/AbstractDictionary.java index e5c970836..d7186440b 100644 --- a/clearing-parent/dictionary/src/main/java/ru/clearing/platform/dictionary/AbstractDictionary.java +++ b/clearing-parent/dictionary/src/main/java/ru/clearing/platform/dictionary/AbstractDictionary.java @@ -1,22 +1,12 @@ package ru.clearing.platform.dictionary; -public abstract class AbstractDictionary implements Dictionary { - static final long serialVersionUID = ConstDictionarySerializable.serialVersionUID; +import ru.spcex.platform.classes.base.SpcexObjectBase; - protected Long id; +public abstract class AbstractDictionary extends SpcexObjectBase implements Dictionary { + static final long serialVersionUID = ConstDictionarySerializable.serialVersionUID; protected String code; protected String name; - @Override - public Long getId() { - return id; - } - - @Override - public void setId(Long id) { - this.id = id; - } - @Override public String getCode() { return code; diff --git a/clearing-parent/imdg/src/main/java/ru/spcex/clearing/imdg/dictionary/ClearingMemberCategoryDictionaryMapStore.java b/clearing-parent/imdg/src/main/java/ru/spcex/clearing/imdg/dictionary/ClearingMemberCategoryDictionaryMapStore.java new file mode 100644 index 000000000..c9fc60c74 --- /dev/null +++ b/clearing-parent/imdg/src/main/java/ru/spcex/clearing/imdg/dictionary/ClearingMemberCategoryDictionaryMapStore.java @@ -0,0 +1,31 @@ +package ru.spcex.clearing.imdg.dictionary; + +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Component; +import ru.clearing.platform.dictionary.ClearingMemberCategoryDictionary; +import ru.spcex.clearing.imdg.IMDGDistributedNames; +import ru.spcex.clearing.imdg.base.DictionaryTMapStore; + +@Component +public class ClearingMemberCategoryDictionaryMapStore extends DictionaryTMapStore { + + public ClearingMemberCategoryDictionaryMapStore(JdbcTemplate jdbcTemplate) { + super(jdbcTemplate); + } + + @Override + public String getMapName() { + return IMDGDistributedNames.Map_ClearingMemberCategoryDictionary; + } + + @Override + public String getTableName() { + return "CLEARING_MEMBER_CATEGORY_DICTIONARY"; + } + + @Override + public ClearingMemberCategoryDictionary getDictionaryObject() { + return new ClearingMemberCategoryDictionary(); + } + +} \ No newline at end of file diff --git a/clearing-parent/securities-service/src/main/java/ru/spcex/clearing/securities/config/KafkaConfig.java b/clearing-parent/securities-service/src/main/java/ru/spcex/clearing/securities/config/KafkaConfig.java index deb72e8ca..1669d0e82 100644 --- a/clearing-parent/securities-service/src/main/java/ru/spcex/clearing/securities/config/KafkaConfig.java +++ b/clearing-parent/securities-service/src/main/java/ru/spcex/clearing/securities/config/KafkaConfig.java @@ -1,10 +1,12 @@ package ru.spcex.clearing.securities.config; import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.producer.Producer; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import ru.spcex.clearing.platform.messaging.config.KafkaConsumerFactory; +import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory; import ru.spcex.clearing.securities.config.element.SecuritiesServiceSettings; @Configuration @@ -12,8 +14,15 @@ public class KafkaConfig { @Autowired @Bean - public Consumer createProducer(SecuritiesServiceSettings settings) { - return KafkaConsumerFactory.consumer(settings.getKafka()); + public Consumer createConsumer(SecuritiesServiceSettings settings) { + return KafkaConsumerFactory.consumer(settings.getKafkaConsumer()); } + @Autowired + @Bean + public Producer createProducer(SecuritiesServiceSettings settings) { + return KafkaProducerFactory.producer(settings.getKafkaProducer()); + } + + } diff --git a/clearing-parent/securities-service/src/main/java/ru/spcex/clearing/securities/config/element/SecuritiesServiceSettings.java b/clearing-parent/securities-service/src/main/java/ru/spcex/clearing/securities/config/element/SecuritiesServiceSettings.java index 546f25cb6..2ef7db60d 100644 --- a/clearing-parent/securities-service/src/main/java/ru/spcex/clearing/securities/config/element/SecuritiesServiceSettings.java +++ b/clearing-parent/securities-service/src/main/java/ru/spcex/clearing/securities/config/element/SecuritiesServiceSettings.java @@ -4,6 +4,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.PropertySource; import org.springframework.stereotype.Component; import ru.spcex.clearing.platform.messaging.config.element.KafkaConsumerSettings; +import ru.spcex.clearing.platform.messaging.config.element.KafkaProducerSettings; import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams; @Component @@ -11,7 +12,8 @@ import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams; @ConfigurationProperties("securities-service") public class SecuritiesServiceSettings { private HazelcastClientParams hazelcast; - private KafkaConsumerSettings kafka; + private KafkaConsumerSettings kafkaConsumer; + private KafkaProducerSettings kafkaProducer; public HazelcastClientParams getHazelcast() { return hazelcast; @@ -21,11 +23,19 @@ public class SecuritiesServiceSettings { this.hazelcast = hazelcast; } - public KafkaConsumerSettings getKafka() { - return kafka; + public KafkaConsumerSettings getKafkaConsumer() { + return kafkaConsumer; } - public void setKafka(KafkaConsumerSettings kafka) { - this.kafka = kafka; + public void setKafkaConsumer(KafkaConsumerSettings kafkaConsumer) { + this.kafkaConsumer = kafkaConsumer; + } + + public KafkaProducerSettings getKafkaProducer() { + return kafkaProducer; + } + + public void setKafkaProducer(KafkaProducerSettings kafkaProducer) { + this.kafkaProducer = kafkaProducer; } } diff --git a/clearing-parent/securities-service/src/main/java/ru/spcex/clearing/securities/service/cud/MoneyMarketSecurityService.java b/clearing-parent/securities-service/src/main/java/ru/spcex/clearing/securities/service/cud/MoneyMarketSecurityService.java index 279c09bd5..a195fc683 100644 --- a/clearing-parent/securities-service/src/main/java/ru/spcex/clearing/securities/service/cud/MoneyMarketSecurityService.java +++ b/clearing-parent/securities-service/src/main/java/ru/spcex/clearing/securities/service/cud/MoneyMarketSecurityService.java @@ -1,6 +1,7 @@ package ru.spcex.clearing.securities.service.cud; import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.producer.Producer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.InitializingBean; @@ -19,7 +20,6 @@ import ru.spcex.clearing.platform.messaging.service.Status; import ru.spcex.platform.imdg.api.Imdg; import ru.spcex.platform.imdg.api.ImdgProvider; import ru.spcex.platform.imdg.api.ImdgTransaction; -import ru.spcex.platform.utils.log.ExceptionUtils; import java.math.BigDecimal; @@ -29,15 +29,16 @@ public class MoneyMarketSecurityService extends QueueConsumer implements Initial private final Imdg moneyMarketSecurityMap; private final ImdgProvider imdgProvider; @Autowired - public MoneyMarketSecurityService(Consumer kafkaQueue, ImdgProvider imdgProvider) { - super(kafkaQueue); + public MoneyMarketSecurityService(Consumer kafkaQueue, Producer kafkaProducer, + ImdgProvider imdgProvider) { + super(kafkaQueue, kafkaProducer); this.moneyMarketSecurityMap = imdgProvider.getImdg(IMDGDistributedNames.Map_MoneyMarketSecurity, MoneyMarketSecurity.class); this.imdgProvider = imdgProvider; } @Override protected boolean needsProcessing(String destination, BaseRequest request) { - Imdg requestSpecificImdg = imdgProvider.getImdg(destination, RequestInfo.class); + Imdg requestSpecificImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class); RequestInfo requestFound = requestSpecificImdg.getSingleObjectByID(request.getId()); if (requestFound == null) { log.error("couldn't find request info by id={}, action type={}", request.getId(), request.getActionType()); @@ -66,8 +67,8 @@ public class MoneyMarketSecurityService extends QueueConsumer implements Initial } private void newMoneyMarket(BaseRequest userRequest) { - Imdg requestInfoImdg = imdgProvider.getImdg(Consts.DESTINATION_MONEY_MARKET_SECURITY_NEW, RequestInfo.class); - RequestInfo reqInfo = requestInfoImdg.getSingleObjectByID(userRequest.getId()); +// Imdg requestInfoImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class); +// RequestInfo reqInfo = requestInfoImdg.getSingleObjectByID(userRequest.getId()); ImdgTransaction transaction = imdgProvider.newTransaction(); MoneyMarketSecurityNewRequest req = userRequest.getRequestPayload(); log.debug("MoneyMarketSecurityNewRequest received"); @@ -82,17 +83,17 @@ public class MoneyMarketSecurityService extends QueueConsumer implements Initial try { transaction.beginTransaction(); Imdg moneyMarketSecurityMap = transaction.getImdg(IMDGDistributedNames.Map_MoneyMarketSecurity, MoneyMarketSecurity.class); - Imdg reqInfoMap = transaction.getImdg(Consts.DESTINATION_MONEY_MARKET_SECURITY_NEW, RequestInfo.class); + Imdg reqInfoMap = transaction.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class); //одним скопом выполняем реквест moneyMarketSecurityMap.insert(mms); - RequestInfo.update(reqInfo, Status.Success, "success"); +// RequestInfo.update(reqInfo, Status.Success, "success"); //и сохраняем обновленный - reqInfoMap.insert(reqInfo); +// reqInfoMap.insert(reqInfo); transaction.commitTransaction(); } catch (Throwable e) { transaction.rollbackTransaction(); - RequestInfo.update(reqInfo, Status.Error, ExceptionUtils.getStackTrace(e)); - requestInfoImdg.insert(reqInfo); +// RequestInfo.update(reqInfo, Status.Error, ExceptionUtils.getStackTrace(e)); +// requestInfoImdg.insert(reqInfo); } log.debug("successfully processed, request id {}, new object id {}", userRequest.getId(), mms.getId()); } diff --git a/clearing-parent/securities-service/src/main/resources/application.properties b/clearing-parent/securities-service/src/main/resources/application.properties index ee89c6710..1fae05658 100644 --- a/clearing-parent/securities-service/src/main/resources/application.properties +++ b/clearing-parent/securities-service/src/main/resources/application.properties @@ -4,13 +4,20 @@ securities-service.hazelcast.cluster-members=127.0.0.1:5701 securities-service.hazelcast.login=dev securities-service.hazelcast.password=dev-pass -securities-service.kafka.bootstrap-servers=localhost:9092 -securities-service.kafka.group-id=dev-group -securities-service.kafka.enable-auto-commit=false -securities-service.kafka.session-timeout-ms=30000 -securities-service.kafka.auto-offset-reset=latest -securities-service.kafka.linger-ms=1 -securities-service.kafka.buffer-memory=33554432 +securities-service.kafka-consumer.bootstrap-servers=localhost:9092 +securities-service.kafka-consumer.group-id=dev-group-securities-service +securities-service.kafka-consumer.enable-auto-commit=false +securities-service.kafka-consumer.session-timeout-ms=30000 +securities-service.kafka-consumer.auto-offset-reset=latest +securities-service.kafka-consumer.linger-ms=1 +securities-service.kafka-consumer.buffer-memory=33554432 + +securities-service.kafka-producer.bootstrap-servers=localhost:9092 +securities-service.kafka-producer.acks=all +securities-service.kafka-producer.retries=0 +securities-service.kafka-producer.batch-size=16384 +securities-service.kafka-producer.linger-ms=1 +securities-service.kafka-producer.buffer-memory=33554432 #props.put("bootstrap.servers", kafkaSettings.getBootstrapServers()); #if (kafkaSettings.getGroupId() != null && kafkaSettings.getGroupId().length() > 0) { # props.put("group.id", kafkaSettings.getGroupId()); diff --git a/clearing-parent/utility-service/src/main/java/ru/spcex/clearing/utility/config/KafkaConfig.java b/clearing-parent/utility-service/src/main/java/ru/spcex/clearing/utility/config/KafkaConfig.java index bc8a2546a..2dde7ea72 100644 --- a/clearing-parent/utility-service/src/main/java/ru/spcex/clearing/utility/config/KafkaConfig.java +++ b/clearing-parent/utility-service/src/main/java/ru/spcex/clearing/utility/config/KafkaConfig.java @@ -1,12 +1,14 @@ package ru.spcex.clearing.utility.config; import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.producer.Producer; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.config.ConfigurableBeanFactory; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Scope; import ru.spcex.clearing.platform.messaging.config.KafkaConsumerFactory; +import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory; import ru.spcex.clearing.utility.config.settings.UtilityServiceSettings; @Configuration @@ -14,7 +16,14 @@ public class KafkaConfig { @Autowired @Bean @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE) - public Consumer createProducer(UtilityServiceSettings settings) { - return KafkaConsumerFactory.consumer(settings.getKafka()); + public Consumer createConsumer(UtilityServiceSettings settings) { + return KafkaConsumerFactory.consumer(settings.getKafkaConsumer()); } + + @Autowired + @Bean + public Producer createProducer(UtilityServiceSettings settings) { + return KafkaProducerFactory.producer(settings.getKafkaProducer()); + } + } diff --git a/clearing-parent/utility-service/src/main/java/ru/spcex/clearing/utility/config/settings/UtilityServiceSettings.java b/clearing-parent/utility-service/src/main/java/ru/spcex/clearing/utility/config/settings/UtilityServiceSettings.java index 54f22e60c..f7db55b39 100644 --- a/clearing-parent/utility-service/src/main/java/ru/spcex/clearing/utility/config/settings/UtilityServiceSettings.java +++ b/clearing-parent/utility-service/src/main/java/ru/spcex/clearing/utility/config/settings/UtilityServiceSettings.java @@ -4,6 +4,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.PropertySource; import org.springframework.stereotype.Component; import ru.spcex.clearing.platform.messaging.config.element.KafkaConsumerSettings; +import ru.spcex.clearing.platform.messaging.config.element.KafkaProducerSettings; import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams; @Component @@ -11,7 +12,8 @@ import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams; @ConfigurationProperties("utility-service") public class UtilityServiceSettings { private HazelcastClientParams hazelcast; - private KafkaConsumerSettings kafka; + private KafkaConsumerSettings kafkaConsumer; + private KafkaProducerSettings kafkaProducer; public HazelcastClientParams getHazelcast() { return hazelcast; @@ -21,11 +23,19 @@ public class UtilityServiceSettings { this.hazelcast = hazelcast; } - public KafkaConsumerSettings getKafka() { - return kafka; + public KafkaConsumerSettings getKafkaConsumer() { + return kafkaConsumer; } - public void setKafka(KafkaConsumerSettings kafka) { - this.kafka = kafka; + public void setKafkaConsumer(KafkaConsumerSettings kafkaConsumer) { + this.kafkaConsumer = kafkaConsumer; + } + + public KafkaProducerSettings getKafkaProducer() { + return kafkaProducer; + } + + public void setKafkaProducer(KafkaProducerSettings kafkaProducer) { + this.kafkaProducer = kafkaProducer; } } diff --git a/clearing-parent/utility-service/src/main/java/ru/spcex/clearing/utility/service/KeyRateService.java b/clearing-parent/utility-service/src/main/java/ru/spcex/clearing/utility/service/KeyRateService.java index 92c3cdf26..b96fa1032 100644 --- a/clearing-parent/utility-service/src/main/java/ru/spcex/clearing/utility/service/KeyRateService.java +++ b/clearing-parent/utility-service/src/main/java/ru/spcex/clearing/utility/service/KeyRateService.java @@ -1,6 +1,7 @@ package ru.spcex.clearing.utility.service; import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.producer.Producer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.InitializingBean; @@ -23,8 +24,9 @@ public class KeyRateService extends QueueConsumer implements InitializingBean { private final Imdg keyRateMap; @Autowired - public KeyRateService(Consumer kafkaQueue, ImdgProvider imdgProvider) { - super(kafkaQueue); + public KeyRateService(Consumer kafkaQueue, Producer kafkaProducer, + ImdgProvider imdgProvider) { + super(kafkaQueue, kafkaProducer); this.keyRateMap = imdgProvider.getImdg(IMDGDistributedNames.Map_KeyRate, KeyRate.class); } diff --git a/clearing-parent/utility-service/src/main/java/ru/spcex/clearing/utility/service/UserSettingService.java b/clearing-parent/utility-service/src/main/java/ru/spcex/clearing/utility/service/UserSettingService.java index f34af7255..0c069f2fb 100644 --- a/clearing-parent/utility-service/src/main/java/ru/spcex/clearing/utility/service/UserSettingService.java +++ b/clearing-parent/utility-service/src/main/java/ru/spcex/clearing/utility/service/UserSettingService.java @@ -1,6 +1,7 @@ package ru.spcex.clearing.utility.service; import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.producer.Producer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.InitializingBean; @@ -23,8 +24,9 @@ public class UserSettingService extends QueueConsumer implements InitializingBea private final Imdg userSettingsMap; @Autowired - public UserSettingService(Consumer kafkaQueue, ImdgProvider imdgProvider) { - super(kafkaQueue); + public UserSettingService(Consumer kafkaQueue, Producer kafkaProducer, + ImdgProvider imdgProvider) { + super(kafkaQueue, kafkaProducer); this.userSettingsMap = imdgProvider.getImdg(IMDGDistributedNames.Map_UserSettings, UserSettings.class); } diff --git a/clearing-parent/utility-service/src/main/resources/application.properties b/clearing-parent/utility-service/src/main/resources/application.properties index fbc306eab..73a36eb86 100644 --- a/clearing-parent/utility-service/src/main/resources/application.properties +++ b/clearing-parent/utility-service/src/main/resources/application.properties @@ -4,10 +4,17 @@ utility-service.hazelcast.cluster-members=127.0.0.1:5701 utility-service.hazelcast.login=dev utility-service.hazelcast.password=dev-pass -utility-service.kafka.bootstrap-servers=localhost:9092 -utility-service.kafka.group-id=dev-group-utility-service -utility-service.kafka.enable-auto-commit=true -utility-service.kafka.session-timeout-ms=30000 -utility-service.kafka.auto-offset-reset=latest -utility-service.kafka.linger-ms=1 -utility-service.kafka.buffer-memory=33554432 \ No newline at end of file +utility-service.kafka-consumer.bootstrap-servers=localhost:9092 +utility-service.kafka-consumer.group-id=dev-group-utility-service +utility-service.kafka-consumer.enable-auto-commit=true +utility-service.kafka-consumer.session-timeout-ms=30000 +utility-service.kafka-consumer.auto-offset-reset=latest +utility-service.kafka-consumer.linger-ms=1 +utility-service.kafka-consumer.buffer-memory=33554432 + +utility-service.kafka-producer.bootstrap-servers=localhost:9092 +utility-service.kafka-producer.acks=all +utility-service.kafka-producer.retries=0 +utility-service.kafka-producer.batch-size=16384 +utility-service.kafka-producer.linger-ms=1 +utility-service.kafka-producer.buffer-memory=33554432 \ No newline at end of file diff --git a/platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/adapter/ImdgHazelcast.java b/platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/adapter/ImdgHazelcast.java index cb848122f..940adedcb 100644 --- a/platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/adapter/ImdgHazelcast.java +++ b/platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/adapter/ImdgHazelcast.java @@ -29,6 +29,11 @@ public class ImdgHazelcast implements Imdg { return map.values(); } + @Override + public Integer size() { + return map.size(); + } + @Override public void insert(T paramT) { if (paramT.getId() == null) { diff --git a/platform-parent/platform-imdg-api/src/main/java/ru/spcex/clearing/imdg/IMDGDistributedNames.java b/platform-parent/platform-imdg-api/src/main/java/ru/spcex/clearing/imdg/IMDGDistributedNames.java index 4b7889a9f..dd97b9940 100644 --- a/platform-parent/platform-imdg-api/src/main/java/ru/spcex/clearing/imdg/IMDGDistributedNames.java +++ b/platform-parent/platform-imdg-api/src/main/java/ru/spcex/clearing/imdg/IMDGDistributedNames.java @@ -101,6 +101,7 @@ public final class IMDGDistributedNames { public static final String Map_AccountStatusDictionary = "Map_AccountStatusDictionary"; public static final String Map_KeyRate = "Map_KeyRate"; public static final String Map_MoneyMarketSecurity = "Map_MoneyMarketSecurity"; + public static final String Map_RequestInfo = "Map_RequestInfo"; public static final String MAP_SEQUENCE_NAME = "MAP_SEQUENCE_NAME"; // todo вынести idGenerator отдельно и завернуть в метод, чтобы не напрямую обращаться. private IMDGDistributedNames() { diff --git a/platform-parent/platform-imdg-api/src/main/java/ru/spcex/platform/imdg/api/Imdg.java b/platform-parent/platform-imdg-api/src/main/java/ru/spcex/platform/imdg/api/Imdg.java index 845eeb86f..ca765d5b5 100644 --- a/platform-parent/platform-imdg-api/src/main/java/ru/spcex/platform/imdg/api/Imdg.java +++ b/platform-parent/platform-imdg-api/src/main/java/ru/spcex/platform/imdg/api/Imdg.java @@ -57,4 +57,8 @@ public interface Imdg { default Collection projectionsAttributeBySql(String paramString, String... paramVarArgs) { throw new UnsupportedOperationException("not implemented projectionsAttributeBySql"); } + + default Integer size() { + throw new UnsupportedOperationException("not implemented size"); + } } diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java index 212900ad1..3eca7e9a6 100644 --- a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java @@ -33,4 +33,6 @@ public interface Consts { String ACCOUNT_NEW = "account-new"; String BALANCE_ACCOUNT_NEW = "balance-account-new"; + String REQUEST_INFO_UPDATE = "request-info-update"; + } diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/json/serialize/EnumSerializer.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/json/serialize/EnumSerializer.java new file mode 100644 index 000000000..6cf93cc26 --- /dev/null +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/json/serialize/EnumSerializer.java @@ -0,0 +1,19 @@ +package ru.spcex.clearing.platform.messaging.domain.json.serialize; + +import com.fasterxml.jackson.core.JsonGenerator; +import com.fasterxml.jackson.databind.JsonSerializer; +import com.fasterxml.jackson.databind.SerializerProvider; + +import java.io.IOException; +import java.time.format.DateTimeFormatter; + +public class EnumSerializer extends JsonSerializer> { + private static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd"); + + + @Override + public void serialize(Enum value, JsonGenerator gen, SerializerProvider serializers) throws IOException { + if (value == null) return; + gen.writeString(value.name()); + } +} diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/logic/functional/BuilderConsumerStep.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/logic/functional/BuilderConsumerStep.java index 88d687f2c..224b7b1c7 100644 --- a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/logic/functional/BuilderConsumerStep.java +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/logic/functional/BuilderConsumerStep.java @@ -3,7 +3,9 @@ package ru.spcex.clearing.platform.messaging.logic.functional; import ru.spcex.clearing.platform.messaging.domain.BaseRequest; import java.util.function.Consumer; +import java.util.function.Function; public interface BuilderConsumerStep { BuilderDestinationStep setConsumer(Consumer> consumer); + BuilderDestinationStep setFunction(Function, Object> function); } diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/logic/functional/ConsumerSpecificClass.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/logic/functional/ConsumerSpecificClass.java index 597cf8b04..0b6e96404 100644 --- a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/logic/functional/ConsumerSpecificClass.java +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/logic/functional/ConsumerSpecificClass.java @@ -4,6 +4,7 @@ import ru.spcex.clearing.platform.messaging.domain.BaseRequest; import java.util.function.BiConsumer; import java.util.function.Consumer; +import java.util.function.Function; /** * вспомогательный класс @@ -18,6 +19,7 @@ public class ConsumerSpecificClass implements BuilderConsumerStep, Builder * действие-обработчик реквеста (метод из конкретного менеджера) */ private java.util.function.Consumer> consumer; + private java.util.function.Function, Object> function; private ConsumerSpecificClass(Class clazz) { this.clazz = clazz; @@ -29,7 +31,17 @@ public class ConsumerSpecificClass implements BuilderConsumerStep, Builder @Override public BuilderDestinationStep setConsumer(Consumer> consumer) { - this.consumer = consumer; + if (this.consumer == null) { + this.consumer = consumer; + } else { + this.consumer = this.consumer.andThen(consumer); + } + return this; + } + + @Override + public BuilderDestinationStep setFunction(Function, Object> function) { + this.function = function; return this; } @@ -38,10 +50,18 @@ public class ConsumerSpecificClass implements BuilderConsumerStep, Builder } @SuppressWarnings("unchecked") - public void acceptRaw(Object obj) { - this.consumer.accept((BaseRequest) obj); + public Object acceptRaw(Object obj) { + if (this.consumer != null) { + this.consumer.accept((BaseRequest) obj); + return null; + } else { + return this.function.apply((BaseRequest) obj); + } } + public Consumer> getConsumer() { + return consumer; + } public Class getClazz() { return clazz; } diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/QueueConsumer.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/QueueConsumer.java index 7a1f96915..0e11d3bb8 100644 --- a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/QueueConsumer.java +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/QueueConsumer.java @@ -5,10 +5,15 @@ import com.fasterxml.jackson.databind.ObjectMapper; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.kafka.common.errors.WakeupException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +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.logic.functional.BuilderConsumerStep; import ru.spcex.clearing.platform.messaging.logic.functional.ConsumerSpecificClass; import ru.spcex.platform.utils.log.ExceptionUtils; @@ -19,6 +24,7 @@ import java.util.HashMap; import java.util.Map; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.atomic.AtomicBoolean; /** @@ -28,33 +34,40 @@ public class QueueConsumer implements AutoCloseable { private final Logger log = LoggerFactory.getLogger(getClass()); private final AtomicBoolean closed = new AtomicBoolean(false); private final Consumer consumer; - private final ExecutorService executor; + private Producer producer; + private final ExecutorService inputExecutor; + private final ExecutorService outputExecutor; protected final Map> callbacks; private final ObjectMapper json; + protected boolean supportStartOffsetTimeWindow; public QueueConsumer(Consumer kafkaQueue) { this.consumer = kafkaQueue; this.callbacks = new HashMap<>(); - this.executor = Executors.newSingleThreadExecutor(); + this.inputExecutor = Executors.newSingleThreadExecutor(); + this.outputExecutor = Executors.newSingleThreadExecutor(); this.json = new ObjectMapper(); + this.supportStartOffsetTimeWindow = false; + } + + public QueueConsumer(Consumer kafkaQueue, Producer kafkaResponseQueue) { + this(kafkaQueue); + this.producer = kafkaResponseQueue; } protected boolean needsProcessing(String topicName, BaseRequest request) { return true; } - protected boolean supportStartOffsetTimeWindow() { - return false; - } - public void init() { - executor.submit(() -> { + inputExecutor.submit(() -> { try { - if (supportStartOffsetTimeWindow()) { + if (supportStartOffsetTimeWindow) { consumer.subscribe(callbacks.keySet(), new OffsetChanger(consumer, callbacks.keySet())); } else { consumer.subscribe(callbacks.keySet()); } + Object o = null; while (!closed.get()) { try { ConsumerRecords records = consumer.poll(Duration.of(10, ChronoUnit.SECONDS)); @@ -62,13 +75,19 @@ public class QueueConsumer implements AutoCloseable { ConsumerSpecificClass callback = callbacks.get(next.topic()); Class clazz = callback.getClazz(); JavaType payloadType = json.getTypeFactory().constructParametricType(BaseRequest.class, clazz); - Object o = json.readValue((String) next.value(), payloadType); + o = json.readValue((String) next.value(), payloadType); if (needsProcessing(next.topic(), (BaseRequest) o)) { - callback.acceptRaw(o); + Object topicResponse = callback.acceptRaw(o); + if (producer != null) { + sendResponse((BaseRequest) o, topicResponse); + } } } } catch (Throwable e) { log.error(ExceptionUtils.getStackTrace(e)); + if (producer != null) { + sendErrorResponse((BaseRequest) o); + } } } } catch (WakeupException e) { @@ -81,6 +100,51 @@ public class QueueConsumer implements AutoCloseable { }); } + //мб перенести в другой класс + private void sendErrorResponse(BaseRequest o) { + outputExecutor.submit(() -> { + try { + Future send; + //default response + BaseRequest req = new BaseRequest<>(); + RequestInfoUpdate success = new RequestInfoUpdate(); + success.setId(o.getId()); + success.setStatus(Status.Error); + req.setRequestPayload(success); + req.setId(o.getId()); + req.setActionType(ActionType.SYSTEM); + send = producer.send(new ProducerRecord<>(Consts.REQUEST_INFO_UPDATE, req)); + send.get(); + } catch (Exception e) { + log.error(ExceptionUtils.getStackTrace(e)); + } + }); + } + + private void sendResponse(BaseRequest o, Object response) { + outputExecutor.submit(() -> { + try { + Future send; + BaseRequest req = new BaseRequest<>(); + req.setId(o.getId()); + req.setActionType(ActionType.SYSTEM); + if (response != null) { + req.setRequestPayload(response); + } else { + //default response + RequestInfoUpdate success = new RequestInfoUpdate(); + success.setId(o.getId()); + success.setStatus(Status.Success); + req.setRequestPayload(success); + } + send = producer.send(new ProducerRecord<>(Consts.REQUEST_INFO_UPDATE, req)); + send.get(); + } catch (Exception e) { + log.error(ExceptionUtils.getStackTrace(e)); + } + }); + } + protected BuilderConsumerStep callback(Class clazz) { return ConsumerSpecificClass.build(clazz); } diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/RequestInfoUpdate.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/RequestInfoUpdate.java new file mode 100644 index 000000000..d83b8e9c1 --- /dev/null +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/RequestInfoUpdate.java @@ -0,0 +1,22 @@ +package ru.spcex.clearing.platform.messaging.service; + +public class RequestInfoUpdate { + private Long id; + private Status status; + + public Long getId() { + return id; + } + + public void setId(Long id) { + this.id = id; + } + + public Status getStatus() { + return status; + } + + public void setStatus(Status status) { + this.status = status; + } +}