mqtt通讯组件修改

manager/pom.xml

@@ -159,6 +159,11 @@
             <artifactId>mqtt-component</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>cc.iotkit</groupId>
+            <artifactId>converter</artifactId>
+        </dependency>
+
     </dependencies>
 
     <build>

manager/src/main/java/cc/iotkit/manager/config/KeycloakSecurityConfig.java

@@ -54,7 +54,7 @@ public class KeycloakSecurityConfig extends KeycloakWebSecurityConfigurerAdapter
         http
                 .authorizeRequests()
                 .antMatchers("/*.html", "/favicon.ico", "/v2/api-docs", "/webjars/**", "/swagger-resources/**", "/*.js").permitAll()
-                .antMatchers("/device_behaviour/**").permitAll()
+                .antMatchers("/protocol/**").permitAll()//todo for test
                 .antMatchers("/**/save*").hasRole("iot_write")
                 .antMatchers("/**/del*").hasRole("iot_write")
                 .antMatchers("/**/add*").hasRole("iot_write")

manager/src/main/java/cc/iotkit/manager/controller/ProtocolController.java

@@ -14,6 +14,7 @@ import cc.iotkit.model.UserInfo;
 import cc.iotkit.model.protocol.ProtocolGateway;
 import cc.iotkit.protocol.server.service.GatewayService;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.io.FileUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.data.domain.Page;
@@ -22,6 +23,8 @@ import org.springframework.data.domain.Sort;
 import org.springframework.web.bind.annotation.*;
 
 import javax.annotation.PostConstruct;
+import java.io.File;
+import java.io.IOException;
 import java.util.Optional;
 
 @Slf4j
@@ -130,13 +133,20 @@ public class ProtocolController {
         return new Paging<>(gateways.getTotalElements(), gateways.getContent());
     }
 
-    @PostConstruct
-    public void init() {
+    @GetMapping("/registerMqtt")
+    public void registerMqtt() throws IOException {
         MqttComponent component = new MqttComponent();
+        component.create("{\"port\":2883,\"ssl\":false}");
         ScriptConverter converter = new ScriptConverter();
-        converter.setScript("");
+        converter.setScript(FileUtils.readFileToString(new File("/Users/sjg/home/gitee/open-source/converter.js"), "UTF-8"));
         component.setConverter(converter);
         componentManager.register("123", component);
-        componentManager.start("123", "");
+        componentManager.start("123", FileUtils.readFileToString(new File("/Users/sjg/home/gitee/open-source/component.js"), "UTF-8"));
+    }
+
+    @GetMapping("/deregisterMqtt")
+    public void deregisterMqtt() {
+        componentManager.stop("123");
+        componentManager.deRegister("123");
     }
 }

protocol-gateway/component-server/src/main/java/cc/iotkit/comps/MessageHandler.java

@@ -1,5 +1,6 @@
 package cc.iotkit.comps;
 
+import cc.iotkit.common.exception.BizException;
 import cc.iotkit.comp.IMessageHandler;
 import cc.iotkit.comp.model.DeviceMessage;
 import cc.iotkit.comps.model.AuthInfo;
