协议网关修改

.gitignore

@@ -25,3 +25,4 @@ hs_err_pid*
 target
 *.iml
 *.yml
+log

communication/component/pom.xml

@@ -1,16 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0"
-         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <parent>
-        <artifactId>iotkit-parent</artifactId>
-        <groupId>cc.iotkit</groupId>
-        <version>0.0.1-SNAPSHOT</version>
-        <relativePath>../../pom.xml</relativePath>
-    </parent>
-    <modelVersion>4.0.0</modelVersion>
-
-    <artifactId>component</artifactId>
-
-
-</project>

communication/pom.xml

@@ -1,15 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0"
-         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <parent>
-        <artifactId>iotkit-parent</artifactId>
-        <groupId>cc.iotkit</groupId>
-        <version>0.0.1-SNAPSHOT</version>
-    </parent>
-    <modelVersion>4.0.0</modelVersion>
-
-    <artifactId>communication</artifactId>
-
-
-</project>

manager/pom.xml

@@ -12,6 +12,7 @@
     <artifactId>manager</artifactId>
 
     <dependencies>
+
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-data-mongodb</artifactId>
@@ -145,7 +146,7 @@
 
         <dependency>
             <groupId>cc.iotkit</groupId>
-            <artifactId>gateway-server</artifactId>
+            <artifactId>protocol-server</artifactId>
         </dependency>
 
     </dependencies>

manager/src/main/java/cc/iotkit/manager/Application.java

@@ -1,9 +1,11 @@
 package cc.iotkit.manager;
 
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.cloud.openfeign.EnableFeignClients;
 
