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

company-service http://jira.mfd.msk:8088/browse/CLS-272 исправил использование BiDirectionQueueExchanger

Browse source
This commit is contained in:
AKurakin 2023-05-04 16:29:52 +03:00
parent f1a803aacf
commit 864bbcf125
4 changed files with 5 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -55,7 +55,7 @@ public class BiDirectionQueueExchanger<TOut extends BaseRequest<?>> extends Queu
/** /**
* Синхронно-ассинхронный обмен сообщениями * Синхронно-ассинхронный обмен сообщениями
* *
* @param kafkaQueue * @param kafkaQueue уникальный kafka Consumer (Spring prototype). Нельзя переиспользовать существующие.
* @param kafkaProducer * @param kafkaProducer
* @param outQueue отправляет в очередь * @param outQueue отправляет в очередь
* @param inQueue слушает очередь/топик, ожидает ответов. Пример: Consts.CONTINUE_CLEARING * @param inQueue слушает очередь/топик, ожидает ответов. Пример: Consts.CONTINUE_CLEARING

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

@ -70,7 +70,7 @@ public class ClientCodeService extends QueueConsumer implements InitializingBean
BiDirectionQueueExchanger<BaseRequest<TradingClearingRegistryNewRequest>> accountServiceExchanger; BiDirectionQueueExchanger<BaseRequest<TradingClearingRegistryNewRequest>> accountServiceExchanger;
@Autowired @Autowired
public ClientCodeService(Consumer<String, Object> kafkaQueue, public ClientCodeService(Consumer<String, Object> kafkaQueue1, Consumer<String, Object> kafkaQueue2,
Producer<String, Object> kafkaProducer, Producer<String, Object> kafkaProducer,
ImdgProvider imdgProvider, ImdgProvider imdgProvider,
ValidationHelper validationHelper, ValidationHelper validationHelper,
@ -79,7 +79,7 @@ public class ClientCodeService extends QueueConsumer implements InitializingBean
@Qualifier("clientCodeNewRequestValidator") Function<ClientCodeNewRequest, IValidator> clientCodeNewRequestValidator, @Qualifier("clientCodeNewRequestValidator") Function<ClientCodeNewRequest, IValidator> clientCodeNewRequestValidator,
@Qualifier("clientCodeUpdateRequestValidator") Function<ClientCodeUpdateRequest, IValidator> clientCodeUpdateRequestValidator, @Qualifier("clientCodeUpdateRequestValidator") Function<ClientCodeUpdateRequest, IValidator> clientCodeUpdateRequestValidator,
@Qualifier("clientCodeDeleteRequestValidator") Function<CommonDeleteRequest, IValidator> clientCodeDeleteRequestValidator) { @Qualifier("clientCodeDeleteRequestValidator") Function<CommonDeleteRequest, IValidator> clientCodeDeleteRequestValidator) {
super(kafkaQueue, kafkaProducer); super(kafkaQueue1, kafkaProducer);
this.kafkaProducer = kafkaProducer; this.kafkaProducer = kafkaProducer;
this.idGenerator = imdgProvider.getImdgIdGenerator(); this.idGenerator = imdgProvider.getImdgIdGenerator();
this.clientCodeMap = imdgProvider.getImdg(IMDGDistributedNames.Map_ClientCode, ClientCode.class); this.clientCodeMap = imdgProvider.getImdg(IMDGDistributedNames.Map_ClientCode, ClientCode.class);
@ -92,7 +92,7 @@ public class ClientCodeService extends QueueConsumer implements InitializingBean
this.messageResolver = messageResolver; this.messageResolver = messageResolver;
this.userRoleVerification = userRoleVerification; this.userRoleVerification = userRoleVerification;
this.accountServiceExchanger = new BiDirectionQueueExchanger<>(kafkaQueue, kafkaProducer, this.accountServiceExchanger = new BiDirectionQueueExchanger<>(kafkaQueue2, kafkaProducer,
Consts.DESTINATION_TRADING_CLEARING_REGISTRY_NEW, Consts.DESTINATION_TRADING_CLEARING_REGISTRY_NEW,
Consts.REQUEST_INFO_UPDATE, true, // REQUEST_INFO_UPDATE - стандартная очередь, для результатов всех реквестов. DESTINATION_TRADING_CLEARING_REGISTRY_REPLY, Consts.REQUEST_INFO_UPDATE, true, // REQUEST_INFO_UPDATE - стандартная очередь, для результатов всех реквестов. DESTINATION_TRADING_CLEARING_REGISTRY_REPLY,
60000 60000

1
clearing-parent/company-service/src/test/java/ru/spcex/clearing/company/service/ClientCodeServiceTest.java
View file

@ -45,7 +45,6 @@ import static ru.spcex.clearing.test.MatcherFactory.usingIgnoringFieldsComparato
import static ru.spcex.clearing.test.TestUtils.*; import static ru.spcex.clearing.test.TestUtils.*;
import static ru.spcex.clearing.test.config.ImdgTestConfig.waitAvailableImdgProviderAndAddAdminWithDefaultId; import static ru.spcex.clearing.test.config.ImdgTestConfig.waitAvailableImdgProviderAndAddAdminWithDefaultId;
//todo тесты случайно ломаются из-за BiDirectionQueueExchanger. Когда там выключаю initReplyListener() то тесты стабильно работают.
@ExtendWith(SpringExtension.class) @ExtendWith(SpringExtension.class)
@ContextConfiguration(classes = { @ContextConfiguration(classes = {
ClientCodeService.class, ClientCodeService.class,

1
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/QueueConsumer.java
View file

@ -74,6 +74,7 @@ public class QueueConsumer implements AutoCloseable {
public void init() { public void init() {
inputExecutor.submit(() -> { inputExecutor.submit(() -> {
try { try {
log.debug("Using kafka consumer {} for subscribe on \"{}\"", consumer, callbacks.keySet());
if (supportStartOffsetTimeWindow) { if (supportStartOffsetTimeWindow) {
consumer.subscribe(callbacks.keySet(), new OffsetChanger(consumer, callbacks.keySet())); consumer.subscribe(callbacks.keySet(), new OffsetChanger(consumer, callbacks.keySet()));
} else { } else {