@@ -23,7 +24,7 @@ import java.util.Map;
 public class MessageHandler implements IMessageHandler {
     private final NashornScriptEngine engine = (NashornScriptEngine) (new ScriptEngineManager()).getEngineByName("nashorn");
 
-    private final String script;
+    private final Object scriptObj;
 
     private final IConverter converter;
 
@@ -32,10 +33,9 @@ public class MessageHandler implements IMessageHandler {
     @SneakyThrows
     public MessageHandler(String script, IConverter converter,
                           DeviceBehaviourService deviceBehaviourService) {
-        this.script = script;
         this.converter = converter;
         this.deviceBehaviourService = deviceBehaviourService;
-        engine.eval(script);
+        scriptObj = engine.eval(script);
     }
 
     public void register(Map<String, Object> head, String msg) {
@@ -49,7 +49,7 @@ public class MessageHandler implements IMessageHandler {
 
     public void onReceive(Map<String, Object> head, String type, String msg) {
         try {
-            ScriptObjectMirror result = (ScriptObjectMirror) engine.invokeFunction("onReceive", head, type, msg);
+            ScriptObjectMirror result = (ScriptObjectMirror) engine.invokeMethod(scriptObj, "onReceive", head, type, msg);
             Object rstType = result.get("type");
             if (rstType == null) {
                 return;
@@ -57,7 +57,7 @@ public class MessageHandler implements IMessageHandler {
             //取脚本执行后返回的数据
             Object data = result.get("data");
             if (!(data instanceof Map)) {
-                return;
+                throw new BizException("script result data is incorrect");
             }
             Map<String, Object> dataMap = (Map) data;
 
@@ -83,18 +83,20 @@ public class MessageHandler implements IMessageHandler {
                 doReport(message);
             }
 
+        } catch (BizException e) {
+            throw e;
         } catch (Throwable e) {
-            log.error("onReceive error", e);
+            throw new BizException("receive component message error", e);
         }
     }
 
     private void doRegister(RegisterInfo reg) throws ScriptException, NoSuchMethodException {
         try {
             deviceBehaviourService.register(reg);
-            engine.invokeFunction("onRegistered", reg, true);
+            engine.invokeMethod(scriptObj, "onRegistered", reg, "true");
         } catch (Throwable e) {
             log.error("register error", e);
-            engine.invokeFunction("onRegistered", reg, false);
+            engine.invokeMethod(scriptObj, "onRegistered", reg, "false");
         }
     }
 
@@ -104,10 +106,10 @@ public class MessageHandler implements IMessageHandler {
                     auth.getDeviceName(),
                     auth.getProductSecret(),
                     auth.getDeviceSecret());
-            engine.invokeFunction("onAuthed", auth, true);
+            engine.invokeMethod(scriptObj, "onAuthed", auth, true);
         } catch (Throwable e) {
             log.error("device auth error", e);
-            engine.invokeFunction("onAuthed", auth, false);
+            engine.invokeMethod(scriptObj, "onAuthed", auth, false);
         }
     }
 
@@ -128,5 +130,4 @@ public class MessageHandler implements IMessageHandler {
             log.error("report device message error", e);
         }
     }
-
 }

protocol-gateway/component-server/src/main/java/cc/iotkit/comps/service/DeviceBehaviourService.java

@@ -104,10 +104,11 @@ public class DeviceBehaviourService {
         } else {
             //不存在,注册新设备
             device = new DeviceInfo();
+            device.setId(newDeviceId(info.getDeviceName()));
             device.setParentId(parentId);
             device.setUid(uid);
-            device.setDeviceId(newDeviceId(info.getDeviceName()));
-            device.setProductKey(info.getProductKey());
+            device.setDeviceId(device.getId());
+            device.setProductKey(pk);
             device.setDeviceName(info.getDeviceName());
             device.setTag(info.getTag());
             device.setState(new DeviceInfo.State(false, null, null));

protocol-gateway/mqtt-component/src/main/java/cc/iotkit/comp/mqtt/MqttComponent.java

@@ -3,8 +3,10 @@ package cc.iotkit.comp.mqtt;
 import cc.iotkit.common.exception.BizException;
 import cc.iotkit.common.utils.JsonUtil;
 import cc.iotkit.comp.AbstractComponent;
+import cc.iotkit.comp.IMessageHandler;
 import io.vertx.core.Future;
 import io.vertx.core.Vertx;
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 
 import java.util.concurrent.CountDownLatch;
@@ -13,18 +15,21 @@ import java.util.concurrent.CountDownLatch;
 public class MqttComponent extends AbstractComponent {
 
     private Vertx vertx;
-    private final CountDownLatch countDownLatch = new CountDownLatch(1);
+    private CountDownLatch countDownLatch;
     private String deployedId;
-    private MqttConfig mqttConfig;
+    private MqttVerticle mqttVerticle;
 
     public void create(String config) {
         vertx = Vertx.vertx();
-        mqttConfig = JsonUtil.parse(config, MqttConfig.class);
+        MqttConfig mqttConfig = JsonUtil.parse(config, MqttConfig.class);
+        mqttVerticle = new MqttVerticle(mqttConfig);
     }
 
     public void start() {
         try {
-            Future<String> future = vertx.deployVerticle(new MqttVerticle(mqttConfig, getHandler()));
+            mqttVerticle.setExecutor(getHandler());
+            countDownLatch = new CountDownLatch(1);
+            Future<String> future = vertx.deployVerticle(mqttVerticle);
             future.onSuccess((s -> {
                 deployedId = s;
                 countDownLatch.countDown();
@@ -40,13 +45,14 @@ public class MqttComponent extends AbstractComponent {
         }
     }
 
+    @SneakyThrows
     public void stop() {
+        mqttVerticle.stop();
         Future<Void> future = vertx.undeploy(deployedId);
         future.onSuccess(unused -> log.info("stop mqtt component success"));
     }
 
     public void destroy() {
-        vertx.close();
     }
 
 }

protocol-gateway/mqtt-component/src/main/java/cc/iotkit/comp/mqtt/MqttVerticle.java

@@ -26,10 +26,13 @@ public class MqttVerticle extends AbstractVerticle {
 
     private final MqttConfig config;
 
-    private final IMessageHandler executor;
+    private IMessageHandler executor;
 
-    public MqttVerticle(MqttConfig config, IMessageHandler executor) {
+    public MqttVerticle(MqttConfig config) {
         this.config = config;
+    }
+
+    public void setExecutor(IMessageHandler executor) {
         this.executor = executor;
     }
 
@@ -62,6 +65,7 @@ public class MqttVerticle extends AbstractVerticle {
             } catch (Throwable e) {
                 log.error("auth failed", e);
                 endpoint.reject(MqttConnectReturnCode.CONNECTION_REFUSED_NOT_AUTHORIZED);
+                return;
             }
 
             log.info("MQTT client keep alive timeout = {} ", endpoint.keepAliveTimeSeconds());

protocol-gateway/mqtt-component/src/main/resources/component.js

@@ -0,0 +1,84 @@
+new (function () {
+    !function(n){"use strict";function d(n,t){var r=(65535&n)+(65535&t);return(n>>16)+(t>>16)+(r>>16)<<16|65535&r}function f(n,t,r,e,o,u){return d((u=d(d(t,n),d(e,u)))<<o|u>>>32-o,r)}function l(n,t,r,e,o,u,c){return f(t&r|~t&e,n,t,o,u,c)}function g(n,t,r,e,o,u,c){return f(t&e|r&~e,n,t,o,u,c)}function v(n,t,r,e,o,u,c){return f(t^r^e,n,t,o,u,c)}function m(n,t,r,e,o,u,c){return f(r^(t|~e),n,t,o,u,c)}function c(n,t){var r,e,o,u;n[t>>5]|=128<<t%32,n[14+(t+64>>>9<<4)]=t;for(var c=1732584193,f=-271733879,i=-1732584194,a=271733878,h=0;h<n.length;h+=16)c=l(r=c,e=f,o=i,u=a,n[h],7,-680876936),a=l(a,c,f,i,n[h+1],12,-389564586),i=l(i,a,c,f,n[h+2],17,606105819),f=l(f,i,a,c,n[h+3],22,-1044525330),c=l(c,f,i,a,n[h+4],7,-176418897),a=l(a,c,f,i,n[h+5],12,1200080426),i=l(i,a,c,f,n[h+6],17,-1473231341),f=l(f,i,a,c,n[h+7],22,-45705983),c=l(c,f,i,a,n[h+8],7,1770035416),a=l(a,c,f,i,n[h+9],12,-1958414417),i=l(i,a,c,f,n[h+10],17,-42063),f=l(f,i,a,c,n[h+11],22,-1990404162),c=l(c,f,i,a,n[h+12],7,1804603682),a=l(a,c,f,i,n[h+13],12,-40341101),i=l(i,a,c,f,n[h+14],17,-1502002290),c=g(c,f=l(f,i,a,c,n[h+15],22,1236535329),i,a,n[h+1],5,-165796510),a=g(a,c,f,i,n[h+6],9,-1069501632),i=g(i,a,c,f,n[h+11],14,643717713),f=g(f,i,a,c,n[h],20,-373897302),c=g(c,f,i,a,n[h+5],5,-701558691),a=g(a,c,f,i,n[h+10],9,38016083),i=g(i,a,c,f,n[h+15],14,-660478335),f=g(f,i,a,c,n[h+4],20,-405537848),c=g(c,f,i,a,n[h+9],5,568446438),a=g(a,c,f,i,n[h+14],9,-1019803690),i=g(i,a,c,f,n[h+3],14,-187363961),f=g(f,i,a,c,n[h+8],20,1163531501),c=g(c,f,i,a,n[h+13],5,-1444681467),a=g(a,c,f,i,n[h+2],9,-51403784),i=g(i,a,c,f,n[h+7],14,1735328473),c=v(c,f=g(f,i,a,c,n[h+12],20,-1926607734),i,a,n[h+5],4,-378558),a=v(a,c,f,i,n[h+8],11,-2022574463),i=v(i,a,c,f,n[h+11],16,1839030562),f=v(f,i,a,c,n[h+14],23,-35309556),c=v(c,f,i,a,n[h+1],4,-1530992060),a=v(a,c,f,i,n[h+4],11,1272893353),i=v(i,a,c,f,n[h+7],16,-155497632),f=v(f,i,a,c,n[h+10],23,-1094730640),c=v(c,f,i,a,n[h+13],4,681279174),a=v(a,c,f,i,n[h],11,-358537222),i=v(i,a,c,f,n[h+3],16,-722521979),f=v(f,i,a,c,n[h+6],23,76029189),c=v(c,f,i,a,n[h+9],4,-640364487),a=v(a,c,f,i,n[h+12],11,-421815835),i=v(i,a,c,f,n[h+15],16,530742520),c=m(c,f=v(f,i,a,c,n[h+2],23,-995338651),i,a,n[h],6,-198630844),a=m(a,c,f,i,n[h+7],10,1126891415),i=m(i,a,c,f,n[h+14],15,-1416354905),f=m(f,i,a,c,n[h+5],21,-57434055),c=m(c,f,i,a,n[h+12],6,1700485571),a=m(a,c,f,i,n[h+3],10,-1894986606),i=m(i,a,c,f,n[h+10],15,-1051523),f=m(f,i,a,c,n[h+1],21,-2054922799),c=m(c,f,i,a,n[h+8],6,1873313359),a=m(a,c,f,i,n[h+15],10,-30611744),i=m(i,a,c,f,n[h+6],15,-1560198380),f=m(f,i,a,c,n[h+13],21,1309151649),c=m(c,f,i,a,n[h+4],6,-145523070),a=m(a,c,f,i,n[h+11],10,-1120210379),i=m(i,a,c,f,n[h+2],15,718787259),f=m(f,i,a,c,n[h+9],21,-343485551),c=d(c,r),f=d(f,e),i=d(i,o),a=d(a,u);return[c,f,i,a]}function i(n){for(var t="",r=32*n.length,e=0;e<r;e+=8)t+=String.fromCharCode(n[e>>5]>>>e%32&255);return t}function a(n){var t=[];for(t[(n.length>>2)-1]=void 0,e=0;e<t.length;e+=1)t[e]=0;for(var r=8*n.length,e=0;e<r;e+=8)t[e>>5]|=(255&n.charCodeAt(e/8))<<e%32;return t}function e(n){for(var t,r="0123456789abcdef",e="",o=0;o<n.length;o+=1)t=n.charCodeAt(o),e+=r.charAt(t>>>4&15)+r.charAt(15&t);return e}function r(n){return unescape(encodeURIComponent(n))}function o(n){return i(c(a(n=r(n)),8*n.length))}function u(n,t){return function(n,t){var r,e=a(n),o=[],u=[];for(o[15]=u[15]=void 0,16<e.length&&(e=c(e,8*n.length)),r=0;r<16;r+=1)o[r]=909522486^e[r],u[r]=1549556828^e[r];return t=c(o.concat(a(t)),512+8*t.length),i(c(u.concat(t),640))}(r(n),r(t))}function t(n,t,r){return t?r?u(t,n):e(u(t,n)):r?o(n):e(o(n))}"function"==typeof define&&define.amd?define(function(){return t}):"object"==typeof module&&module.exports?module.exports=t:n.md5=t}(this);
+    var md5=this.md5;
+
+    var registered={};
+
+    function register(payload){
+        var auth= JSON.parse(payload);
+        var arr= auth.clientid.split("_");
+        if(arr.length<3){
+            throw new Error("incorrect clientid");
+        }
+
+        var pk=arr[0];
+        var dn=arr[1];
+        var model=arr[2];
+        var pwd=md5("xdkKUymrEGSCYWswqCvSPyRSFvH5j7CU"+auth.clientid);
+        if(pwd!=auth.password){
+            throw new Error("incorrect password");
+        }
+        return {
+            type:"register",
+            data:{
+                productKey:pk,
+                deviceName:dn,
+                model:model
+            }
+        };
+    }
+
+    function online(topic){
+        var arr= topic.split('/');
+        if(arr.length<6){
+            throw new Error("incorrect topic")
+        }
+
+        var pk=arr[2];
+        var dn=arr[3];
+        return {
+            type:"state",
+            data:{
+                productKey:pk,
+                deviceName:dn,
+                state:"online"
+            }
+        }
+    }
+
+    //必须提供onReceive方法
+    this.onReceive=function(head,type,payload){
+        if(type=='auth'){
+            return register(payload);
+        }
+
+        if(type=='subscribe'){
+            return online(payload);
+        }
+
+        var topic=head['topic'];
+        if(!topic){
+            throw new Error("topic is blank")
+        }
+
+        var arr= topic.split('/');
+        if(arr.length<6){
+            throw new Error("incorrect topic")
+        }
+
+        var pk=arr[2];
+        var dn=arr[3];
+        return {
+            type:"report",
+            data:{
+                productKey:pk,
+                deviceName:dn,
+                content:payload
+            }
+        }
+    }
+
+    this.onRegistered=function(regInfo,result){
+    }
+})()
+  

protocol-gateway/mqtt-component/src/main/resources/converter.js

@@ -0,0 +1,51 @@
+new (function () {
+    this.decode = function (msg) {
+      //对msg进行解析，并返回物模型数据
+      var mqttMsg = JSON.parse(msg.content);
+      var topic = mqttMsg.topic;
+      var payload = mqttMsg.payload;
+  
+      if (topic.endsWith("/property/post")) {
+        //属性上报
+        return {
+          mid: msg.mid,
+          productKey: msg.productKey, //可根据消息内容判断填写不同产品
+          deviceName: msg.deviceName,
+          type:"property",
+          identifier: "report", //属性上报
+          occur: new Date().getTime(), //时间戳，设备上的事件或数据产生的本地时间
+          time: new Date().getTime(), //时间戳，消息上报时间
+          data: payload.params,
+        };
+      } else if (topic.indexOf("/event/") > 0) {
+        var identifier = topic.substring(topic.lastIndexOf("/") + 1);
+        //事件上报
+        return {
+          mid: msg.mid,
+          productKey: msg.productKey,
+          deviceName: msg.deviceName,
+          type:"event",
+          identifier: identifier,
+          occur: new Date().getTime(),
+          time: new Date().getTime(),
+          data: payload.params,
+        };
+      } else if (topic.endsWith("_reply")) {
+        var identifier = topic.substring(topic.lastIndexOf("/") + 1);
+        //服务回复
+        return {
+          mid: msg.mid,
+          productKey: msg.productKey,
+          deviceName: msg.deviceName,
+          type:"service",
+          identifier: identifier.replace("_reply", "Reply"),
+          occur: new Date().getTime(),
+          time: new Date().getTime(),
+          code: payload.code,
+          data: payload.data,
+        };
+      }
+      return null;
+    };
+  })()
+