diff --git a/clearing-parent/clearing-utils/src/main/java/ru/spcex/clearing/util/services/exchangers/BiDirectionQueueExchanger.java b/clearing-parent/clearing-utils/src/main/java/ru/spcex/clearing/util/services/exchangers/BiDirectionQueueExchanger.java new file mode 100644 index 000000000..0f08fe624 --- /dev/null +++ b/clearing-parent/clearing-utils/src/main/java/ru/spcex/clearing/util/services/exchangers/BiDirectionQueueExchanger.java @@ -0,0 +1,66 @@ +package ru.spcex.clearing.util.services.exchangers; + +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.DisposableBean; +import org.springframework.beans.factory.InitializingBean; +import ru.spcex.clearing.platform.messaging.domain.BaseRequest; +import ru.spcex.clearing.platform.messaging.service.QueueConsumer; + +/** + * Синхронный обмен сообщениями с ассинхронным сервисом. + * Метод exchange() отправляет сообщение в очередь и дожидается ответа из другой очереди + */ +public class BiDirectionQueueExchanger> extends QueueConsumer implements InitializingBean, DisposableBean { + protected final Logger log = LoggerFactory.getLogger(getClass()); + protected final Object sync = new Object(); + protected String outQueue; + protected String inQueue; + protected Class listenClass; + protected long timeout; + + /** + * Синхронно-ассинхронный обмен сообщениями + * + * @param kafkaQueue + * @param kafkaProducer + * @param outQueue отправляет в очередь + * @param inQueue слушает очередь, ожидает ответов + * @param listenClass типы объектов из inQueue + * @param timeout - максимальное ожидание ответа, в миллисекундах, 0 - неограничено + */ + public BiDirectionQueueExchanger(Consumer kafkaQueue, Producer kafkaProducer, + String outQueue, + String inQueue, Class listenClass, + long timeout) { + super(kafkaQueue, kafkaProducer); + this.outQueue = outQueue; + this.inQueue = inQueue; + this.listenClass = listenClass; + this.timeout = timeout; + } + + /** + * Отправить сообщение message в outQueue и дождаться ответа из очереди inQueue + * + * @param message + * @return + * @throws InterruptedException + */ + public TOut exchange(TIn message) throws InterruptedException { + //todo impl BiDirectionQueueExchanger + return null; + } + + @Override + public void destroy() throws Exception { + log.debug("Listener {} for async exchange {}-{} close", this, outQueue, inQueue); + } + + @Override + public void afterPropertiesSet() throws Exception { + log.debug("Listener {} for async exchange {}-{} ready", this, outQueue, inQueue); + } +}