emqx集成

common/src/main/java/cc/iotkit/common/Constants.java

@@ -160,4 +160,8 @@ public interface Constants {
          */
         String GET_DEVICE = "/device/{deviceId}";
     }
+
+    interface  MQTT {
+        String DEVICE_SUBSCRIBE_TOPIC = "^/sys/.+/.+/c/#$";
+    }
 }

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

@@ -19,6 +19,7 @@ public class Application {
         if (EmbeddedRedisConfig.embeddedEnable()) {
             EmbeddedRedisConfig.startEmbeddedRedisServer();
         }
+        System.setProperty("nashorn.args","--no-deprecation-warning");
 
         SpringApplication.run(Application.class, args);
     }

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

@@ -1,7 +1,10 @@
 package cc.iotkit.comps.config;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
 import lombok.Data;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
 import java.nio.file.Path;
@@ -26,4 +29,9 @@ public class ComponentConfig {
         return Paths.get(converterDir, conId)
                 .toAbsolutePath().normalize();
     }
+
+    @Bean("objectMapper")
+    public ObjectMapper myMapper() {
+        return new ObjectMapper().disable(SerializationFeature.FAIL_ON_EMPTY_BEANS);
+    }
 }

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

@@ -15,14 +15,13 @@ import cc.iotkit.model.product.Product;
 import cc.iotkit.model.product.ProductModel;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
-import org.apache.pulsar.client.api.Producer;
-import org.apache.pulsar.client.api.PulsarClient;
-import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.*;
 import org.apache.pulsar.client.impl.schema.JSONSchema;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
 import javax.annotation.PostConstruct;