+@Slf4j
 @EnableFeignClients(basePackages = {"cc.iotkit.deviceapi"})
 @SpringBootApplication(scanBasePackages = {"cc.iotkit"})
 public class Application {
@@ -11,4 +13,5 @@ public class Application {
     public static void main(String[] args) {
         SpringApplication.run(Application.class, args);
     }
+
 }

pom.xml

@@ -12,9 +12,8 @@
         <module>dao</module>
         <module>tppa-server</module>
         <module>protocol-gateway</module>
-        <module>communication</module>
+        <module>protocol-gateway/mqtt-component</module>
-        <module>communication/mqtt-component</module>
+        <module>protocol-gateway/component</module>
-        <module>communication/component</module>
     </modules>
     <parent>
         <groupId>org.springframework.boot</groupId>
@@ -176,6 +175,24 @@
                 <version>2.6.0</version>
             </dependency>
 
+            <dependency>
+                <groupId>io.vertx</groupId>
+                <artifactId>vertx-core</artifactId>
+                <version>4.2.6</version>
+            </dependency>
+
+            <dependency>
+                <groupId>io.vertx</groupId>
+                <artifactId>vertx-web</artifactId>
+                <version>4.2.6</version>
+            </dependency>
+
+            <dependency>
+                <groupId>io.vertx</groupId>
+                <artifactId>vertx-mqtt</artifactId>
+                <version>4.2.6</version>
+            </dependency>
+
             <dependency>
                 <groupId>cc.iotkit</groupId>
                 <artifactId>model</artifactId>
@@ -220,7 +237,13 @@
 
             <dependency>
                 <groupId>cc.iotkit</groupId>
-                <artifactId>gateway-server</artifactId>
+                <artifactId>protocol-server</artifactId>
+                <version>${project.version}</version>
+            </dependency>
+
+            <dependency>
+                <groupId>cc.iotkit</groupId>
+                <artifactId>component</artifactId>
                 <version>${project.version}</version>
             </dependency>
 

protocol-gateway/component/pom.xml

@@ -10,11 +10,19 @@
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
-    <artifactId>mqtt-component</artifactId>
+    <artifactId>component</artifactId>
-
 
     <dependencies>
 
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>cc.iotkit</groupId>
+            <artifactId>common</artifactId>
+        </dependency>
 
     </dependencies>
 

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

@@ -0,0 +1,10 @@
+package cc.iotkit.comp;
+
+import lombok.Data;
+
+@Data
+public class AbstractComponent {
+
+    protected MessageHandler messageHandler;
+
+}

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

@@ -0,0 +1,15 @@
+package cc.iotkit.comp;
+
+public interface Component {
+
+    void create(String config);
+
+    void start();
+
+    void stop();
+
+    void destroy();
+
+    void setHandler(MessageHandler handler);
+
+}

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

@@ -0,0 +1,37 @@
+package cc.iotkit.comp;
+
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class ComponentManager {
+
+    private final Map<String, Component> components = new HashMap<>();
+
+    public void register(String id, Component component) {
+        components.put(id, component);
+    }
+
+    public void deRegister(String id) {
+        Component component = components.remove(id);
+        component.destroy();
+    }
+
+    public void start(String id, String script) {
+        Component component = components.get(id);
+        if (component == null) {
+            return;
+        }
+        component.setHandler(new MessageHandler(script));
+        component.start();
+    }
+
+    public void stop(String id) {
+        Component component = components.get(id);
+        if (component == null) {
+            return;
+        }
+        component.stop();
+    }
+
+}

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

@@ -0,0 +1,7 @@
+package cc.iotkit.comp;
+
+import lombok.Data;
+
+@Data
+public class Device {
+}

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

@@ -0,0 +1,61 @@
+package cc.iotkit.comp;
+
+import cc.iotkit.common.utils.JsonUtil;
+import cc.iotkit.comp.model.RegisterInfo;
+import jdk.nashorn.api.scripting.NashornScriptEngine;
+import jdk.nashorn.api.scripting.ScriptObjectMirror;
+import lombok.Data;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+import javax.script.ScriptEngineManager;
+import java.util.Map;
+
+@Slf4j
+@Data
+public class MessageHandler {
+    private final NashornScriptEngine engine = (NashornScriptEngine) (new ScriptEngineManager()).getEngineByName("nashorn");
+
+    private final String script;
+
+    @SneakyThrows
+    public MessageHandler(String script) {
+        this.script = script;
+        engine.eval(script);
+    }
+
+    public void register(Map<String, Object> head, String msg) {
+    }
+
+    public void auth(Map<String, Object> head, String msg) {
+    }
+
+    public void state(Map<String, Object> head, String msg) {
+    }
+
+    public void onReceive(Map<String, Object> head, String type, String msg) {
+        try {
+            ScriptObjectMirror obj = (ScriptObjectMirror) engine.invokeFunction("onReceive", head, type, msg);
+            Object rstType = obj.get("type");
+            if (rstType == null) {
+                return;
+            }
+            Object data = obj.get("data");
+
+            if ("register".equals(rstType)) {
+                RegisterInfo regInfo = getData(data, RegisterInfo.class);
+            } else if ("report".equals(rstType)) {
+
+            }
+
+        } catch (Throwable e) {
+            log.error("onReceive error", e);
+        }
+
+    }
+
+    private <T> T getData(Object data, Class<T> cls) {
+        return JsonUtil.parse(JsonUtil.toJsonString(data), cls);
+    }
+
+}

protocol-gateway/component/src/main/java/cc/iotkit/comp/model/RegisterInfo.java

@@ -0,0 +1,49 @@
+package cc.iotkit.comp.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 注册信息
+ */
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+@Builder
+public class RegisterInfo {
+
+    private String productKey;
+
+    private String deviceName;
+
+    private String model;
+
+    private Map<String, Object> tag;
+
+    private List<SubDevice> subDevices;
+
+    public RegisterInfo(String productKey, String deviceName, String model) {
+        this.productKey = productKey;
+        this.deviceName = deviceName;
+        this.model = model;
+    }
+
+    @Data
+    @NoArgsConstructor
+    @AllArgsConstructor
+    public static class SubDevice {
+
+        private String productKey;
+
+        private String deviceName;
+
+        private String model;
+
+        private Map<String, Object> tag;
+    }
+}

protocol-gateway/decode-function/src/main/java/cc/iotkit/protocol/function/DecodeFunction.java

@@ -14,7 +14,7 @@ import java.util.concurrent.ConcurrentHashMap;
 /**
  * 上行消息转换函数
  */
-public class UplinkTranslateFunction implements Function<DeviceMessage, ThingModelMessage> {
+public class DecodeFunction implements Function<DeviceMessage, ThingModelMessage> {
 
     private static final NashornScriptEngine engine = (NashornScriptEngine) (new ScriptEngineManager()).getEngineByName("nashorn");
     private static final Map<String, CompiledScript> compiledScripts = new ConcurrentHashMap<>();

protocol-gateway/mqtt-component/pom.xml

@@ -0,0 +1,49 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>iotkit-parent</artifactId>
+        <groupId>cc.iotkit</groupId>
+        <version>0.0.1-SNAPSHOT</version>
+        <relativePath>../../pom.xml</relativePath>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>mqtt-component</artifactId>
+
+    <dependencies>
+
+        <dependency>
+            <groupId>io.vertx</groupId>
+            <artifactId>vertx-core</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.vertx</groupId>
+            <artifactId>vertx-mqtt</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>cc.iotkit</groupId>
+            <artifactId>common</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>cc.iotkit</groupId>
+            <artifactId>component</artifactId>
+        </dependency>
+
+    </dependencies>
+
+</project>

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

@@ -0,0 +1,52 @@
+package cc.iotkit.comp.mqtt;
+
+import cc.iotkit.common.exception.BizException;
+import cc.iotkit.common.utils.JsonUtil;
+import cc.iotkit.comp.AbstractComponent;
+import io.vertx.core.Future;
+import io.vertx.core.Vertx;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.concurrent.CountDownLatch;
+
+@Slf4j
+public class MqttComponent extends AbstractComponent {
+
+    private Vertx vertx;
+    private final CountDownLatch countDownLatch = new CountDownLatch(1);
+    private String deployedId;
+    private MqttConfig mqttConfig;
+
+    public void create(String config) {
+        vertx = Vertx.vertx();
+        mqttConfig = JsonUtil.parse(config, MqttConfig.class);
+    }
+
+    public void start() {
+        try {
+            Future<String> future = vertx.deployVerticle(new MqttVerticle(mqttConfig, getMessageHandler()));
+            future.onSuccess((s -> {
+                deployedId = s;
+                countDownLatch.countDown();
+            }));
+            future.onFailure((e) -> {
+                countDownLatch.countDown();
+                log.error("start mqtt component failed", e);
+            });
+            countDownLatch.await();
+            future.succeeded();
+        } catch (Throwable e) {
+            throw new BizException("start mqtt component error", e);
+        }
+    }
+
+    public void 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/MqttConfig.java

@@ -0,0 +1,16 @@
+package cc.iotkit.comp.mqtt;
+
+import lombok.Data;
+
+@Data
+public class MqttConfig {
+
+    private int port;
+
+    private String sslKey;
+
+    private String sslCert;
+
+    private boolean ssl;
+
+}

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

@@ -0,0 +1,130 @@
+package cc.iotkit.comp.mqtt;
+
+import cc.iotkit.comp.MessageHandler;
+import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
+import io.netty.handler.codec.mqtt.MqttProperties;
+import io.netty.handler.codec.mqtt.MqttQoS;
+import io.vertx.core.AbstractVerticle;
+import io.vertx.core.net.PemKeyCertOptions;
+import io.vertx.mqtt.MqttAuth;
+import io.vertx.mqtt.MqttServer;
+import io.vertx.mqtt.MqttServerOptions;
+import io.vertx.mqtt.MqttTopicSubscription;
+import io.vertx.mqtt.messages.codes.MqttSubAckReasonCode;
+import lombok.extern.slf4j.Slf4j;
+
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@Slf4j
+public class MqttVerticle extends AbstractVerticle {
+
+    private MqttServer mqttServer;
+
+    private final MqttConfig config;
+
+    private final MessageHandler executor;
+
+    public MqttVerticle(MqttConfig config, MessageHandler executor) {
+        this.config = config;
+        this.executor = executor;
+    }
+
+    @Override
+    public void start() throws Exception {
+        MqttServerOptions options = new MqttServerOptions()
+                .setPort(config.getPort());
+        if (config.isSsl()) {
+            options = options.setSsl(true)
+                    .setKeyCertOptions(new PemKeyCertOptions()
+                            .setKeyPath(config.getSslKey())
+                            .setCertPath(config.getSslCert()));
+        }
+
+        mqttServer = MqttServer.create(vertx, options);
+        mqttServer.endpointHandler(endpoint -> {
+            log.info("MQTT client:{} request to connect, clean session = {}", endpoint.clientIdentifier(), endpoint.isCleanSession());
+
+            MqttAuth auth = endpoint.auth();
+            if (auth == null) {
+                return;
+            }
+
+            String authJson = auth.toJson()
+                    .put("clientid", endpoint.clientIdentifier()).toString();
+
+            log.info("MQTT client auth,username:{},password:{}", auth.getUsername(), auth.getPassword());
+            try {
+                executor.onReceive(new HashMap<>(), "auth", authJson);
+            } catch (Throwable e) {
+                log.error("auth failed", e);
+                endpoint.reject(MqttConnectReturnCode.CONNECTION_REFUSED_NOT_AUTHORIZED);
+            }
+
+            log.info("MQTT client keep alive timeout = {} ", endpoint.keepAliveTimeSeconds());
+
+            endpoint.accept(false);
+            endpoint.disconnectMessageHandler(disconnectMessage -> {
+                log.info("Received disconnect from client, reason code = {}", disconnectMessage.code());
+                executor.onReceive(new HashMap<>(), "disconnect", authJson);
+            }).subscribeHandler(subscribe -> {
+                List<MqttSubAckReasonCode> reasonCodes = new ArrayList<>();
+                for (MqttTopicSubscription s : subscribe.topicSubscriptions()) {
+                    log.info("Subscription for {},with QoS {}", s.topicName(), s.qualityOfService());
+                    try {
+                        executor.onReceive(new HashMap<>(), "subscribe", s.topicName());
+                        reasonCodes.add(MqttSubAckReasonCode.qosGranted(s.qualityOfService()));
+                    } catch (Throwable e) {
+                        log.error("subscribe failed,topic:" + s.topicName(), e);
+                        reasonCodes.add(MqttSubAckReasonCode.NOT_AUTHORIZED);
+                    }
+                }
+                // ack the subscriptions request
+                endpoint.subscribeAcknowledge(subscribe.messageId(), reasonCodes, MqttProperties.NO_PROPERTIES);
+
+            }).unsubscribeHandler(unsubscribe -> {
+                for (String t : unsubscribe.topics()) {
+                    log.info("Unsubscription for {}", t);
+                    try {
+                        executor.onReceive(new HashMap<>(), "unsubscribe", t);
+                    } catch (Throwable e) {
+                        log.error("unsubscribe failed,topic:" + t, e);
+                    }
+                }
+                // ack the subscriptions request
+                endpoint.unsubscribeAcknowledge(unsubscribe.messageId());
+            }).publishHandler(message -> {
+                String payload = message.payload().toString(Charset.defaultCharset());
+                log.info("Received message:{}, with QoS {}", payload,
+                        message.qosLevel());
+                try {
+                    Map<String, Object> head = new HashMap<>();
+                    head.put("topic", message.topicName());
+                    executor.onReceive(head, "", payload);
+                } catch (Throwable e) {
+                    log.error("handler message failed,topic:" + message.topicName(), e);
+                }
+
+                if (message.qosLevel() == MqttQoS.AT_LEAST_ONCE) {
+                    endpoint.publishAcknowledge(message.messageId());
+                } else if (message.qosLevel() == MqttQoS.EXACTLY_ONCE) {
+                    endpoint.publishReceived(message.messageId());
+                }
+            }).publishReleaseHandler(endpoint::publishComplete);
+        }).listen(ar -> {
+            if (ar.succeeded()) {
+                log.info("MQTT server is listening on port " + ar.result().actualPort());
+            } else {
+                log.error("Error on starting the server", ar.cause());
+            }
+        });
+    }
+
+    @Override
+    public void stop() throws Exception {
+        mqttServer.close(voidAsyncResult -> log.info("close mqtt server..."));
+    }
+}

protocol-gateway/pom.xml

@@ -13,9 +13,8 @@
     <packaging>pom</packaging>
     <modules>
         <module>gateway-client</module>
-        <module>gateway-server</module>
+        <module>protocol-server</module>
-        <module>gateway-server/fun-test</module>
+        <module>decode-function</module>
-        <module>protocol-function</module>
     </modules>
 
     <properties>

protocol-gateway/protocol-function/src/main/java/cc/iotkit/protocol/function/MessageDistributionFunction.java

@@ -1,52 +0,0 @@
-package cc.iotkit.protocol.function;
-
-import cc.iotkit.common.utils.JsonUtil;
-import jdk.nashorn.api.scripting.NashornScriptEngine;
-import org.apache.pulsar.functions.api.Context;
-import org.apache.pulsar.functions.api.Function;
-
-import javax.script.Bindings;
-import javax.script.CompiledScript;
-import javax.script.ScriptEngineManager;
-import javax.script.SimpleBindings;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Optional;
-import java.util.concurrent.ConcurrentHashMap;
-
-/**
- * 消息分发函数
- */
-public class MessageDistributionFunction implements Function<ThingModelMessage, ThingModelMessage> {
-
-    private static final NashornScriptEngine engine = (NashornScriptEngine) (new ScriptEngineManager()).getEngineByName("nashorn");
-    private static final Map<String, CompiledScript> compiledScripts = new ConcurrentHashMap<>();
-
-    @Override
-    public ThingModelMessage process(ThingModelMessage msg, Context context) throws Exception {
-        Optional<Object> optName = context.getUserConfigValue("name");
-        Optional<Object> optScript = context.getUserConfigValue("script");
-        if (!optName.isPresent() || !optScript.isPresent()) {
-            return null;
-        }
-
-        String name = optName.get().toString();
-        compiledScripts.putIfAbsent(name, engine.compile(optScript.get().toString()));
-
-        CompiledScript script = compiledScripts.get(name);
-        context.getLogger().debug(script.toString());
-
-        Map<String, Object> data = new HashMap<>();
-        data.putIfAbsent("msg", msg);
-        Bindings bindings = new SimpleBindings(data);
-        Object result = script.eval(bindings);
-
-        if (result == null) {
-            context.getLogger().error("translate msg failed:{}", JsonUtil.toJsonString(msg));
-            return null;
-        }
-        return JsonUtil.parse(JsonUtil.toJsonString(result), ThingModelMessage.class);
-    }
-
-
-}

protocol-gateway/protocol-server/pom.xml

@@ -9,7 +9,7 @@
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
-    <artifactId>gateway-server</artifactId>
+    <artifactId>protocol-server</artifactId>
 
     <properties>
         <maven.compiler.source>8</maven.compiler.source>

protocol-gateway/protocol-server/src/main/java/cc/iotkit/protocol/server/service/GatewayService.java

@@ -2,7 +2,7 @@ package cc.iotkit.protocol.server.service;
 
 import cc.iotkit.common.Constants;
 import cc.iotkit.common.utils.JsonUtil;
-import cc.iotkit.protocol.function.UplinkTranslateFunction;
+import cc.iotkit.protocol.function.DecodeFunction;
 import cc.iotkit.protocol.server.config.ServerConfig;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.client.admin.PulsarAdmin;
@@ -54,7 +54,7 @@ public class GatewayService {
 //        String functionClass = "cc.iotkit.protocol.function.UplinkTranslateFunction";
 //        String jarFile = "/Users/sjg/home/gitee/open-source/iotkit-parent/protocol-gateway/gateway-server/fun-test/target/fun-test-0.0.1-SNAPSHOT-jar-with-dependencies.jar";
 
-        String functionClass = UplinkTranslateFunction.class.getName();
+        String functionClass = DecodeFunction.class.getName();
         String functionName = functionClass.substring(functionClass.lastIndexOf(".") + 1) + "_" + gatewayId;
 
         FunctionConfig functionConfig = new FunctionConfig();
@@ -86,7 +86,7 @@ public class GatewayService {
 
     public void deleteFunction(String tenant, String gatewayId) throws PulsarClientException, PulsarAdminException {
         String namespace = "default";
-        String functionClass = UplinkTranslateFunction.class.getName();
+        String functionClass = DecodeFunction.class.getName();
         String functionName = functionClass.substring(functionClass.lastIndexOf(".") + 1) + "_" + gatewayId;
         PulsarAdmin pulsarAdmin = getPulsarAdmin();
         if (!pulsarAdmin.functions().getFunctions(tenant, namespace).contains(functionName)) {