feat: supports mqtt (#22)

* feat: supports mqtt

* x

* x

* x

* x

* x

* x

Author: chenquan

Cargo.lock

@@ -45,3 +45,4 @@ reqwest = { version = "0.12", features = ["json"] }
 clap = { version = "4.3", features = ["derive"] }
 colored = "2.0"
 flume = "=0.10"
+rumqttc = "0.24.0"

Cargo.toml

@@ -0,0 +1,43 @@
+# 流式SQL处理示例配置
+
+streams:
+  - input:
+      type: "mqtt"
+      # 内存输入配置，提供一些测试消息
+      host: "broker.emqx.io"
+      port: 1883
+      qos: 1
+      client_id: "flow_xx_xdsfsfsfsf"
+      topics: ["flow_xx/#"]
+
+#  - input:
+#      type: "memory"
+#      # 内存输入配置，提供一些测试消息
+#      messages:
+#        - '{ "timestamp": 1625000000000, "value": 10, "sensor": "temp_1" }'
+#        - '{ "timestamp": 1625000001000, "value": 15, "sensor": "temp_1" }'
+#        - '{ "timestamp": 1625000002000, "value": 12, "sensor": "temp_1" }'
+
+
+    pipeline:
+      thread_num: 4
+      processors:
+        - type: "sql"
+          # SQL查询语句，支持标准SQL语法
+          query: "SELECT * ,cast(value as string) as tx FROM flow WHERE value > 10"
+          # 表名（用于SQL查询中引用）
+#          table_name: "events"
+
+
+    output:
+##      type: "stdout"
+#      type: "file"
+#      append: false
+#      path: "examples/output.txt"
+      type: "mqtt"
+      # 内存输入配置，提供一些测试消息
+      host: "broker.emqx.io"
+      port: 1883
+      qos: 1
+      client_id: "flow_xx_xdsfsfsfsfsdsd"
+      topic: "flow_xx_y"

examples/mqtt_example.yaml

@@ -73,9 +73,6 @@ impl Input for MemoryInput {
         if let Some(msg) = msg_option {
             Ok(msg)
         } else {
-            // 如果队列为空，则等待一段时间后返回错误
-            // 在实际应用中，这里可能需要实现更复杂的等待机制
-            tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
             Err(Error::Done)
         }
     }
@@ -85,7 +82,7 @@ impl Input for MemoryInput {
         Ok(())
     }
 
-  async  fn close(&self) -> Result<(), Error> {
+    async fn close(&self) -> Result<(), Error> {
         self.connected.store(false, std::sync::atomic::Ordering::SeqCst);
         Ok(())
     }

src/input/memory.rs

@@ -12,6 +12,7 @@ pub mod file;
 pub mod http;
 pub mod kafka;
 pub mod memory;
+pub mod mqtt;
 
 /// 输入组件的特征接口
 #[async_trait]
@@ -77,6 +78,7 @@ pub enum InputConfig {
     Http(http::HttpInputConfig),
     // Kafka(kafka::KafkaInputConfig),
     Memory(memory::MemoryInputConfig),
+    Mqtt(mqtt::MqttInputConfig),
 }
 
 impl InputConfig {
@@ -87,6 +89,7 @@ impl InputConfig {
             InputConfig::Http(config) => Ok(Arc::new(http::HttpInput::new(config)?)),
             // InputConfig::Kafka(config) => Ok(Arc::new(kafka::KafkaInput::new(config)?)),
             InputConfig::Memory(config) => Ok(Arc::new(memory::MemoryInput::new(config)?)),
+            InputConfig::Mqtt(config) => Ok(Arc::new(mqtt::MqttInput::new(config)?)),
         }
     }
 }