feat: Kafka RabbitMQ Source Connector支持exchange配置

cobolbaby · Dec 2, 2020 · 5282885 · 5282885
1 parent 603bda3
commit 5282885
Show file tree

Hide file tree

Showing 2 changed files with 33 additions and 5 deletions.
diff --git a/.../com/github/themeetgroup/kafka/connect/rabbitmq/source/RabbitMQSourceConnectorConfig.java b/.../com/github/themeetgroup/kafka/connect/rabbitmq/source/RabbitMQSourceConnectorConfig.java
@@ -34,19 +34,30 @@ public class RabbitMQSourceConnectorConfig extends CommonRabbitMQConnectorConfig
       "See `Channel.basicQos(int, boolean) <https://www.rabbitmq.com/releases/rabbitmq-java-client/current-javadoc/com/rabbitmq/client/Channel.html#basicQos-int-boolean->`_";
 
   public static final String PREFETCH_GLOBAL_CONF = "rabbitmq.prefetch.global";
-  public static final String PREFETCH_GLOBAL_DOC = "True if the settings should be applied to the entire channel rather " +
-      "than each consumer. " +
+  public static final String PREFETCH_GLOBAL_DOC = "True if the settings should be applied to the entire channel rather than each consumer. " +
       "See `Channel.basicQos(int, boolean) <https://www.rabbitmq.com/releases/rabbitmq-java-client/current-javadoc/com/rabbitmq/client/Channel.html#basicQos-int-boolean->`_";
 
   public static final String MESSAGE_CONVERTER_CLASSNAME_CONF = "message.converter";
   public static final String MESSAGE_CONVERTER_CLASSNAME_DOC = "Converter to compose the Kafka message. Optional, defaults to " +
       "com.github.themeetgroup.kafka.connect.rabbitmq.source.data.MessageConverter";
 
+  public static final String EXCHANGE_CONF = "rabbitmq.exchange";
+  public static final String EXCHANGE_CONF_DOC = "rabbitmq.exchange";
+
+  public static final String ROUTING_KEY_CONF = "rabbitmq.routing.key";
+  public static final String ROUTING_KEY_CONF_DOC = "rabbitmq.routing.key";
+
+  public static final String QUEUE_TTL_CONF = "rabbitmq.queue.ttl";
+  public static final String QUEUE_TTL_DOC = "rabbitmq.queue.ttl";
+
   public final String kafkaTopic;
   public final List<String> queues;
   public final int prefetchCount;
   public final boolean prefetchGlobal;
   public final String messageConverter;
+  public final String exchange;
+  public final String routingKey;
+  public final Integer ttl;
 
   public RabbitMQSourceConnectorConfig(Map<String, String> settings) {
     super(config(), settings);
@@ -56,6 +67,9 @@ public RabbitMQSourceConnectorConfig(Map<String, String> settings) {
     this.prefetchCount = this.getInt(PREFETCH_COUNT_CONF);
     this.prefetchGlobal = this.getBoolean(PREFETCH_GLOBAL_CONF);
     this.messageConverter = this.getString(MESSAGE_CONVERTER_CLASSNAME_CONF);
+    this.exchange = this.getString(EXCHANGE_CONF);
+    this.routingKey = this.getString(ROUTING_KEY_CONF);
+    this.ttl =  this.getInt(QUEUE_TTL_CONF);
   }
 
   public static ConfigDef config() {
@@ -64,6 +78,9 @@ public static ConfigDef config() {
         .define(PREFETCH_COUNT_CONF, ConfigDef.Type.INT, 0, ConfigDef.Importance.MEDIUM, PREFETCH_COUNT_DOC)
         .define(PREFETCH_GLOBAL_CONF, ConfigDef.Type.BOOLEAN, false, ConfigDef.Importance.MEDIUM, PREFETCH_GLOBAL_DOC)
         .define(QUEUE_CONF, ConfigDef.Type.LIST, ConfigDef.Importance.HIGH, QUEUE_DOC)
-        .define(MESSAGE_CONVERTER_CLASSNAME_CONF, ConfigDef.Type.STRING, ConfigDef.Importance.MEDIUM, MESSAGE_CONVERTER_CLASSNAME_DOC);
+        .define(MESSAGE_CONVERTER_CLASSNAME_CONF, ConfigDef.Type.STRING, ConfigDef.Importance.MEDIUM, MESSAGE_CONVERTER_CLASSNAME_DOC)
+        .define(EXCHANGE_CONF, ConfigDef.Type.STRING, ConfigDef.Importance.HIGH, EXCHANGE_CONF_DOC)
+        .define(ROUTING_KEY_CONF, ConfigDef.Type.STRING, ConfigDef.Importance.HIGH, ROUTING_KEY_CONF_DOC)
+        .define(QUEUE_TTL_CONF, ConfigDef.Type.INT, 0, ConfigDef.Importance.HIGH, QUEUE_TTL_DOC);
   }
 }
diff --git a/src/main/java/com/github/themeetgroup/kafka/connect/rabbitmq/source/RabbitMQSourceTask.java b/src/main/java/com/github/themeetgroup/kafka/connect/rabbitmq/source/RabbitMQSourceTask.java
@@ -29,6 +29,7 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeoutException;
@@ -39,7 +40,7 @@ public class RabbitMQSourceTask extends SourceTask {
   private SourceRecordConcurrentLinkedDeque records;
   private Channel channel;
   private Connection connection;
-
+  
   @Override
   public String version() {
     return VersionUtil.version(this.getClass());
@@ -67,9 +68,19 @@ public void start(Map<String, String> settings) {
     try {
       log.info("Creating Channel");
       this.channel = this.connection.createChannel();
+
+      log.info("Declaring exchange");
+      this.channel.exchangeDeclare(config.exchange, "topic");
+
       log.info("Declaring queues");
+      Map<String, Object> arguments = new HashMap<String, Object>();
+      Integer ttl = config.ttl;
+      if (ttl != 0) {
+        arguments.put("x-message-ttl", ttl);
+      }
       for (String queue : config.queues) {
-        this.channel.queueDeclare(queue, true, false, false, null);
+        this.channel.queueDeclare(queue, true, false, false, arguments);
+        this.channel.queueBind(queue, config.exchange, config.routingKey);
       }
     } catch (IOException e) {
       throw new ConnectException(e);