CLS-259 util добавил ассинхронно-синхронный обмен сообщениями (интерфейс) BiDirectionQueueExchanger

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<TIn, TOut extends BaseRequest<?>> 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<TIn> listenClass;
+    protected long timeout;
+
+    /**
+     * Синхронно-ассинхронный обмен сообщениями
+     *
+     * @param kafkaQueue
+     * @param kafkaProducer
+     * @param outQueue      отправляет в очередь
+     * @param inQueue       слушает очередь, ожидает ответов
+     * @param listenClass   типы объектов из inQueue
+     * @param timeout - максимальное ожидание ответа, в миллисекундах, 0 - неограничено
+     */
+    public BiDirectionQueueExchanger(Consumer<String, Object> kafkaQueue, Producer<String, Object> kafkaProducer,
+                                     String outQueue,
+                                     String inQueue, Class<TIn> 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);
+    }
+}