规则引擎传输

Author: 1ssqq1lxr

smqtt-common/src/main/java/io/github/quickmsg/common/rule/source/SourceBean.java

@@ -39,7 +39,7 @@ public interface SourceBean {
      *
      * @param object {@link Map}
      */
-    void transmit(Map<String, Object> object);
+    void transmit(Object object);
 
     /**
      * 关闭资源

smqtt-rule/smqtt-rule-engine/src/main/java/io/github/quickmsg/rule/node/TransmitRuleNode.java

@@ -34,22 +34,18 @@ public TransmitRuleNode(Source source, String script) {
     @Override
     public void execute(ContextView contextView) {
         HeapMqttMessage heapMqttMessage = contextView.get(HeapMqttMessage.class);
-        Map<String, Object> param;
+        Object param;
         if (script != null) {
-            Object obj = triggerScript(script, context -> heapMqttMessage.getKeyMap().forEach(context::set));
-            param = JacksonUtil.json2Map(obj.toString(), String.class, Object.class);
-
+            param = triggerScript(script, context -> heapMqttMessage.getKeyMap().forEach(context::set));
         } else {
             param = heapMqttMessage.getKeyMap();
         }
         SourceBean sourceBean = SourceManager.getSourceBean(source);
-        if(sourceBean == null){
-            log.warn("[ please set source {}]",source);
-        }
-        else{
+        if (sourceBean != null) {
             sourceBean.transmit(param);
+            executeNext(contextView);
+
         }
-        executeNext(contextView);
     }
 
     @Override

smqtt-rule/smqtt-rule-source/smqtt-rule-source-db/src/main/java/io/github/quickmsg/source/db/DbSourceBean.java

@@ -57,10 +57,10 @@ public Boolean bootstrap(Map<String, Object> sourceParam) {
      * @param object 对象
      */
     @Override
-    public void transmit(Map<String, Object> object) {
+    public void transmit(Object object) {
         try (Connection connection = HikariCPConnectionProvider.singleTon().getConnection()) {
             DSLContext dslContext = DSL.using(connection);
-            dslContext.execute(object.get("sql").toString());
+            dslContext.execute(object.toString());
         } catch (Exception e) {
             log.error("execute sql error", e);
         }

smqtt-rule/smqtt-rule-source/smqtt-rule-source-http/src/main/java/io/github/quickmsg/http/HttpSourceBean.java

@@ -43,14 +43,11 @@ public Boolean bootstrap(Map<String, Object> sourceParam) {
 
 
     @Override
-    public void transmit(Map<String, Object> object) {
-        if (httpParam.getAdditions() != null && httpParam.getAdditions().size() > 0) {
-            httpParam.getAdditions().forEach(object::put);
-        }
+    public void transmit(Object object) {
         httpClient
                 .post()
                 .uri(httpParam.getUrl())
-                .send(Mono.just(PooledByteBufAllocator.DEFAULT.directBuffer().writeBytes(JacksonUtil.bean2Json(object).getBytes())))
+                .send(Mono.just(PooledByteBufAllocator.DEFAULT.directBuffer().writeBytes(JacksonUtil.dynamicJson(object).getBytes())))
                 .response()
                 .log()
                 .subscribe();

smqtt-rule/smqtt-rule-source/smqtt-rule-source-kafka/src/main/java/io/github/quickmsg/source/mqtt/KafkaSourceBean.java

@@ -59,11 +59,9 @@ public Boolean bootstrap(Map<String, Object> sourceParam) {
      * @param object 对象
      */
     @Override
-    public void transmit(Map<String, Object> object) {
-        String json = JacksonUtil.bean2Json(object);
-        log.info("kafka send msg {}", json);
+    public void transmit(Object object) {
         if (producer != null) {
-            ProducerRecord<String, Object> record = new ProducerRecord<>(topic, json);
+            ProducerRecord<String, Object> record = new ProducerRecord<>(topic, JacksonUtil.dynamicJson(object));
             producer.send(record);
         }
     }

smqtt-rule/smqtt-rule-source/smqtt-rule-source-mqtt/src/main/java/io/github/quickmsg/source/mqtt/MqttSourceBean.java

@@ -110,10 +110,11 @@ public Boolean bootstrap(Map<String, Object> sourceParam) {
     /**
      * 转发数据
      *
-     * @param object 对象
+     * @param param 对象
      */
     @Override
-    public void transmit(Map<String, Object> object) {
+    public void transmit(Object param) {
+        Map<String,Object> object = (Map<String,Object>)param;
         String topic = (String) object.get("topic");
         Object msg = object.get("msg");
         String bytes = msg instanceof Map ? JacksonUtil.map2Json((Map<? extends Object, ? extends Object>) msg) : msg.toString();

smqtt-rule/smqtt-rule-source/smqtt-rule-source-rabbitmq/src/main/java/io/github/quickmsg/source/rabbitmq/RabbitmqSourceBean.java

@@ -72,13 +72,8 @@ public Boolean bootstrap(Map<String, Object> sourceParam) {
      * @param object 对象
      */
     @Override
-    public void transmit(Map<String, Object> object) {
-        String json = JacksonUtil.bean2Json(object);
-        log.info("transmit={}", json);
-        // String clientId = (String) object.get("clientIdentifier");
-        // String topic = (String) object.get("topic");
-        // String msg = (String) object.get("msg");
-        corePublish(queueName, json);
+    public void transmit(Object object) {
+        corePublish(queueName, JacksonUtil.dynamicJson(object));
     }
 
     /**

smqtt-rule/smqtt-rule-source/smqtt-rule-source-rocketmq/src/main/java/io/github/quickmsg/source/rocketmq/RocketmqSourceBean.java

@@ -65,8 +65,8 @@ public Boolean bootstrap(Map<String, Object> sourceParam) {
      * @param object 对象
      */
     @Override
-    public void transmit(Map<String, Object> object) {
-        String json = JacksonUtil.bean2Json(object);
+    public void transmit(Object object) {
+        String json = JacksonUtil.dynamicJson(object);
         if (producer != null) {
             Message message = new Message(topic, tags, json.getBytes());
             //发送消息v