+import java.nio.charset.StandardCharsets;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Optional;
@@ -46,7 +45,10 @@ public class DeviceBehaviourService {
 //    @Autowired
     private DeviceStateHolder deviceStateHolder;
 
-    private Producer<ThingModelMessage> deviceMessageProducer;
+    //旧实现，ThingModelMessage序列化失败
+    //private Producer<ThingModelMessage> deviceMessageProducer;
+
+    private Producer<byte[]> deviceMessageProducer;
 
     @PostConstruct
     public void init() throws PulsarClientException {
@@ -54,9 +56,16 @@ public class DeviceBehaviourService {
         PulsarClient client = PulsarClient.builder()
                 .serviceUrl(serverConfig.getPulsarBrokerUrl())
                 .build();
+        /**
+         旧实现，ThingModelMessage序列化失败
         deviceMessageProducer = client.newProducer(JSONSchema.of(ThingModelMessage.class))
                 .topic("persistent://iotkit/default/" + Constants.THING_MODEL_MESSAGE_TOPIC)
                 .create();
+         */
+
+        deviceMessageProducer = client.newProducer()
+                .topic("persistent://iotkit/default/" + Constants.THING_MODEL_MESSAGE_TOPIC)
+                .create();
     }
 
     public void register(RegisterInfo info) {
@@ -245,8 +254,18 @@ public class DeviceBehaviourService {
                 message.setTime(System.currentTimeMillis());
             }
             message.setDeviceId(device.getDeviceId());
-            deviceMessageProducer.send(message);
-        } catch (PulsarClientException e) {
+
+            // 旧实现，ThingModelMessage序列化失败
+            //deviceMessageProducer.send(message);
+
+            // 新实现，用JsonUtil.toJsonString序列化ThingModelMessage，解决 ThingModelMessage序列化失败的问题
+            TypedMessageBuilder<byte[]> builder = deviceMessageProducer.newMessage();
+            builder.value(JsonUtil.toJsonString(message).getBytes(StandardCharsets.UTF_8));
+            builder.send();
+
+
+        }
+        catch (PulsarClientException e) {
             log.error("send thing model message error", e);
         }
     }

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

@@ -0,0 +1,133 @@
+package cc.iotkit.comp.utils;
+
+import org.springframework.aop.framework.AopContext;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.NoSuchBeanDefinitionException;
+import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.stereotype.Component;
+
+/**
+ * spring工具类 方便在非spring管理环境中获取bean
+ */
+@Component
+public final class SpringUtils implements BeanFactoryPostProcessor, ApplicationContextAware
+{
+    /** Spring应用上下文环境 */
+    private static ConfigurableListableBeanFactory beanFactory;
+
+    private static ApplicationContext applicationContext;
+
+    @Override
+    public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException
+    {
+        SpringUtils.beanFactory = beanFactory;
+    }
+
+    @Override
+    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException
+    {
+        SpringUtils.applicationContext = applicationContext;
+    }
+
+    /**
+     * 获取对象
+     *
+     * @param name
+     * @return Object 一个以所给名字注册的bean的实例
+     * @throws org.springframework.beans.BeansException
+     *
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> T getBean(String name) throws BeansException
+    {
+        return (T) beanFactory.getBean(name);
+    }
+
+    /**
+     * 获取类型为requiredType的对象
+     *
+     * @param clz
+     * @return
+     * @throws org.springframework.beans.BeansException
+     *
+     */
+    public static <T> T getBean(Class<T> clz) throws BeansException
+    {
+        T result = (T) beanFactory.getBean(clz);
+        return result;
+    }
+
+    /**
+     * 如果BeanFactory包含一个与所给名称匹配的bean定义，则返回true
+     *
+     * @param name
+     * @return boolean
+     */
+    public static boolean containsBean(String name)
+    {
+        return beanFactory.containsBean(name);
+    }
+
+    /**
+     * 判断以给定名字注册的bean定义是一个singleton还是一个prototype。 如果与给定名字相应的bean定义没有被找到，将会抛出一个异常（NoSuchBeanDefinitionException）
+     *
+     * @param name
+     * @return boolean
+     * @throws org.springframework.beans.factory.NoSuchBeanDefinitionException
+     *
+     */
+    public static boolean isSingleton(String name) throws NoSuchBeanDefinitionException
+    {
+        return beanFactory.isSingleton(name);
+    }
+
+    /**
+     * @param name
+     * @return Class 注册对象的类型
+     * @throws org.springframework.beans.factory.NoSuchBeanDefinitionException
+     *
+     */
+    public static Class<?> getType(String name) throws NoSuchBeanDefinitionException
+    {
+        return beanFactory.getType(name);
+    }
+
+    /**
+     * 如果给定的bean名字在bean定义中有别名，则返回这些别名
+     *
+     * @param name
+     * @return
+     * @throws org.springframework.beans.factory.NoSuchBeanDefinitionException
+     *
+     */
+    public static String[] getAliases(String name) throws NoSuchBeanDefinitionException
+    {
+        return beanFactory.getAliases(name);
+    }
+
+    /**
+     * 获取aop代理对象
+     *
+     * @param invoker
+     * @return
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> T getAopProxy(T invoker)
+    {
+        return (T) AopContext.currentProxy();
+    }
+
+    /**
+     * 获取当前的环境配置，无配置返回null
+     *
+     * @return 当前的环境配置
+     */
+    public static String[] getActiveProfiles()
+    {
+        return applicationContext.getEnvironment().getActiveProfiles();
+    }
+
+}

protocol-gateway/emqx-component/dependency-reduced-pom.xml

@@ -26,6 +26,21 @@
               <include>io.vertx:vertx-core</include>
               <include>io.vertx:vertx-web-proxy</include>
               <include>io.vertx:vertx-mqtt</include>
+              <include>io.vertx:vertx-web</include>
+              <include>io.vertx:vertx-http-proxy</include>
+              <include>org.luaj:luaj-jse</include>
+              <include>io.netty:netty-common</include>
+              <include>io.netty:netty-transport</include>
+              <include>io.netty:netty-handler</include>
+              <include>io.netty:netty-resolver</include>
+              <include>io.netty:netty-buffer</include>
+              <include>io.netty:netty-handler</include>
+              <include>io.netty:netty-proxy</include>
+              <include>io.netty:netty-codec</include>
+              <include>io.netty:netty-codec-mqtt</include>
+              <include>io.netty:netty-codec-dns</include>
+              <include>io.netty:netty-resolver-dns</include>
+              <include>io.netty:netty-tcnative-boringssl-static</include>
             </includes>
           </artifactSet>
         </configuration>
@@ -76,5 +91,17 @@
       <version>0.2.0-SNAPSHOT</version>
       <scope>compile</scope>
     </dependency>
+    <dependency>
+      <groupId>org.luaj</groupId>
+      <artifactId>luaj-jse</artifactId>
+      <version>3.0.1</version>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>cc.iotkit</groupId>
+      <artifactId>dao</artifactId>
+      <version>0.2.0-SNAPSHOT</version>
+      <scope>compile</scope>
+    </dependency>
   </dependencies>
 </project>

protocol-gateway/emqx-component/pom.xml

@@ -43,6 +43,15 @@
             <artifactId>component</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.luaj</groupId>
+            <artifactId>luaj-jse</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>cc.iotkit</groupId>
+            <artifactId>dao</artifactId>
+        </dependency>
     </dependencies>
 
     <build>
@@ -62,9 +71,31 @@
                 <configuration>
                     <artifactSet>
                         <includes>
+<!--                            <include>io.vertx:vertx-core</include>-->
+<!--                            <include>io.vertx:vertx-web-proxy</include>-->
+<!--                            <include>io.vertx:vertx-mqtt</include>-->
+
                             <include>io.vertx:vertx-core</include>
                             <include>io.vertx:vertx-web-proxy</include>
                             <include>io.vertx:vertx-mqtt</include>
+                            <include>io.vertx:vertx-web</include>
+                            <include>io.vertx:vertx-http-proxy</include>
+
+                            <include>org.luaj:luaj-jse</include>
+
+                            <include>io.netty:netty-common</include>
+                            <include>io.netty:netty-transport</include>
+                            <include>io.netty:netty-handler</include>
+                            <include>io.netty:netty-resolver</include>
+                            <include>io.netty:netty-buffer</include>
+                            <include>io.netty:netty-handler</include>
+                            <include>io.netty:netty-proxy</include>
+                            <include>io.netty:netty-codec</include>
+                            <include>io.netty:netty-codec-mqtt</include>
+                            <include>io.netty:netty-codec-dns</include>
+                            <include>io.netty:netty-resolver-dns</include>
+
+                            <include>io.netty:netty-tcnative-boringssl-static</include>
                         </includes>
                     </artifactSet>
                 </configuration>

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

@@ -1,11 +1,16 @@
 package cc.iotkit.comp.emqx;
 
+import cc.iotkit.common.Constants;
+import cc.iotkit.common.utils.JsonUtil;
 import cc.iotkit.comp.IMessageHandler;
+import cc.iotkit.comp.utils.SpringUtils;
+import cc.iotkit.dao.DeviceRepository;
 import io.vertx.core.AbstractVerticle;
 import io.vertx.core.http.HttpMethod;
 import io.vertx.core.http.HttpServer;
 import io.vertx.core.json.JsonObject;
 import io.vertx.ext.web.Router;
+import io.vertx.ext.web.handler.BodyHandler;
 import lombok.extern.slf4j.Slf4j;
 
 import java.util.HashMap;
@@ -31,12 +36,25 @@ public class AuthVerticle extends AbstractVerticle {
     @Override
     public void start() throws Exception {
         backendServer = vertx.createHttpServer();
+
+        //第一步 声明Router&初始化Router
         Router backendRouter = Router.router(vertx);
+        //获取body参数，得先添加这句
+        backendRouter.route().handler(BodyHandler.create());
 
+        //第二步 配置Router解析url
         backendRouter.route(HttpMethod.POST, "/mqtt/auth").handler(rc -> {
             JsonObject json = rc.getBodyAsJson();
+
+            String clientid = json.getString("clientid", "");
+            String username = json.getString("username", "");
+            String password = json.getString("password", "");
+
+            log.info(String.format("clientid: %s, username: %s, password: %s", clientid, username, password));
+
             try {
-                executor.onReceive(new HashMap<>(), "auth", json.toString());
+
+                //executor.onReceive(new HashMap<>(), "auth", json.toString());
                 rc.response().setStatusCode(200)
                         .end();
             } catch (Throwable e) {
@@ -48,9 +66,29 @@ public class AuthVerticle extends AbstractVerticle {
         backendRouter.route(HttpMethod.POST, "/mqtt/acl").handler(rc -> {
             JsonObject json = rc.getBodyAsJson();
             try {
+                String clientid = json.getString("clientid", "");
+                String topic = json.getString("topic", "");
+                String access = json.getString("access", "").equals("1") ? "subscribe" : "publish"; //1 - subscribe, 2 - publish
+
+                log.info(String.format("clientid: %s, username: %s, password: %s", clientid, topic, access));
+
+
                 Map<String, Object> head = new HashMap<>();
-                head.put("topic", json.getString("topic"));
-                executor.onReceive(head, "subscribe", json.toString());
+                head.put("topic", topic);
+
+                /**
+                 * 1、匹配clientId, 匹配topic (topic白名单)
+                 */
+                if (topic.matches(Constants.MQTT.DEVICE_SUBSCRIBE_TOPIC)) {
+                    DeviceRepository deviceRepository = SpringUtils.getBean(DeviceRepository.class);
+
+                    String dd = JsonUtil.toJsonString(deviceRepository.findAll().get(0));
+                    log.info(dd);
+
+                    executor.onReceive(head, access, json.toString());
+                }
+
+
                 rc.response().setStatusCode(200)
                         .end();
             } catch (Throwable e) {

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

@@ -1,32 +1,50 @@
 package cc.iotkit.comp.emqx;
 
+import cc.iotkit.common.Constants;
 import cc.iotkit.common.exception.BizException;
 import cc.iotkit.common.utils.JsonUtil;
 import cc.iotkit.comp.AbstractDeviceComponent;
 import cc.iotkit.comp.CompConfig;
+import cc.iotkit.comp.IMessageHandler;
 import cc.iotkit.comp.model.DeviceState;
 import cc.iotkit.converter.DeviceMessage;
+import cc.iotkit.converter.ThingService;
+import cc.iotkit.model.device.message.ThingModelMessage;
+import com.fasterxml.jackson.databind.JsonNode;
+import io.netty.handler.codec.mqtt.MqttQoS;
 import io.vertx.core.Future;
 import io.vertx.core.Vertx;
+import io.vertx.core.buffer.Buffer;
 import io.vertx.mqtt.MqttClient;
 import io.vertx.mqtt.MqttClientOptions;
-import io.vertx.mqtt.messages.MqttConnAckMessage;
-import lombok.SneakyThrows;
-import lombok.extern.slf4j.Slf4j;
+import lombok.*;
+import org.apache.commons.beanutils.BeanUtils;
+import org.apache.commons.lang3.RandomStringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
+import java.lang.reflect.InvocationTargetException;
+import java.nio.charset.Charset;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 
-@Slf4j
+
 public class EmqxDeviceComponent extends AbstractDeviceComponent {
 
+    private static final Logger log = LoggerFactory.getLogger(EmqxDeviceComponent.class);
     private Vertx vertx;
     private AuthVerticle authVerticle;
+    //private MqttVerticle mqttVerticle;
     private CountDownLatch countDownLatch;
     private String deployedId;
     private EmqxConfig mqttConfig;
+    MqttClient client;
+
+    private final Map<String, Device> deviceChildToParent = new HashMap<>();
+
+    private TransparentConverter transparentConverter = new TransparentConverter();
 
     public void create(CompConfig config) {
         super.create(config);
@@ -44,11 +62,13 @@ public class EmqxDeviceComponent extends AbstractDeviceComponent {
             future.onSuccess((s -> {
                 deployedId = s;
                 countDownLatch.countDown();
+                log.error("start emqx auth component success", s);
             }));
             future.onFailure((e) -> {
                 countDownLatch.countDown();
                 log.error("start emqx auth component failed", e);
             });
+
             countDownLatch.await();
 
             MqttClientOptions options = new MqttClientOptions()
@@ -62,54 +82,134 @@ public class EmqxDeviceComponent extends AbstractDeviceComponent {
                 options.setSsl(true)
                         .setTrustAll(true);
             }
-            MqttClient client = MqttClient.create(vertx, options);
+            client = MqttClient.create(vertx, options);
+
+
+            // handler will be called when we have a message in topic we subscribe for
+            /*client.publishHandler(p -> {
+                log.info("Client received message on [{}] payload [{}] with QoS [{}]", p.topicName(), p.payload().toString(Charset.defaultCharset()), p.qosLevel());
+            });*/
 
-            Future<MqttConnAckMessage> connFuture =
-                    client.connect(mqttConfig.getPort(), mqttConfig.getBroker());
-            connFuture.onSuccess(ack -> log.info("connect emqx broker success"))
-                    .onFailure(e -> log.error("connect emqx broker failed", e));
 
             List<String> topics = mqttConfig.getSubscribeTopics();
             Map<String, Integer> subscribes = new HashMap<>();
-            for (String topic : topics) {
+
+            subscribes.put("/sys/+/+/s/#", 1);
+            subscribes.put("/sys/client/connected", 1);
+            subscribes.put("/sys/client/disconnected", 1);
+            subscribes.put("/sys/session/subscribed", 1);
+            subscribes.put("/sys/session/unsubscribed", 1);
+
+            //"/sys/+/+/s/#","/sys/client/disconnected"
+
+            /*for (String topic : topics) {
                 subscribes.put(topic, 1);
-            }
+            }*/
+
+            // handler will be called when we have a message in topic we subscribe for
+            client.publishHandler(p -> {
+                log.info("Client received message on [{}] payload [{}] with QoS [{}]", p.topicName(), p.payload().toString(Charset.defaultCharset()), p.qosLevel());
 
-            client.publishHandler(s -> {
-                        String topic = s.topicName();
-                        String payload = s.payload().toString();
-                        log.info("receive message,topic:{},payload:{}", topic, payload);
-
-//
-//                //取消订阅
-//                if (topic.equals("/sys/session/topic/unsubscribed")) {
-//                    topicUnsubscribed(payload);
-//                    return;
-//                }
-//
-//                //连接断开
-//                if (topic.equals("/sys/client/disconnected")) {
-//                    disconnectedHandler.handler(payload);
-//                    return;
-//                }
-//
-//                String[] parts = topic.split("/");
-//                if (parts.length < 5) {
-//                    log.error("message topic is illegal.");
-//                    return;
-//                }
-//                String productKey = parts[2];
-//                String deviceName = parts[3];
-//
-//                //子设备注册
-//                if (topic.endsWith("/register")) {
+                String topic = p.topicName();
+                String payload = p.payload().toString();
 
+                try {
+                    IMessageHandler messageHandler = getHandler();
 
+                    if (messageHandler != null) {
                         Map<String, Object> head = new HashMap<>();
                         head.put("topic", topic);
-                        getHandler().onReceive(head, "", payload);
-                    }).subscribe(subscribes).onSuccess(a -> log.info("subscribe topic success"))
-                    .onFailure(e -> log.error("subscribe topic failed", e));
+                        if (topic.equals("/sys/client/connected")) {
+                            JsonNode payloadJson = JsonUtil.parse(payload);
+                            String clientId = payloadJson.get("clientid").textValue();
+                            log.warn("client connection connected,clientId:{}", clientId);
+                            head.put("clientId", clientId);
+                            messageHandler.onReceive(head, "connect", payload);
+                            return;
+                        }
+
+                        //连接断开
+                        if (topic.equals("/sys/client/disconnected")) {
+                            JsonNode payloadJson = JsonUtil.parse(payload);
+                            String clientId = payloadJson.get("clientid").textValue();
+                            log.warn("client connection closed,clientId:{}", clientId);
+                            head.put("clientId", clientId);
+                            messageHandler.onReceive(head, "disconnect", payload);
+                            return;
+                        }
+
+                        /**
+                        ** 子设备在线离线状态（topic: ^/sys/.+/.+/c/#$）**： 改为从 从 acl 访问控制 获取离线在线状态。
+
+
+                        if (topic.equals("/sys/session/subscribed")) {
+                            JsonNode payloadJson = JsonUtil.parse(payload);
+                            String _topic = payloadJson.get("topic").textValue();
+
+                            //在线
+                            if (_topic.matches(Constants.MQTT.DEVICE_SUBSCRIBE_TOPIC)) {
+                                //head.put("topic", _topic);
+                                String clientId = payloadJson.get("clientid").textValue();
+                                log.warn("session subscribe, topic:{}", _topic);
+                                head.put("clientId", clientId);
+                                messageHandler.onReceive(head, "subscribe", payload);
+                            }
+
+                            return;
+                        }
+
+
+                        if (topic.equals("/sys/session/unsubscribed")) {
+                            JsonNode payloadJson = JsonUtil.parse(payload);
+                            String _topic = payloadJson.get("topic").textValue();
+
+                            //离线
+                            if (_topic.matches(Constants.MQTT.DEVICE_SUBSCRIBE_TOPIC)) {
+                                //head.put("topic", _topic);
+                                String clientId = payloadJson.get("clientid").textValue();
+                                log.warn("session unsubscribe, topic:{}", _topic);
+                                head.put("clientId", clientId);
+                                messageHandler.onReceive(head, "unsubscribe", payload);
+                            }
+
+                            return;
+                        }*/
+
+                        String[] parts = topic.split("/");
+                        if (parts.length < 5) {
+                            log.error("message topic is illegal.");
+                            return;
+                        }
+
+                        messageHandler.onReceive(head, "", payload);
+
+                    }
+                } catch (Exception e) {
+                    log.error("message topic is illegal.", e);
+                }
+            });
+
+            client.connect(mqttConfig.getPort(), mqttConfig.getBroker(), s -> {
+                if (s.succeeded()) {
+                    log.info("client connect success.");
+                    client.subscribe(subscribes, e -> {
+                        if (e.succeeded()) {
+                            log.info("===>subscribe success: {}", e.result());
+                        } else {
+                            log.error("===>subscribe fail: ", e.cause());
+                        }
+                    });
+
+                } else {
+                    log.error("client connect fail: ", s.cause());
+                }
+            }).exceptionHandler(event -> {
+                log.error("client fail: ", event.getCause());
+            });
+
+            /** client.pingResponseHandler(s -> {
+             log.info("We have just received PINGRESP packet");
+             });*/
 
         } catch (Throwable e) {
             throw new BizException("start emqx auth component error", e);
@@ -122,6 +222,9 @@ public class EmqxDeviceComponent extends AbstractDeviceComponent {
         authVerticle.stop();
         Future<Void> future = vertx.undeploy(deployedId);
         future.onSuccess(unused -> log.info("stop emqx auth component success"));
+        client.disconnect()
+                .onSuccess(unused -> log.info("stop emqx component success"))
+                .onFailure(unused -> log.info("stop emqx component failure"));
     }
 
     @Override
@@ -131,17 +234,95 @@ public class EmqxDeviceComponent extends AbstractDeviceComponent {
 
     @Override
     public void onDeviceStateChange(DeviceState state) {
+        DeviceState.Parent parent = state.getParent();
+        if (parent == null) {
+            return;
+        }
+        Device device = new Device(state.getProductKey(), state.getDeviceName());
 
+        if (DeviceState.STATE_ONLINE.equals(state.getState())) {
+            //保存子设备所属父设备
+            deviceChildToParent.put(device.toString(),
+                    new Device(parent.getProductKey(), parent.getDeviceName())
+            );
+        } else {
+            //删除关系
+            deviceChildToParent.remove(device.toString());
+        }
     }
 
     @Override
     public void send(DeviceMessage message) {
+        Device child = new Device(message.getProductKey(), message.getDeviceName());
+        //作为子设备查找父设备
+        Device parent = deviceChildToParent.get(child.toString());
+        if (parent == null) {
+            parent = child;
+        }
+
+        Object obj = message.getContent();
+        if (!(obj instanceof Map)) {
+            throw new BizException("message content is not Map");
+        }
+        Message msg = new Message();
+        try {
+            //obj中的key,如果bean中有这个属性，就把这个key对应的value值赋给msg的属性
+            BeanUtils.populate(msg, (Map<String, ? extends Object>) obj);
+        } catch (Throwable e) {
+            throw new BizException("message content is incorrect");
+        }
+
+        log.info("publish topic:{},payload:{}", msg.getTopic(), msg.getPayload());
 
+        client.publish(msg.getTopic(),
+                Buffer.buffer(msg.getPayload()),
+                MqttQoS.AT_LEAST_ONCE,
+                false,
+                false);
     }
 
     @Override
     public boolean exist(String productKey, String deviceName) {
-        return false;
+        return true;
+
+        /*//先作为子设备查找是否存在父设备
+        Device device = deviceChildToParent.get(new Device(productKey, deviceName).toString());
+        if (device != null) {
+            return true;
+        }*/
+
+        //return mqttVerticle.exist(productKey, deviceName);
+    }
+
+
+    /**
+     * 透传解码
+     */
+    public ThingModelMessage transparentDecode(Map<String, Object> msg) throws InvocationTargetException, IllegalAccessException {
+        TransparentMsg transparentMsg = new TransparentMsg();
+        BeanUtils.populate(transparentMsg, msg);
+        return transparentConverter.decode(transparentMsg);
+    }
+
+    /**
+     * 透传编码
+     */
+    public DeviceMessage transparentEncode(ThingService<?> service, cc.iotkit.converter.Device device) {
+        return transparentConverter.encode(service, device);
     }
 
+    @Data
+    public static class Message {
+        private String topic;
+        private String payload;
+    }
+
+    @Data
+    @NoArgsConstructor
+    @AllArgsConstructor
+    @ToString
+    public static class Device {
+        private String productKey;
+        private String deviceName;
+    }
 }

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

@@ -0,0 +1,19 @@
+package cc.iotkit.comp.emqx;
+
+import cc.iotkit.converter.ThingService;
+import cc.iotkit.model.device.message.ThingModelMessage;
+
+public interface IScripter {
+
+    void setScript(String script);
+
+    /**
+     * 透传解码
+     */
+    ThingModelMessage decode(TransparentMsg msg);
+
+    /**
+     * 透传编码
+     */
+    TransparentMsg encode(ThingService<?> service);
+}

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

@@ -0,0 +1,19 @@
+package cc.iotkit.comp.emqx;
+
+import cc.iotkit.converter.ThingService;
+import cc.iotkit.model.device.message.ThingModelMessage;
+
+public class JsScripter implements IScripter {
+
+    @Override
+    public void setScript(String script) {
+    }
+
+    public ThingModelMessage decode(TransparentMsg msg) {
+        return null;
+    }
+
+    public TransparentMsg encode(ThingService<?> service) {
+        return null;
+    }
+}

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

@@ -0,0 +1,126 @@
+package cc.iotkit.comp.emqx;
+
+import cc.iotkit.converter.ThingService;
+import cc.iotkit.model.device.message.ThingModelMessage;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.beanutils.BeanUtils;
+import org.luaj.vm2.LuaTable;
+import org.luaj.vm2.LuaValue;
+import org.luaj.vm2.script.LuaScriptEngine;
+
+import javax.script.Compilable;
+import javax.script.CompiledScript;
+import javax.script.ScriptEngineManager;
+import javax.script.SimpleBindings;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@Slf4j
+public class LuaScripter implements IScripter {
+
+    private final LuaScriptEngine engine = (LuaScriptEngine) (
+            new ScriptEngineManager().getEngineByName("luaj"));
+
+    private LuaValue decoder;
+    private LuaValue encoder;
+
+    @Override
+    public void setScript(String script) {
+        try {
+            CompiledScript compiledScript = ((Compilable) engine).compile(script);
+            SimpleBindings bindings = new SimpleBindings();
+            compiledScript.eval(bindings);
+            decoder = (LuaValue) bindings.get("decode");
+            encoder = (LuaValue) bindings.get("encode");
+        } catch (Throwable e) {
+            log.error("compile script error", e);
+        }
+    }
+
+    public ThingModelMessage decode(TransparentMsg msg) {
+        try {
+            LuaTable table = new LuaTable();
+            table.set("model", msg.getModel());
+            table.set("mac", msg.getMac());
+            table.set("data", msg.getData());
+            Map result = (Map) parse(decoder.call(table));
+            ThingModelMessage modelMessage = new ThingModelMessage();
+            BeanUtils.populate(modelMessage, result);
+
+            modelMessage.setProductKey(msg.getProductKey());
+            modelMessage.setDeviceName(msg.getMac());
+            return modelMessage;
+        } catch (Throwable e) {
+            log.error("execute decode script error", e);
+        }
+        return null;
+    }
+
+    public TransparentMsg encode(ThingService<?> service) {
+        try {
+            LuaTable table = new LuaTable();
+            table.set("identifier", service.getIdentifier());
+            table.set("type", service.getType());
+            table.set("productKey", service.getProductKey());
+            table.set("deviceName", service.getDeviceName());
+            table.set("mid", service.getMid());
+            Object params = service.getParams();
+            LuaTable tableParams = new LuaTable();
+            if (params instanceof Map) {
+                ((Map<?, ?>) params).forEach((key, val) -> tableParams.set(key.toString(), parse(val)));
+            }
+            table.set("params", tableParams);
+            LuaValue result = encoder.call(table);
+            Map map = (Map) parse(result);
+            TransparentMsg message = new TransparentMsg();
+            BeanUtils.populate(message, map);
+            return message;
+        } catch (Throwable e) {
+            log.error("execute encode script error", e);
+        }
+        return null;
+    }
+
+    private Object parse(LuaValue value) {
+        String type = value.typename();
+        switch (type) {
+            case "string":
+                return value.toString();
+            case "number":
+            case "int":
+                return value.toint();
+            case "table":
+                Map<String, Object> data = new HashMap<>();
+                LuaTable table = (LuaTable) value;
+                int arrLen = table.rawlen();
+                if (arrLen > 0) {
+                    //数组转换
+                    List<Object> list = new ArrayList<>();
+                    for (LuaValue key : table.keys()) {
+                        list.add(parse(table.get(key)));
+                    }
+                    return list;
+                } else {
+                    //map转换
+                    for (LuaValue key : table.keys()) {
+                        data.put(key.toString(), parse(table.get(key)));
+                    }
+                }
+                return data;
+        }
+        return null;
+    }
+
+    private LuaValue parse(Object value) {
+        if (value instanceof String) {
+            return LuaValue.valueOf(value.toString());
+        }
+        if (value instanceof Integer) {
+            return LuaValue.valueOf((Integer) value);
+        }
+        return new LuaTable();
+    }
+
+}

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

@@ -0,0 +1,88 @@
+package cc.iotkit.comp.emqx;
+
+
+import cc.iotkit.converter.Device;
+import cc.iotkit.converter.DeviceMessage;
+import cc.iotkit.converter.ThingService;
+import cc.iotkit.dao.DeviceCache;
+import cc.iotkit.dao.ProductCache;
+import cc.iotkit.model.device.DeviceInfo;
+import cc.iotkit.model.device.message.ThingModelMessage;
+import cc.iotkit.model.product.ProductModel;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Slf4j
+public class TransparentConverter {
+
+    private final Map<String, IScripter> scripters = new HashMap<>();
+    private final Map<String, String> scripts = new HashMap<>();
+
+    /**
+     * 透传解码
+     */
+    public ThingModelMessage decode(TransparentMsg msg) {
+        //通过上报消息中的model取得对应的产品
+        String productKey = checkScriptUpdate(msg.getModel());
+        msg.setProductKey(productKey);
+        return scripters.get(productKey).decode(msg);
+    }
+
+    /**
+     * 透传编码
+     */
+    public DeviceMessage encode(ThingService<?> service, Device device) {
+        String productKey = service.getProductKey();
+        checkScriptUpdate(device.getModel());
+        TransparentMsg transparentMsg = scripters.get(productKey).encode(service);
+        //转换成网关消息
+        String deviceName = service.getDeviceName();
+        DeviceInfo gateway = getGatewayInfo(productKey, deviceName);
+        DeviceMessage message = new DeviceMessage();
+        message.setProductKey(gateway.getProductKey());
+        message.setDeviceName(gateway.getDeviceName());
+        message.setMid(transparentMsg.getMid());
+        //透传格式消息内容,mac、model、data
+        message.setContent(transparentMsg);
+        return message;
+    }
+
+    private ProductModel getScript(String model) {
+        return ProductCache.getInstance().getProductScriptByModel(model);
+    }
+
+    private DeviceInfo getGatewayInfo(String subPk, String subDn) {
+        String parentId = DeviceCache.getInstance().getDeviceInfo(subPk, subDn).getParentId();
+        return DeviceCache.getInstance().get(parentId);
+    }
+
+    /**
+     * 检查产品脚本是否更新
+     */
+    private String checkScriptUpdate(String model) {
+        ProductModel productModel = getScript(model);
+        String productKey = productModel.getProductKey();
+        String script = productModel.getScript();
+
+        String oldScript = scripts.get(productKey);
+        if (script.equals(oldScript)) {
+            return productKey;
+        }
+
+        String type = productModel.getType();
+        if (ProductModel.TYPE_LUA.equals(type)) {
+            scripters.putIfAbsent(productKey, new LuaScripter());
+        } else if (ProductModel.TYPE_JS.equals(type)) {
+            scripters.putIfAbsent(productKey, new JsScripter());
+        }
+
+        //更新脚本
+        IScripter scripter = scripters.get(productKey);
+        scripter.setScript(script);
+        scripts.put(productKey, script);
+        return productKey;
+    }
+
+}

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

@@ -0,0 +1,21 @@
+package cc.iotkit.comp.emqx;
+
+import lombok.Data;
+
+@Data
+public class TransparentMsg {
+
+    private String productKey;
+
+    /**
+     * 生成给设备端的消息id
+     */
+    private String mid;
+
+    private String model;
+
+    private String mac;
+
+    private String data;
+
+}