协议网关修改

common/src/main/java/cc/iotkit/common/utils/JsonUtil.java

@@ -5,8 +5,14 @@ import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
+import jdk.nashorn.api.scripting.ScriptObjectMirror;
 import lombok.SneakyThrows;
 
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
 public final class JsonUtil {
 
     private final static ObjectMapper MAPPER = new ObjectMapper()
@@ -32,4 +38,33 @@ public final class JsonUtil {
     public static JsonNode parse(String json) {
         return MAPPER.readTree(json);
     }
+
+    public static Object toObject(ScriptObjectMirror mirror) {
+        if (mirror.isEmpty()) {
+            return null;
+        }
+        if (mirror.isArray()) {
+            List<Object> list = new ArrayList<>();
+            for (Map.Entry<String, Object> entry : mirror.entrySet()) {
+                Object result = entry.getValue();
+                if (result instanceof ScriptObjectMirror) {
+                    list.add(toObject((ScriptObjectMirror) result));
+                } else {
+                    list.add(result);
+                }
+            }
+            return list;
+        }
+
+        Map<String, Object> map = new HashMap<>();
+        for (Map.Entry<String, Object> entry : mirror.entrySet()) {
+            Object result = entry.getValue();
+            if (result instanceof ScriptObjectMirror) {
+                map.put(entry.getKey(), toObject((ScriptObjectMirror) result));
+            } else {
+                map.put(entry.getKey(), result);
+            }
+        }
+        return map;
+    }
 }

common/src/main/java/cc/iotkit/common/utils/ThreadUtil.java

@@ -0,0 +1,25 @@
+package cc.iotkit.common.utils;
+
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+
+public class ThreadUtil {
+
+    public static ScheduledThreadPoolExecutor newScheduled(int poolSize, String threadName) {
+        return new ScheduledThreadPoolExecutor(poolSize, (Runnable r) -> {
+            SecurityManager s = System.getSecurityManager();
+            ThreadGroup group = (s != null) ? s.getThreadGroup() :
+                    Thread.currentThread().getThreadGroup();
+            Thread t = new Thread(group, r,
+                    threadName,
+                    0);
+            if (t.isDaemon()) {
+                t.setDaemon(false);
+            }
+            if (t.getPriority() != Thread.NORM_PRIORITY) {
+                t.setPriority(Thread.NORM_PRIORITY);
+            }
+            return t;
+        });
+    }
+
+}

dao/src/main/java/cc/iotkit/dao/AligenieProductDao.java

@@ -1,28 +0,0 @@
-package cc.iotkit.dao;
-
-import cc.iotkit.model.aligenie.AligenieProduct;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.cache.annotation.Cacheable;
-import org.springframework.data.domain.Example;
-import org.springframework.data.mongodb.core.MongoTemplate;
-import org.springframework.stereotype.Repository;
-
-@Repository
-public class AligenieProductDao extends BaseDao<AligenieProduct> {
-
-    private final AligenieProductRepository aligenieProductRepository;
-
-    @Autowired
-    public AligenieProductDao(MongoTemplate mongoTemplate,
-                              AligenieProductRepository aligenieProductRepository) {
-        super(mongoTemplate, AligenieProduct.class);
-        this.aligenieProductRepository = aligenieProductRepository;
-    }
-
-    @Cacheable(value = "cache_getAligenieProduct", key = "'getAligenieProduct'+#pk", unless = "#result == null")
-    public AligenieProduct getAligenieProduct(String pk) {
-        return aligenieProductRepository.findOne(Example.of(
-                AligenieProduct.builder().productKey(pk).build()
-        )).orElse(null);
-    }
-}

dao/src/main/java/cc/iotkit/dao/BaseDao.java

@@ -1,59 +0,0 @@
-package cc.iotkit.dao;
-
-import org.springframework.data.domain.Sort;
-import org.springframework.data.mongodb.core.MongoTemplate;
-import org.springframework.data.mongodb.core.query.Criteria;
-import org.springframework.data.mongodb.core.query.Query;
-
-import java.util.List;
-
-import static org.springframework.data.mongodb.core.query.Criteria.where;
-import static org.springframework.data.mongodb.core.query.Query.query;
-
-public class BaseDao<T> {
-
-    protected MongoTemplate mongoTemplate;
-
-    private Class<T> cls;
-
-    public BaseDao(MongoTemplate mongoTemplate, Class<T> cls) {
-        this.mongoTemplate = mongoTemplate;
-        this.cls = cls;
-    }
-
-    public List<T> find(Criteria condition) {
-        Query query = new Query();
-        query.addCriteria(condition);
-        return mongoTemplate.find(query, cls);
-    }
-
-    public List<T> find(Criteria condition, long skip, int count, Sort.Order order) {
-        Query query = new Query();
-        query.addCriteria(condition)
-                .with(Sort.by(order))
-                .skip(skip)
-                .limit(count);
-        return mongoTemplate.find(query, cls);
-    }
-
-    public long count(Criteria condition) {
-        Query query = new Query();
-        query.addCriteria(condition);
-        return mongoTemplate.count(query, cls);
-    }
-
-    public <T> T save(String id, T entity) {
-        if (id == null) {
-            return mongoTemplate.save(entity);
-        } else {
-            mongoTemplate.updateFirst(query(where("_id").is(id)),
-                    DaoTool.update(entity), entity.getClass());
-            return (T) mongoTemplate.findById(id, entity.getClass());
-        }
-    }
-
-    public <T> T save(T entity) {
-        return mongoTemplate.save(entity);
-    }
-
-}

dao/src/main/java/cc/iotkit/dao/DeviceCache.java

@@ -4,24 +4,14 @@ import cc.iotkit.common.Constants;
 import cc.iotkit.model.device.DeviceInfo;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.cache.annotation.Cacheable;
-import org.springframework.data.mongodb.core.MongoTemplate;
-import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.stereotype.Repository;
 
-import static org.springframework.data.mongodb.core.query.Criteria.where;
-import static org.springframework.data.mongodb.core.query.Query.query;
-
 @Repository
-public class DeviceCache extends BaseDao<DeviceInfo> {
+public class DeviceCache {
 
     @Autowired
     private DeviceRepository deviceRepository;
 
-    @Autowired
-    public DeviceCache(MongoTemplate mongoTemplate) {
-        super(mongoTemplate, DeviceInfo.class);
-    }
-
     @Cacheable(value = Constants.DEVICE_CACHE, key = "#pk+'_'+#dn")
     public DeviceInfo findByProductKeyAndDeviceName(String pk, String dn) {
         return deviceRepository.findByProductKeyAndDeviceName(pk, dn);

dao/src/main/java/cc/iotkit/dao/DeviceDao.java

@@ -0,0 +1,44 @@
+package cc.iotkit.dao;
+
+import cc.iotkit.model.Paging;
+import cc.iotkit.model.device.DeviceInfo;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.query.Criteria;
+import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.data.mongodb.core.query.Update;
+import org.springframework.stereotype.Repository;
+
+import java.util.Map;
+
+@Repository
+public class DeviceDao {
+
+    @Autowired
+    private MongoTemplate mongoTemplate;
+
+    public Paging<DeviceInfo> find(Criteria condition, int size, int page) {
+        Query query = Query.query(condition);
+        return new Paging<>(
+                mongoTemplate.count(query, DeviceInfo.class),
+                mongoTemplate.find(
+                        query.with(PageRequest.of(page-1, size, Sort.by(Sort.Order.desc("createAt"))))
+                        , DeviceInfo.class)
+        );
+    }
+
+    /**
+     * 更新设备属性
+     */
+    public void updateProperties(String deviceId, Map<String, Object> properties) {
+        Query query = Query.query(new Criteria().and("deviceId").is(deviceId));
+        Update update = new Update();
+        for (String key : properties.keySet()) {
+            update.set("property." + key, properties.get(key));
+        }
+        mongoTemplate.updateFirst(query, update, DeviceInfo.class);
+    }
+
+}

dao/src/main/java/cc/iotkit/dao/DeviceEventDao.java

@@ -1,15 +0,0 @@
-package cc.iotkit.dao;
-
-import cc.iotkit.model.device.message.DeviceEvent;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.data.mongodb.core.MongoTemplate;
-import org.springframework.stereotype.Repository;
-
-@Repository
-public class DeviceEventDao extends BaseDao<DeviceEvent> {
-
-    @Autowired
-    public DeviceEventDao(MongoTemplate mongoTemplate) {
-        super(mongoTemplate, DeviceEvent.class);
-    }
-}

dao/src/main/java/cc/iotkit/dao/DevicePropertyDao.java

@@ -0,0 +1,40 @@
+package cc.iotkit.dao;
+
+import cc.iotkit.model.device.message.DeviceProperty;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.search.sort.FieldSortBuilder;
+import org.elasticsearch.search.sort.SortOrder;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
+import org.springframework.data.elasticsearch.core.SearchHit;
+import org.springframework.data.elasticsearch.core.SearchHits;
+import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
+import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
+import org.springframework.stereotype.Repository;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+@Repository
+public class DevicePropertyDao {
+
+    @Autowired
+    private ElasticsearchRestTemplate template;
+
+    public List<DeviceProperty> findDevicePropertyHistory(String deviceId, String name, long start, long end) {
+        NativeSearchQuery query = new NativeSearchQueryBuilder()
+                .withQuery(
+                        QueryBuilders.boolQuery()
+                                .must(QueryBuilders.termQuery("deviceId", deviceId))
+                                .must(QueryBuilders.termQuery("name", name))
+                                .must(QueryBuilders.rangeQuery("time")
+                                        .from(start, true).to(end, true))
+                )
+                .withSorts(new FieldSortBuilder("time").order(SortOrder.ASC))
+                .build();
+        SearchHits<DeviceProperty> result = template.search(query, DeviceProperty.class);
+        return result.getSearchHits().stream()
+                .map(SearchHit::getContent).collect(Collectors.toList());
+    }
+
+}

dao/src/main/java/cc/iotkit/dao/DevicePropertyRepository.java

@@ -0,0 +1,11 @@
+package cc.iotkit.dao;
+
+import cc.iotkit.model.device.message.DeviceProperty;
+import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;
+import org.springframework.stereotype.Repository;
+
+@Repository
+public interface DevicePropertyRepository extends ElasticsearchRepository<DeviceProperty, String> {
+
+
+}

dao/src/main/java/cc/iotkit/dao/DeviceRepository.java

@@ -4,6 +4,8 @@ import cc.iotkit.model.device.DeviceInfo;
 import org.springframework.data.mongodb.repository.MongoRepository;
 import org.springframework.stereotype.Repository;
 
+import java.util.List;
+
 @Repository
 public interface DeviceRepository extends MongoRepository<DeviceInfo, String> {
 
@@ -11,4 +13,6 @@ public interface DeviceRepository extends MongoRepository<DeviceInfo, String> {
 
     DeviceInfo findByDeviceId(String deviceId);
 
+    List<DeviceInfo> findByParentId(String parentId);
+
 }

dao/src/main/java/cc/iotkit/dao/ThingModelMessageDao.java

@@ -0,0 +1,45 @@
+package cc.iotkit.dao;
+
+import cc.iotkit.model.Paging;
+import cc.iotkit.model.device.message.ThingModelMessage;
+import org.apache.commons.lang3.StringUtils;
+import org.elasticsearch.index.query.BoolQueryBuilder;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
+import org.springframework.data.elasticsearch.core.SearchHit;
+import org.springframework.data.elasticsearch.core.SearchHits;
+import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
+import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
+import org.springframework.stereotype.Repository;
+
+import java.util.stream.Collectors;
+
+@Repository
+public class ThingModelMessageDao {
+
+    @Autowired
+    private ElasticsearchRestTemplate template;
+
+    public Paging<ThingModelMessage> findByTypeAndIdentifier(String deviceId, String type,
+                                                             String identifier,
+                                                             int page, int size) {
+        BoolQueryBuilder builder = QueryBuilders.boolQuery();
+        builder.must(QueryBuilders.termQuery("deviceId", deviceId));
+        if (StringUtils.isNotBlank(type)) {
+            builder.must(QueryBuilders.termQuery("type", type));
+        }
+        if (StringUtils.isNotBlank(identifier)) {
+            builder.must(QueryBuilders.matchPhraseQuery("identifier", identifier));
+        }
+        NativeSearchQuery query = new NativeSearchQueryBuilder().withQuery(builder)
+                .withPageable(PageRequest.of(page, size, Sort.by(Sort.Order.desc("time"))))
+                .build();
+        SearchHits<ThingModelMessage> result = template.search(query, ThingModelMessage.class);
+        return new Paging<>(result.getTotalHits(), result.getSearchHits().stream()
+                .map(SearchHit::getContent).collect(Collectors.toList()));
+    }
+
+}

dao/src/main/java/cc/iotkit/dao/ThingModelMessageRepository.java

@@ -1,9 +1,14 @@
 package cc.iotkit.dao;
 
 import cc.iotkit.model.device.message.ThingModelMessage;
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.Pageable;
 import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;
 import org.springframework.stereotype.Repository;
 
 @Repository
 public interface ThingModelMessageRepository extends ElasticsearchRepository<ThingModelMessage, String> {
+
+    Page<ThingModelMessage> findByTypeAndIdentifier(String type, String identifier, Pageable pageable);
+
 }

dao/src/main/java/cc/iotkit/dao/UserInfoDao.java

@@ -1,15 +0,0 @@
-package cc.iotkit.dao;
-
-import cc.iotkit.model.UserInfo;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.data.mongodb.core.MongoTemplate;
-import org.springframework.stereotype.Repository;
-
-@Repository
-public class UserInfoDao extends BaseDao<UserInfo> {
-
-    @Autowired
-    public UserInfoDao(MongoTemplate mongoTemplate) {
-        super(mongoTemplate, UserInfo.class);
-    }
-}

device-server/device-api/pom.xml

@@ -1,35 +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>device-server</artifactId>
-        <groupId>cc.iotkit</groupId>
-        <version>0.0.1-SNAPSHOT</version>
-    </parent>
-    <modelVersion>4.0.0</modelVersion>
-
-    <artifactId>device-api</artifactId>
-
-
-    <dependencies>
-        <dependency>
-            <groupId>org.springframework.cloud</groupId>
-            <artifactId>spring-cloud-openfeign-core</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>org.springframework</groupId>
-            <artifactId>spring-web</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>org.projectlombok</groupId>
-            <artifactId>lombok</artifactId>
-            <scope>compile</scope>
-        </dependency>
-        <dependency>
-            <groupId>cc.iotkit</groupId>
-            <artifactId>model</artifactId>
-        </dependency>
-    </dependencies>
-
-</project>

device-server/device-api/src/main/java/cc/iotkit/deviceapi/IDeviceManager.java

@@ -1,44 +0,0 @@
-package cc.iotkit.deviceapi;
-
-import cc.iotkit.model.device.DeviceInfo;
-import org.springframework.cloud.openfeign.FeignClient;
-import org.springframework.stereotype.Component;
-import org.springframework.web.bind.annotation.*;
-
-import java.util.Map;
-
-@Component
-@FeignClient(value = "iot-device-manager",url = "localhost:8091")
-public interface IDeviceManager {
-
-    /**
-     * 设备注册
-     */
-    @PostMapping("/register")
-    @ResponseBody
-    DeviceInfo register(@RequestParam("parentId") String parentId,
-                        @RequestParam("productKey") String productKey,
-                        @RequestParam("deviceName") String deviceName,
-                        @RequestParam("model") String model);
-
-    /**
-     * 解绑子设备
-     */
-    @PostMapping("/{deviceId}/unbind")
-    void unbind(@PathVariable("deviceId") String deviceId);
-
-    /**
-     * 设置属性
-     */
-    @PostMapping("/{deviceId}/property/set")
-    String setProperty(@PathVariable("deviceId") String deviceId,
-                       @RequestBody Map<String, Object> properties);
-
-    /**
-     * 调用服务
-     */
-    @PostMapping("/{deviceId}/{identifier}/invoke")
-    String invokeService(@PathVariable("deviceId") String deviceId,
-                         @PathVariable("identifier") String identifier,
-                         @RequestBody Map<String, Object> properties);
-}

device-server/device-api/src/main/java/cc/iotkit/deviceapi/IDeviceService.java

@@ -1,16 +0,0 @@
-package cc.iotkit.deviceapi;
-
-import org.springframework.cloud.openfeign.FeignClient;
-import org.springframework.stereotype.Component;
-import org.springframework.web.bind.annotation.*;
-
-@Component
-@FeignClient(value = "iot-device-service", url = "localhost:8091")
-public interface IDeviceService {
-
-    /**
-     * 调用服务
-     */
-    @PostMapping("/invoke")
-    String invoke(@RequestBody Service service);
-}

device-server/device-api/src/main/java/cc/iotkit/deviceapi/Service.java

@@ -1,31 +0,0 @@
-package cc.iotkit.deviceapi;
-
-import lombok.Data;
-
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-@Data
-public class Service {
-
-    private String device;
-
-    private String identifier;
-
-    private List<Parameter> inputData;
-
-    public Map<String, Object> parseInputData() {
-        Map<String, Object> data = new HashMap<>();
-        for (Parameter p : inputData) {
-            data.put(p.getIdentifier(), p.getValue());
-        }
-        return data;
-    }
-
-    @Data
-    public static class Parameter {
-        private String identifier;
-        private Object value;
-    }
-}

device-server/mqtt-server/pom.xml

@@ -1,81 +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>device-server</artifactId>
-        <groupId>cc.iotkit</groupId>
-        <version>0.0.1-SNAPSHOT</version>
-    </parent>
-    <modelVersion>4.0.0</modelVersion>
-
-    <artifactId>mqtt-server</artifactId>
-
-    <dependencies>
-
-        <dependency>
-            <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-web</artifactId>
-        </dependency>
-
-        <dependency>
-            <groupId>org.projectlombok</groupId>
-            <artifactId>lombok</artifactId>
-            <optional>true</optional>
-        </dependency>
-
-        <dependency>
-            <groupId>commons-codec</groupId>
-            <artifactId>commons-codec</artifactId>
-        </dependency>
-
-        <dependency>
-            <groupId>org.bouncycastle</groupId>
-            <artifactId>bcprov-jdk15on</artifactId>
-        </dependency>
-
-        <dependency>
-            <groupId>org.eclipse.paho</groupId>
-            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
-        </dependency>
-
-        <dependency>
-            <groupId>cc.iotkit</groupId>
-            <artifactId>model</artifactId>
-        </dependency>
-
-        <dependency>
-            <groupId>cc.iotkit</groupId>
-            <artifactId>common</artifactId>
-        </dependency>
-
-        <dependency>
-            <groupId>cc.iotkit</groupId>
-            <artifactId>device-api</artifactId>
-        </dependency>
-
-        <dependency>
-            <groupId>cc.iotkit</groupId>
-            <artifactId>gateway-client</artifactId>
-        </dependency>
-
-    </dependencies>
-
-    <build>
-        <plugins>
-            <plugin>
-                <groupId>org.springframework.boot</groupId>
-                <artifactId>spring-boot-maven-plugin</artifactId>
-                <configuration>
-                    <excludes>
-                        <exclude>
-                            <groupId>org.projectlombok</groupId>
-                            <artifactId>lombok</artifactId>
-                        </exclude>
-                    </excludes>
-                </configuration>
-            </plugin>
-        </plugins>
-    </build>
-
-</project>

device-server/mqtt-server/src/main/java/cc/iotkit/server/mqtt/Application.java

@@ -1,12 +0,0 @@
-package cc.iotkit.server.mqtt;
-
-import org.springframework.boot.SpringApplication;
-import org.springframework.boot.autoconfigure.SpringBootApplication;
-
-@SpringBootApplication
-public class Application {
-
-    public static void main(String[] args) {
-        SpringApplication.run(Application.class, args);
-    }
-}

device-server/mqtt-server/src/main/java/cc/iotkit/server/mqtt/config/AutobeanConfig.java

@@ -1,19 +0,0 @@
-package cc.iotkit.server.mqtt.config;
-
-import cc.iotkit.protocol.client.DeviceBehaviourClient;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-
-@Configuration
-public class AutobeanConfig {
-
-    @Value("${gateway.server:}")
-    private String gatewayServer;
-
-    @Bean
-    public DeviceBehaviourClient getDeviceBehaviourClient() {
-        return new DeviceBehaviourClient(gatewayServer);
-    }
-
-}

device-server/mqtt-server/src/main/java/cc/iotkit/server/mqtt/controller/DeviceController.java

@@ -1,47 +0,0 @@
-package cc.iotkit.server.mqtt.controller;
-
-import cc.iotkit.common.utils.JsonUtil;
-import cc.iotkit.protocol.DeviceGateway;
-import cc.iotkit.protocol.DeviceMessage;
-import cc.iotkit.protocol.OtaInfo;
-import cc.iotkit.protocol.Result;
-import cc.iotkit.server.mqtt.service.MqttManager;
-import lombok.Data;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.web.bind.annotation.RestController;
-
-@Slf4j
-@RestController
-public class DeviceController implements DeviceGateway {
-
-    @Autowired
-    private MqttManager mqttManager;
-
-    @Override
-    public Result sendMessage(DeviceMessage msg) {
-        try {
-            MqttMessage mqttMessage = JsonUtil.parse(msg.getContent(), MqttMessage.class);
-            mqttManager.sendMsg(mqttMessage.getTopic(), mqttMessage.getPayload());
-            return new Result(true, "");
-        } catch (Throwable e) {
-            log.error("send message error", e);
-            return new Result(false, e.getMessage());
-        }
-    }
-
-    @Override
-    public Result sendOta(OtaInfo ota) {
-        return null;
-    }
-
-    @Data
-    public static class MqttMessage {
-
-        private String topic;
-
-        private String payload;
-
-    }
-
-}

device-server/mqtt-server/src/main/java/cc/iotkit/server/mqtt/controller/MqttAuthController.java

@@ -1,63 +0,0 @@
-package cc.iotkit.server.mqtt.controller;
-
-
-import cc.iotkit.common.Constants;
-import cc.iotkit.common.utils.CodecUtil;
-import cc.iotkit.common.utils.JsonUtil;
-import cc.iotkit.server.mqtt.model.EmqAcl;
-import cc.iotkit.server.mqtt.model.EmqAuthInfo;
-import cc.iotkit.server.mqtt.service.DeviceAuthService;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.web.bind.annotation.PostMapping;
-import org.springframework.web.bind.annotation.RequestBody;
-import org.springframework.web.bind.annotation.RestController;
-
-import javax.servlet.http.HttpServletResponse;
-
-@Slf4j
-@RestController
-public class MqttAuthController {
-
-    @Autowired
-    private DeviceAuthService deviceAuthService;
-
-    @PostMapping("/mqtt/auth")
-    public void auth(@RequestBody EmqAuthInfo auth) {
-        log.info("mqtt auth:" + JsonUtil.toJsonString(auth));
-        String clientId = auth.getClientid();
-        if (isSupperUser(clientId)) {
-            return;
-        }
-        deviceAuthService.auth(auth);
-    }
-
-    @PostMapping("/mqtt/acl")
-    public void acl(@RequestBody EmqAcl acl) {
-        log.info("mqtt acl:{}", JsonUtil.toJsonString(acl));
-        if (isSupperUser(acl.getClientid())) {
-            return;
-        }
-        deviceAuthService.acl(acl);
-        log.info("topic:{}, acl success", acl.getTopic());
-    }
-
-    @PostMapping("/mqtt/superuser")
-    public void superuser(@RequestBody EmqAcl acl, HttpServletResponse response) {
-        response.setStatus(HttpServletResponse.SC_BAD_GATEWAY);
-    }
-
-    public boolean isSupperUser(String clientId) {
-        try {
-            if (!clientId.startsWith("su_")) {
-                return false;
-            }
-            clientId = clientId.replaceFirst("su_", "");
-            return CodecUtil.aesDecrypt(clientId, Constants.PRODUCT_SECRET).startsWith("admin_");
-        } catch (Throwable e) {
-            log.error("aesDecrypt error.", e);
-            return false;
-        }
-    }
-
-}

device-server/mqtt-server/src/main/java/cc/iotkit/server/mqtt/handler/DisconnectedHandler.java

@@ -1,41 +0,0 @@
-package cc.iotkit.server.mqtt.handler;
-
-import cc.iotkit.common.utils.JsonUtil;
-import cc.iotkit.protocol.client.DeviceBehaviourClient;
-import com.fasterxml.jackson.core.type.TypeReference;
-import lombok.Data;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Component;
-
-@Slf4j
-@Component
-public class DisconnectedHandler {
-
-    @Autowired
-    private DeviceBehaviourClient behaviourClient;
-
-    public void handler(String msg) {
-        Disconnected disconnected = JsonUtil.parse(msg, new TypeReference<Disconnected>() {
-        });
-        String clientId = disconnected.getClientid();
-        String[] parts = clientId.split("_");
-        if (parts.length < 2) {
-            return;
-        }
-        String pk = parts[0];
-        String dn = parts[1];
-        behaviourClient.offline(pk, dn);
-        log.info("client disconnected, offline,pk:{},dn:{}", pk, dn);
-    }
-
-    @Data
-    private static class Disconnected {
-        private String reason;
-        private String clientid;
-        private String username;
-        private String peername;
-        private String sockname;
-    }
-
-}

device-server/mqtt-server/src/main/java/cc/iotkit/server/mqtt/model/EmqAcl.java

@@ -1,19 +0,0 @@
-package cc.iotkit.server.mqtt.model;
-
-import lombok.Data;
-
-@Data
-public class EmqAcl {
-
-    private String access;
-
-    private String username;
-
-    private String clientid;
-
-    private String ipaddr;
-
-    private String protocol;
-
-    private String topic;
-}

device-server/mqtt-server/src/main/java/cc/iotkit/server/mqtt/model/EmqAuthInfo.java

@@ -1,18 +0,0 @@
-package cc.iotkit.server.mqtt.model;
-
-import lombok.Data;
-
-@Data
-public class EmqAuthInfo {
-
-    private String clientid;
-
-    private String password;
-
-    private String username;
-
-    private String ipaddress;
-
-    private String protocol;
-
-}

device-server/mqtt-server/src/main/java/cc/iotkit/server/mqtt/model/Request.java

@@ -1,17 +0,0 @@
-package cc.iotkit.server.mqtt.model;
-
-
-import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-
-@Data
-@NoArgsConstructor
-@AllArgsConstructor
-@JsonIgnoreProperties(ignoreUnknown = true)
-public class Request<T> {
-    protected String id;
-
-    protected T params;
-}

device-server/mqtt-server/src/main/java/cc/iotkit/server/mqtt/model/Response.java

@@ -1,25 +0,0 @@
-package cc.iotkit.server.mqtt.model;
-
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-
-@Data
-@NoArgsConstructor
-@AllArgsConstructor
-public class Response<T> {
-
-    private String id;
-
-    private int code;
-
-    private T data;
-
-    public static Empty empty() {
-        return new Empty();
-    }
-
-    @Data
-    public static class Empty {
-    }
-}

device-server/mqtt-server/src/main/java/cc/iotkit/server/mqtt/service/DeviceAuthService.java

@@ -1,71 +0,0 @@
-package cc.iotkit.server.mqtt.service;
-
-import cc.iotkit.common.Constants;
-import cc.iotkit.common.utils.JsonUtil;
-import cc.iotkit.protocol.RegisterInfo;
-import cc.iotkit.protocol.Result;
-import cc.iotkit.protocol.client.DeviceBehaviourClient;
-import cc.iotkit.server.mqtt.model.EmqAcl;
-import cc.iotkit.server.mqtt.model.EmqAuthInfo;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.codec.digest.DigestUtils;
-import org.apache.commons.lang3.StringUtils;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Component;
-
-@Slf4j
-@Component
-public class DeviceAuthService {
-
-    @Autowired
-    private DeviceBehaviourClient behaviourClient;
-
-    public void auth(EmqAuthInfo auth) {
-        String clientId = auth.getClientid();
-        String[] pkDnAndModel = getPkDnAndModel(clientId);
-
-        String hmac = DigestUtils.md5Hex(Constants.PRODUCT_SECRET + clientId);
-        if (!hmac.equalsIgnoreCase(auth.getPassword())) {
-            throw new RuntimeException("password is illegal.");
-        }
-
-        String pk = pkDnAndModel[0];
-        String dn = pkDnAndModel[1];
-        String model = pkDnAndModel[2];
-
-        Result result = behaviourClient.register(new RegisterInfo(pk, dn, model));
-        log.info("register result:{}", JsonUtil.toJsonString(result));
-    }
-
-    public void acl(EmqAcl acl) {
-        String[] pkDn = getPkDnFromTopic(acl.getTopic());
-        String pk = pkDn[2];
-        String dn = pkDn[3];
-        behaviourClient.online(pk, dn);
-    }
-
-    private String[] getPkDnAndModel(String clientId) {
-        if (StringUtils.isBlank(clientId)) {
-            throw new RuntimeException("clientId is blank.");
-        }
-        clientId += "_";
-
-        String[] pkDnAndModel = clientId.split("_", -1);
-        if (pkDnAndModel.length < 3) {
-            throw new RuntimeException("clientId is illegal.");
-        }
-        return pkDnAndModel;
-    }
-
-    private String[] getPkDnFromTopic(String topic) {
-        if (StringUtils.isBlank(topic)) {
-            throw new RuntimeException("topic is blank.");
-        }
-
-        String[] pkDn = topic.split("/", -1);
-        if (pkDn.length < 4) {
-            throw new RuntimeException("topic is illegal.");
-        }
-        return pkDn;
-    }
-}

device-server/mqtt-server/src/main/java/cc/iotkit/server/mqtt/service/MqttManager.java

@@ -1,256 +0,0 @@
-package cc.iotkit.server.mqtt.service;
-
-import cc.iotkit.common.Constants;
-import cc.iotkit.common.utils.CodecUtil;
-import cc.iotkit.common.utils.JsonUtil;
-import cc.iotkit.protocol.DeviceMessage;
-import cc.iotkit.protocol.RegisterInfo;
-import cc.iotkit.protocol.client.DeviceBehaviourClient;
-import cc.iotkit.server.mqtt.handler.DisconnectedHandler;
-import cc.iotkit.server.mqtt.model.Request;
-import cc.iotkit.server.mqtt.model.Response;
-import com.fasterxml.jackson.core.type.TypeReference;
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-import lombok.SneakyThrows;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.StringUtils;
-import org.eclipse.paho.client.mqttv3.*;
-import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.stereotype.Component;
-
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-
-@Slf4j
-@Component
-public class MqttManager implements MqttCallback, IMqttMessageListener {
-
-    @Value("${mqtt.url}")
-    private String url;
-
-    @Value(("${spring.profiles.active:}"))
-    private String env;
-
-    private MqttClient mqttClient;
-
-    @Autowired
-    private DisconnectedHandler disconnectedHandler;
-    @Autowired
-    private DeviceBehaviourClient behaviourClient;
-
-    public MqttManager() {
-        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
-        executorService.scheduleWithFixedDelay(this::createClient, 1, 3, TimeUnit.SECONDS);
-    }
-
-    private MqttConnectOptions getMqttConnectOptions() {
-        MqttConnectOptions options = new MqttConnectOptions();
-        // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，
-        // 这里设置为true表示每次连接到服务器都以新的身份连接
-        options.setCleanSession(true);
-        // 设置连接的用户名
-        options.setUserName("admin");
-        // 设置连接的密码
-        options.setPassword("password".toCharArray());
-        options.setServerURIs(StringUtils.split(url, ","));
-        // 设置超时时间 单位为秒
-        options.setConnectionTimeout(10);
-        // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送心跳判断客户端是否在线，但这个方法并没有重连的机制
-        options.setKeepAliveInterval(20);
-        return options;
-    }
-
-    @SneakyThrows
-    private void createClient() {
-        try {
-            if (mqttClient == null) {
-                MemoryPersistence persistence = new MemoryPersistence();
-                String clientId = "mqtt-server-consumer-" + env;
-                clientId = "su_" + CodecUtil.aesEncrypt("admin_" + clientId, Constants.PRODUCT_SECRET);
-                mqttClient = new MqttClient(url, clientId, persistence);
-                mqttClient.setCallback(this);
-            }
-            if (mqttClient.isConnected()) {
-                return;
-            }
-            connect();
-        } catch (Throwable e) {
-            log.error("create mqttClient error", e);
-        }
-    }
-
-    private void connect() {
-        try {
-            log.info("Connecting to broker:{} ", url);
-            IMqttToken mqttToken = mqttClient.connectWithResult(getMqttConnectOptions());
-            if (mqttToken.isComplete()) {
-                log.info("connect mqtt-broker success");
-            } else {
-                log.error("connect mqtt-broker failed", mqttToken.getException());
-            }
-            IMqttToken response = mqttClient.subscribeWithResponse(
-                    new String[]{"/sys/+/+/s/#", "/sys/session/topic/unsubscribed", "/sys/client/disconnected"});
-            if (response.isComplete()) {
-                log.info("subscribe topics success");
-            } else {
-                log.error("subscribe topics failed", mqttToken.getException());
-            }
-        } catch (Throwable e) {
-            log.error("connect to mqtt-broker error", e);
-        }
-    }
-
-    @SneakyThrows
-    @Override
-    public void connectionLost(Throwable e) {
-        log.error("mqtt connection lost", e);
-        while (true) {
-            try {
-                Thread.sleep(1000);
-                if (mqttClient.isConnected()) {
-                    mqttClient.disconnect();
-                }
-                connect();
-                break;
-            } catch (Throwable e1) {
-                log.error("connect error,retry...", e1);
-            }
-        }
-    }
-
-    @Override
-    public void messageArrived(String topic, MqttMessage mqttMessage) {
-        handleMessage(topic, mqttMessage);
-    }
-
-    @Override
-    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
-    }
-
-    public void handleMessage(String topic, MqttMessage msg) {
-        log.info("receive msg,topic:{},msg:{}", topic, JsonUtil.toJsonString(msg));
-        if (topic == null) {
-            log.error("message topic is null.");
-            return;
-        }
-        int code = 0;
-        String mid = "";
-
-        try {
-            String payload = new String(msg.getPayload());
-
-            //取消订阅
-            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")) {
-                RegisterRequest registerRequest = JsonUtil.parse(payload, RegisterRequest.class);
-                mid = registerRequest.getId();
-
-                RegisterInfo registerInfo = RegisterInfo.builder()
-                        .productKey(productKey)
-                        .deviceName(deviceName)
-                        .subDevices(Collections.singletonList(registerRequest.getParams()))
-                        .build();
-                behaviourClient.register(registerInfo);
-                return;
-            }
-
-            Request<?> request = JsonUtil.parse(payload, Request.class);
-            mid = request.getId();
-            MqtMsg mqtMsg = new MqtMsg(topic, request);
-
-            behaviourClient.messageReport(DeviceMessage.builder()
-                    .productKey(productKey)
-                    .deviceName(deviceName)
-                    .mid(request.getId())
-                    .content(JsonUtil.toJsonString(mqtMsg))
-                    .build());
-
-        } catch (Throwable e) {
-            log.error("message process error", e);
-            code = 500;
-        } finally {
-            reply(topic, mid, code);
-        }
-
-    }
-
-    @SneakyThrows
-    private void reply(String topic, String id, int code) {
-        //回复消息不需要再回复
-        if (topic.endsWith("_reply")) {
-            return;
-        }
-
-        topic = topic.replace("/s/", "/c/") + "_reply";
-        String msg = JsonUtil.toJsonString(new Response<>(id, code, new HashMap<>()));
-        mqttClient.publish(topic, new MqttMessage(msg.getBytes()));
-    }
-
-    private void topicUnsubscribed(String msg) {
-        Unsubscribed unsubscribed = JsonUtil.parse(msg, new TypeReference<Unsubscribed>() {
-        });
-        String topic = unsubscribed.getTopic();
-        String[] parts = topic.split("/");
-        if (parts.length < 4) {
-            return;
-        }
-
-        log.info("device offline,pk:{},dn:{}", parts[2], parts[3]);
-        behaviourClient.offline(parts[2], parts[3]);
-    }
-
-    @SneakyThrows
-    public void sendMsg(String topic, String msg) {
-        mqttClient.publish(topic, new MqttMessage(msg.getBytes()));
-    }
-
-    @Data
-    private static class Unsubscribed {
-        private String clientid;
-        private String username;
-        private String topic;
-        private String peerhost;
-    }
-
-    public static class RegisterRequest extends Request<RegisterInfo.SubDevice> {
-
-    }
-
-    @Data
-    @NoArgsConstructor
-    @AllArgsConstructor
-    private static class MqtMsg {
-
-        private String topic;
-
-        private Object payload;
-
-    }
-
-}

device-server/mqtt-server/src/main/resources/application-dev.yml

@@ -1,17 +0,0 @@
-server:
-  port: 8091
-
-spring:
-  data:
-    mongodb:
-      uri: mongodb://填写mongodb地址/admin
-      database: iotkit
-
-  cache:
-    cache-names: foo,bar
-    caffeine:
-      spec: maximumSize=5000,expireAfterAccess=120s
-
-mqtt:
-  url: tcp://填写mqtt连接地址
-

device-server/mqtt-server/src/main/resources/application.yml

@@ -1,16 +0,0 @@
-server:
-  port: 8091
-
-spring:
-  data:
-    mongodb:
-      uri: mongodb://填写mongodb地址/admin
-      database: iotkit
-
-  cache:
-    cache-names: foo,bar
-    caffeine:
-      spec: maximumSize=5000,expireAfterAccess=120s
-
-mqtt:
-  url: tcp://填写mqtt连接地址

device-server/mqtt-server/src/main/resources/logback-spring.xml

@@ -1,87 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<configuration>
-    <jmxConfigurator/>
-    <include resource="org/springframework/boot/logging/logback/defaults.xml"/>
-    <!-- Example for logging into the build folder of your project -->
-    <property name="LOG_FILE" value="log"/>
-
-    <!-- You can override this to have a custom pattern -->
-    <property name="CONSOLE_LOG_PATTERN"
-              value="%clr(%d{yyyy-MM-dd HH:mm:ss.SSS}){faint} %clr(${LOG_LEVEL_PATTERN:-%5p}) %clr(${PID:- }){magenta} %clr(---){faint} %clr([%15.15t]){faint} %clr(%-40.40logger{39}){cyan} %clr(:){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:-%wEx}"/>
-
-    <!-- Appender to log to console -->
-    <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
-        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
-            <!-- Minimum logging level to be presented in the console logs-->
-            <level>DEBUG</level>
-        </filter>
-        <encoder>
-            <pattern>${CONSOLE_LOG_PATTERN}</pattern>
-            <charset>utf8</charset>
-        </encoder>
-    </appender>
-
-    <!-- Appender to log to file -->
-    <appender name="info" class="ch.qos.logback.core.rolling.RollingFileAppender">
-        <file>${LOG_FILE}/info.log</file>
-        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
-            <fileNamePattern>${LOG_FILE}/info.%d{yyyy-MM-dd}.%i.gz</fileNamePattern>
-            <!-- 如果按天来回滚，则最大保存时间为5天，5天之前的都将被清理掉 -->
-            <maxHistory>5</maxHistory>
-            <!-- 日志总保存量为20GB -->
-            <totalSizeCap>20GB</totalSizeCap>
-            <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
-                <!--文件达到 最大1GB时会被压缩和切割 -->
-                <maxFileSize>1GB</maxFileSize>
-            </timeBasedFileNamingAndTriggeringPolicy>
-        </rollingPolicy>
-        <encoder>
-            <pattern>${CONSOLE_LOG_PATTERN}</pattern>
-            <charset>utf8</charset>
-        </encoder>
-    </appender>
-
-    <!-- Appender to log to file only error level log -->
-    <appender name="error" class="ch.qos.logback.core.rolling.RollingFileAppender">
-        <file>${LOG_FILE}/error.log</file>
-        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
-            <fileNamePattern>${LOG_FILE}/error.%d{yyyy-MM-dd}.%i.gz</fileNamePattern>
-            <!-- 如果按天来回滚，则最大保存时间为5天，5天之前的都将被清理掉 -->
-            <maxHistory>5</maxHistory>
-            <!-- 日志总保存量为5GB -->
-            <totalSizeCap>5GB</totalSizeCap>
-            <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
-                <!--文件达到 最大1GB时会被压缩和切割 -->
-                <maxFileSize>1GB</maxFileSize>
-            </timeBasedFileNamingAndTriggeringPolicy>
-        </rollingPolicy>
-        <encoder>
-            <pattern>${CONSOLE_LOG_PATTERN}</pattern>
-            <charset>utf8</charset>
-        </encoder>
-        <filter class="ch.qos.logback.classic.filter.LevelFilter"><!-- 只打印错误日志 -->
-            <level>ERROR</level>
-            <onMatch>ACCEPT</onMatch>
-            <onMismatch>DENY</onMismatch>
-        </filter>
-    </appender>
-
-    <logger name="cc.iotkit" level="debug" additivity="false">
-        <appender-ref ref="info"/>
-        <appender-ref ref="error"/>
-        <appender-ref ref="console"/>
-    </logger>
-
-<!--    <logger name="org.springframework" level="debug" additivity="false">-->
-<!--        <appender-ref ref="info"/>-->
-<!--        <appender-ref ref="error"/>-->
-<!--        <appender-ref ref="console"/>-->
-<!--    </logger>-->
-
-    <root level="INFO">
-        <appender-ref ref="info"/>
-        <appender-ref ref="error"/>
-        <appender-ref ref="console"/>
-    </root>
-
-</configuration>

device-server/pom.xml

@@ -1,21 +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>device-server</artifactId>
-    <packaging>pom</packaging>
-    <modules>
-        <module>mqtt-server</module>
-        <module>device-api</module>
-        <module>mqtt-client-simulator</module>
-    </modules>
-
-
-</project>

manager/pom.xml

@@ -129,11 +129,6 @@
             <artifactId>common</artifactId>
         </dependency>
 
-        <dependency>
-            <groupId>cc.iotkit</groupId>
-            <artifactId>device-api</artifactId>
-        </dependency>
-
         <dependency>
             <groupId>cc.iotkit</groupId>
             <artifactId>rule-engine</artifactId>
@@ -144,11 +139,6 @@
             <artifactId>dao</artifactId>
         </dependency>
 
-        <dependency>
-            <groupId>cc.iotkit</groupId>
-            <artifactId>protocol-server</artifactId>
-        </dependency>
-
         <dependency>
             <groupId>cc.iotkit</groupId>
             <artifactId>component-server</artifactId>

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

@@ -5,9 +5,9 @@ import cc.iotkit.dao.CategoryRepository;
 import cc.iotkit.dao.ProductRepository;
 import cc.iotkit.manager.model.vo.AppDesignVo;
 import cc.iotkit.manager.service.DataOwnerService;
+import cc.iotkit.model.Paging;
 import cc.iotkit.model.product.AppDesign;
 import cc.iotkit.model.product.Category;
-import cc.iotkit.model.PagingData;
 import cc.iotkit.model.product.Product;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
@@ -37,7 +37,7 @@ public class AppController {
     private DataOwnerService dataOwnerService;
 
     @PostMapping("/designs")
-    public PagingData<AppDesignVo> getDesigns() {
+    public Paging<AppDesignVo> getDesigns() {
 
         List<AppDesignVo> appDesignVos = new ArrayList<>();
         List<Product> products = productRepository.findAll(Example
@@ -64,7 +64,7 @@ public class AppController {
             }
         });
 
-        return new PagingData<>(appDesignRepository.count(),
+        return new Paging<>(appDesignRepository.count(),
                 appDesignVos);
     }
 

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

@@ -1,32 +0,0 @@
-package cc.iotkit.manager.controller;
-
-import org.springframework.data.mongodb.repository.MongoRepository;
-import org.springframework.web.bind.annotation.DeleteMapping;
-import org.springframework.web.bind.annotation.GetMapping;
-import org.springframework.web.bind.annotation.PostMapping;
-
-import java.util.List;
-
-public class DbBaseController<R extends MongoRepository<T, String>, T> {
-
-    protected final R repository;
-
-    public DbBaseController(R r) {
-        this.repository = r;
-    }
-
-    @GetMapping("/list")
-    public List<T> list() {
-        return repository.findAll();
-    }
-
-    @PostMapping("/save")
-    public void save(T t) {
-        repository.save(t);
-    }
-
-    @DeleteMapping("/delete")
-    public void delete(T t) {
-        repository.delete(t);
-    }
-}

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

@@ -1,21 +1,18 @@
 package cc.iotkit.manager.controller;
 
-import cc.iotkit.dao.DeviceCache;
-import cc.iotkit.dao.DeviceEventDao;
-import cc.iotkit.dao.DeviceEventRepository;
-import cc.iotkit.dao.DeviceRepository;
+import cc.iotkit.dao.*;
 import cc.iotkit.manager.service.DataOwnerService;
 import cc.iotkit.manager.service.DeviceService;
 import cc.iotkit.manager.utils.AuthUtil;
-import cc.iotkit.model.device.message.DeviceEvent;
+import cc.iotkit.model.Paging;
 import cc.iotkit.model.device.DeviceInfo;
-import cc.iotkit.model.PagingData;
+import cc.iotkit.model.device.message.DeviceProperty;
+import cc.iotkit.model.device.message.ThingModelMessage;
 import cc.iotkit.model.product.ThingModel;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.domain.Example;
-import org.springframework.data.domain.Sort;
 import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.web.bind.annotation.*;
 
@@ -32,15 +29,15 @@ public class DeviceController {
     @Autowired
     private DeviceRepository deviceRepository;
     @Autowired
-    private DeviceEventRepository deviceEventRepository;
-    @Autowired
-    private DeviceEventDao deviceEventDao;
-    @Autowired
-    private DeviceCache deviceCache;
+    private DeviceDao deviceDao;
     @Autowired
     private DataOwnerService dataOwnerService;
     @Autowired
     private ProductController productController;
+    @Autowired
+    private ThingModelMessageDao thingModelMessageDao;
+    @Autowired
+    private DevicePropertyDao devicePropertyDao;
 
     @PostMapping("/{deviceId}/service/{service}")
     public String invokeService(@PathVariable("deviceId") String deviceId,
@@ -60,11 +57,11 @@ public class DeviceController {
     }
 
     @PostMapping("/list")
-    public PagingData<DeviceInfo> getDevices(int page,
-                                             int limit,
-                                             String pk,
-                                             Boolean online,
-                                             String dn) {
+    public Paging<DeviceInfo> getDevices(int page,
+                                         int size,
+                                         String pk,
+                                         Boolean online,
+                                         String dn) {
         Criteria condition = new Criteria();
         if (!AuthUtil.isAdmin()) {
             condition.and("uid").is(AuthUtil.getUserId());
@@ -78,8 +75,8 @@ public class DeviceController {
         if (online != null) {
             condition.and("state.online").is(online);
         }
-        return new PagingData<>(deviceCache.count(condition),
-                deviceCache.find(condition, (page - 1) * limit, limit, Sort.Order.desc("createAt")));
+
+        return deviceDao.find(condition, size, page);
     }
 
     @GetMapping("/{deviceId}/children")
@@ -112,29 +109,28 @@ public class DeviceController {
         deviceRepository.deleteById(deviceId);
     }
 
-    @PostMapping("/{deviceId}/events")
-    public PagingData<DeviceEvent> events(@PathVariable("deviceId") String deviceId,
-                                          int page,
-                                          int limit,
-                                          String type,
-                                          String identifier) {
-        Criteria condition = Criteria.where("deviceId").is(deviceId);
-        if (StringUtils.isNotBlank(type)) {
-            condition.and("type").is(type);
-        }
-        if (StringUtils.isNotBlank(identifier)) {
-            condition.and("identifier").regex(".*" + identifier + ".*");
-        }
+    @PostMapping("/{deviceId}/logs/{size}/{page}")
+    public Paging<ThingModelMessage> logs(
+            @PathVariable("deviceId") String deviceId,
+            @PathVariable("size") int size,
+            @PathVariable("page") int page,
+            String type, String identifier) {
+        return thingModelMessageDao.findByTypeAndIdentifier(deviceId, type, identifier, page, size);
+    }
 
-        return new PagingData<>(deviceEventDao.count(condition),
-                deviceEventDao.find(condition,
-                        (page - 1) * limit, limit, Sort.Order.desc("createAt")));
+    @GetMapping("/{deviceId}/property/{name}/{start}/{end}")
+    public List<DeviceProperty> getPropertyHistory(
+            @PathVariable("deviceId") String deviceId,
+            @PathVariable("name") String name,
+            @PathVariable("start") long start,
+            @PathVariable("end") long end) {
+        return devicePropertyDao.findDevicePropertyHistory(deviceId, name, start, end);
     }
 
     @PostMapping("/{deviceId}/unbind")
     public void unbindDevice(@PathVariable("deviceId") String deviceId) {
         deviceId = getDetail(deviceId).getDeviceId();
-        deviceService.unbindDevice(deviceId);
+//        deviceService.unbindDevice(deviceId);
     }
 
     @GetMapping("/{deviceId}/thingModel")

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

@@ -6,8 +6,8 @@ import cc.iotkit.dao.ProductRepository;
 import cc.iotkit.dao.ThingModelRepository;
 import cc.iotkit.manager.config.AliyunConfig;
 import cc.iotkit.manager.service.DataOwnerService;
+import cc.iotkit.model.Paging;
 import cc.iotkit.model.product.Category;
-import cc.iotkit.model.PagingData;
 import cc.iotkit.model.product.Product;
 import cc.iotkit.model.product.ThingModel;
 import com.aliyun.oss.OSS;
@@ -42,9 +42,9 @@ public class ProductController {
     private OSS ossClient;
 
     @PostMapping("/list")
-    public PagingData<Product> getProducts(Product form) {
+    public Paging<Product> getProducts(Product form) {
         form = dataOwnerService.wrapExample(form);
-        return new PagingData<>(productRepository.count(Example.of(form)),
+        return new Paging<>(productRepository.count(Example.of(form)),
                 productRepository.findAll(Example.of(form)));
     }
 

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

@@ -2,6 +2,7 @@ package cc.iotkit.manager.controller;
 
 import cc.iotkit.common.exception.BizException;
 import cc.iotkit.common.utils.ReflectUtil;
+import cc.iotkit.comp.CompConfig;
 import cc.iotkit.comp.mqtt.MqttComponent;
 import cc.iotkit.comps.ComponentManager;
 import cc.iotkit.converter.ScriptConverter;
@@ -12,7 +13,6 @@ import cc.iotkit.manager.utils.AuthUtil;
 import cc.iotkit.model.Paging;
 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;
@@ -22,7 +22,6 @@ import org.springframework.data.domain.PageRequest;
 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;
@@ -38,9 +37,6 @@ public class ProtocolController {
     @Autowired
     private ProtocolGatewayRepository gatewayRepository;
 
-    @Autowired
-    private GatewayService gatewayService;
-
     @Autowired
     private DataOwnerService dataOwnerService;
 
@@ -66,7 +62,6 @@ public class ProtocolController {
             gateway.setCreateAt(System.currentTimeMillis());
             gateway.setUid(AuthUtil.getUserId());
             gateway.setUuid(optUser.get().getUid());
-            gatewayService.saveFunction(gateway.getUuid(), gateway.getId(), gateway.getScript(), functionJar);
             gatewayRepository.save(gateway);
         } catch (Throwable e) {
             throw new BizException("add protocol gateway error", e);
@@ -89,7 +84,6 @@ public class ProtocolController {
         dataOwnerService.checkOwner(gateway);
         try {
             gatewayRepository.save(gateway);
-            gatewayService.saveFunction(gateway.getUuid(), gateway.getId(), gateway.getScript(), functionJar);
         } catch (Throwable e) {
             throw new BizException("add protocol gateway error", e);
         }
@@ -105,8 +99,8 @@ public class ProtocolController {
         ProtocolGateway oldGateway = optGateway.get();
         oldGateway.setScript(gateway.getScript());
         try {
-            gatewayService.saveFunction(oldGateway.getUuid(), oldGateway.getId(),
-                    "new (function (){" + oldGateway.getScript() + "})()", functionJar);
+//            gatewayService.saveFunction(oldGateway.getUuid(), oldGateway.getId(),
+//                    "new (function (){" + oldGateway.getScript() + "})()", functionJar);
             gatewayRepository.save(oldGateway);
         } catch (Throwable e) {
             throw new BizException("save protocol gateway script error", e);
@@ -118,7 +112,6 @@ public class ProtocolController {
         dataOwnerService.checkOwner(gatewayRepository, id);
         try {
             gatewayRepository.deleteById(id);
-            gatewayService.deleteFunction(AuthUtil.getUserId(), id);
         } catch (Throwable e) {
             throw new BizException("delete protocol gateway error", e);
         }
@@ -136,7 +129,7 @@ public class ProtocolController {
     @GetMapping("/registerMqtt")
     public void registerMqtt() throws IOException {
         MqttComponent component = new MqttComponent();
-        component.create("{\"port\":2883,\"ssl\":false}");
+        component.create(new CompConfig(300, "{\"port\":2883,\"ssl\":false}"));
         ScriptConverter converter = new ScriptConverter();
         converter.setScript(FileUtils.readFileToString(new File("/Users/sjg/home/gitee/open-source/converter.js"), "UTF-8"));
         component.setConverter(converter);

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

@@ -23,8 +23,6 @@ import java.util.stream.Collectors;
 @RequestMapping("/space")
 public class SpaceController {
 
-    @Autowired
-    private UserInfoDao userInfoDao;
     @Autowired
     private SpaceDeviceRepository spaceDeviceRepository;
     @Autowired
@@ -34,24 +32,24 @@ public class SpaceController {
     @Autowired
     private ProductCache productCache;
 
-    @PostMapping("/list")
-    public PagingData<SpaceInfo> getDevices(int page,
-                                            int limit,
-                                            String address) {
-        Criteria condition = new Criteria();
-        if (StringUtils.isNotBlank(address)) {
-            condition.and("address").regex(".*" + address + ".*");
-        }
-        List<UserInfo> userInfoList = userInfoDao.find(condition, (page - 1) * limit,
-                limit, Sort.Order.desc("createAt"));
-
-        List<SpaceInfo> spaces = userInfoList.stream().map((u ->
-                new SpaceInfo(u.getAddress(), u.getUid())))
-                .collect(Collectors.toList());
-
-        return new PagingData<>(userInfoDao.count(condition),
-                spaces);
-    }
+//    @PostMapping("/list")
+//    public Paging<SpaceInfo> getDevices(int page,
+//                                            int limit,
+//                                            String address) {
+//        Criteria condition = new Criteria();
+//        if (StringUtils.isNotBlank(address)) {
+//            condition.and("address").regex(".*" + address + ".*");
+//        }
+//        List<UserInfo> userInfoList = userInfoDao.find(condition, (page - 1) * limit,
+//                limit, Sort.Order.desc("createAt"));
+//
+//        List<SpaceInfo> spaces = userInfoList.stream().map((u ->
+//                new SpaceInfo(u.getAddress(), u.getUid())))
+//                .collect(Collectors.toList());
+//
+//        return new Paging<>(userInfoDao.count(condition),
+//                spaces);
+//    }
 
     @GetMapping("/{userId}/devices")
     public List<SpaceDeviceVo> getDevices(@PathVariable("userId") String userId) {
@@ -65,7 +63,7 @@ public class SpaceController {
                     .name(sd.getName())
                     .picUrl(product.getImg())
                     .spaceName(sd.getSpaceName())
-                    .online(deviceInfo.getState().getOnline())
+                    .online(deviceInfo.getState().isOnline())
                     .property(deviceInfo.getProperty())
                     .productKey(deviceInfo.getProductKey())
                     .build());

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

@@ -1,18 +0,0 @@
-package cc.iotkit.manager.controller;
-
-import cc.iotkit.dao.UserAccountRepository;
-import cc.iotkit.model.UserAccount;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RestController;
-
-@RestController
-@RequestMapping("/userAccount")
-public class UserAccountController extends DbBaseController<UserAccountRepository, UserAccount> {
-
-    @Autowired
-    public UserAccountController(UserAccountRepository userAccountRepository) {
-        super(userAccountRepository);
-    }
-
-}

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

@@ -4,7 +4,6 @@ import cc.iotkit.common.Constants;
 import cc.iotkit.common.exception.BizException;
 import cc.iotkit.common.utils.ReflectUtil;
 import cc.iotkit.dao.UserInfoRepository;
-import cc.iotkit.manager.service.AligenieService;
 import cc.iotkit.manager.service.KeycloakAdminService;
 import cc.iotkit.manager.service.PulsarAdminService;
 import cc.iotkit.manager.utils.AuthUtil;
@@ -21,25 +20,21 @@ import java.util.UUID;
 
 @RestController
 @RequestMapping("/user")
-public class UserInfoController extends DbBaseController<UserInfoRepository, UserInfo> {
+public class UserInfoController {
 
     @Value("${app.systemRole}")
     private String systemRole;
 
     private final KeycloakAdminService keycloakAdminService;
     private final UserInfoRepository userInfoRepository;
-    private final AligenieService aligenieService;
     private final PulsarAdminService pulsarAdminService;
 
     @Autowired
     public UserInfoController(UserInfoRepository userInfoRepository,
                               KeycloakAdminService keycloakAdminService,
-                              AligenieService aligenieService,
                               PulsarAdminService pulsarAdminService) {
-        super(userInfoRepository);
         this.keycloakAdminService = keycloakAdminService;
         this.userInfoRepository = userInfoRepository;
-        this.aligenieService = aligenieService;
         this.pulsarAdminService = pulsarAdminService;
     }
 
@@ -113,11 +108,5 @@ public class UserInfoController extends DbBaseController<UserInfoRepository, Use
         }
         ReflectUtil.copyNoNulls(user, oldUser);
         userInfoRepository.save(oldUser);
-
-        boolean isAligenie = user.getUsePlatforms().isAligenie();
-        //同步天猫精灵设备
-        if (oldUser.getUsePlatforms().isAligenie() != isAligenie) {
-            aligenieService.syncDevice(user);
-        }
     }
 }

manager/src/main/java/cc/iotkit/manager/controller/aligenie/AligenieDeviceController.java

@@ -1,19 +0,0 @@
-package cc.iotkit.manager.controller.aligenie;
-
-import cc.iotkit.dao.AligenieDeviceRepository;
-import cc.iotkit.manager.controller.DbBaseController;
-import cc.iotkit.model.aligenie.AligenieDevice;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RestController;
-
-@RestController
-@RequestMapping("/aligenieDevice")
-public class AligenieDeviceController extends DbBaseController<AligenieDeviceRepository, AligenieDevice> {
-
-    @Autowired
-    public AligenieDeviceController(AligenieDeviceRepository aligenieDeviceRepository) {
-        super(aligenieDeviceRepository);
-    }
-
-}

manager/src/main/java/cc/iotkit/manager/controller/aligenie/AligenieProductController.java

@@ -1,14 +1,12 @@
 package cc.iotkit.manager.controller.aligenie;
 
-import cc.iotkit.dao.AligenieProductDao;
 import cc.iotkit.dao.AligenieProductRepository;
 import cc.iotkit.dao.ProductCache;
-import cc.iotkit.manager.controller.DbBaseController;
 import cc.iotkit.manager.model.aligenie.AligenieProductVo;
 import cc.iotkit.manager.service.DataOwnerService;
 import cc.iotkit.manager.utils.AuthUtil;
-import cc.iotkit.model.product.Product;
 import cc.iotkit.model.aligenie.AligenieProduct;
+import cc.iotkit.model.product.Product;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.domain.Example;
 import org.springframework.web.bind.annotation.GetMapping;
@@ -21,28 +19,20 @@ import java.util.List;
 
 @RestController
 @RequestMapping("/aligenie/product")
-public class AligenieProductController extends DbBaseController<AligenieProductRepository, AligenieProduct> {
-
-    private final ProductCache productCache;
-    private final AligenieProductDao aligenieProductDao;
-    private final DataOwnerService dataOwnerService;
-
+public class AligenieProductController {
 
     @Autowired
-    public AligenieProductController(AligenieProductRepository aligenieProductRepository,
-                                     ProductCache productCache,
-                                     AligenieProductDao aligenieProductDao,
-                                     DataOwnerService dataOwnerService) {
-        super(aligenieProductRepository);
-        this.productCache = productCache;
-        this.aligenieProductDao = aligenieProductDao;
-        this.dataOwnerService = dataOwnerService;
-    }
+    private ProductCache productCache;
+    @Autowired
+    private DataOwnerService dataOwnerService;
+    @Autowired
+    private AligenieProductRepository aligenieProductRepository;
+
 
     @GetMapping("/products")
     public List<AligenieProductVo> products() {
         List<AligenieProductVo> productVos = new ArrayList<>();
-        List<AligenieProduct> aligenieProducts = repository
+        List<AligenieProduct> aligenieProducts = aligenieProductRepository
                 .findAll(Example
                         .of(AligenieProduct.builder()
                                 .uid(AuthUtil.getUserId())
@@ -61,7 +51,7 @@ public class AligenieProductController extends DbBaseController<AligenieProductR
             product.setCreateAt(System.currentTimeMillis());
         }
 
-        dataOwnerService.checkOwnerSave(repository,product);
-        aligenieProductDao.save(product.getProductId(), product);
+        dataOwnerService.checkOwnerSave(aligenieProductRepository, product);
+        aligenieProductRepository.save(product);
     }
 }

manager/src/main/java/cc/iotkit/manager/controller/api/SpaceController.java

@@ -180,7 +180,7 @@ public class SpaceController {
                 .deviceName(deviceInfo.getDeviceName())
                 .picUrl(product.getImg())
                 .spaceName(spaceName)
-                .online(deviceInfo.getState().getOnline())
+                .online(deviceInfo.getState().isOnline())
                 .property(deviceInfo.getProperty() == null ? new HashMap<>() : deviceInfo.getProperty())
                 .productKey(deviceInfo.getProductKey())
                 .build();

manager/src/main/java/cc/iotkit/manager/model/vo/DeviceLog.java

@@ -0,0 +1,9 @@
+package cc.iotkit.manager.model.vo;
+
+import lombok.Data;
+
+@Data
+public class DeviceLog {
+
+
+}

manager/src/main/java/cc/iotkit/manager/service/AligenieService.java

@@ -1,90 +0,0 @@
-package cc.iotkit.manager.service;
-
-import cc.iotkit.dao.AligenieDeviceRepository;
-import cc.iotkit.dao.AligenieProductDao;
-import cc.iotkit.dao.DeviceCache;
-import cc.iotkit.dao.SpaceDeviceRepository;
-import cc.iotkit.model.device.DeviceInfo;
-import cc.iotkit.model.space.SpaceDevice;
-import cc.iotkit.model.UserInfo;
-import cc.iotkit.model.aligenie.AligenieDevice;
-import cc.iotkit.model.aligenie.AligenieProduct;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.data.domain.Example;
-import org.springframework.stereotype.Service;
-
-import java.util.List;
-
-@Service
-public class AligenieService {
-
-    @Autowired
-    private SpaceDeviceRepository spaceDeviceRepository;
-
-    @Autowired
-    private AligenieDeviceRepository aligenieDeviceRepository;
-
-    @Autowired
-    private AligenieProductDao aligenieProductDao;
-
-    @Autowired
-    private DeviceCache deviceCache;
-
-    public void syncDevice(UserInfo user) {
-        if (!user.getUsePlatforms().isAligenie()) {
-            //清空
-            List<AligenieDevice> aligenieDevices = aligenieDeviceRepository.findAll(Example.of(
-                    AligenieDevice.builder()
-                            .uid(user.getId())
-                            .build()));
-            for (AligenieDevice aligenieDevice : aligenieDevices) {
-                aligenieDeviceRepository.delete(aligenieDevice);
-            }
-            return;
-        }
-
-        //找出用户空间下所有设备
-        List<SpaceDevice> spaceDeviceList = spaceDeviceRepository.findAll(Example.of(
-                SpaceDevice.builder().uid(user.getId()).build()
-        ));
-
-        for (SpaceDevice spaceDevice : spaceDeviceList) {
-            AligenieDevice aligenieDevice = aligenieDeviceRepository.findOne(Example.of(
-                    AligenieDevice.builder()
-                            .deviceId(spaceDevice.getDeviceId())
-                            .uid(user.getId())
-                            .build()
-            )).orElse(null);
-
-            //不存在设备，新增
-            if (aligenieDevice == null) {
-                DeviceInfo deviceInfo = deviceCache.findByDeviceId(spaceDevice.getDeviceId());
-                AligenieProduct aligenieProduct = aligenieProductDao.getAligenieProduct(deviceInfo.getProductKey());
-                if(aligenieProduct==null){
-                    continue;
-                }
-
-                aligenieDeviceRepository.save(
-                        AligenieDevice.builder()
-                                .uid(user.getId())
-                                .deviceId(spaceDevice.getDeviceId())
-                                .name(spaceDevice.getName())
-                                .spaceName(spaceDevice.getSpaceName())
-                                .productId(aligenieProduct.getProductId())
-                                .build()
-                );
-            } else {
-                //存在，更新设备信息
-                aligenieDeviceRepository.save(
-                        AligenieDevice.builder()
-                                .id(aligenieDevice.getId())
-                                .name(spaceDevice.getName())
-                                .spaceName(spaceDevice.getSpaceName())
-                                .build()
-                );
-            }
-
-        }
-    }
-
-}

manager/src/main/java/cc/iotkit/manager/service/DeviceService.java

@@ -1,48 +1,49 @@
 package cc.iotkit.manager.service;
 
+import cc.iotkit.common.exception.BizException;
 import cc.iotkit.common.exception.NotFoundException;
-import cc.iotkit.dao.DeviceCache;
-import cc.iotkit.dao.DeviceEventRepository;
+import cc.iotkit.common.utils.UniqueIdUtil;
+import cc.iotkit.comps.ComponentManager;
+import cc.iotkit.converter.ThingService;
 import cc.iotkit.dao.DeviceRepository;
-import cc.iotkit.dao.ThingModelRepository;
-import cc.iotkit.deviceapi.IDeviceManager;
 import cc.iotkit.model.device.DeviceInfo;
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.NoArgsConstructor;
+import cc.iotkit.model.device.message.ThingModelMessage;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.data.domain.Example;
 import org.springframework.stereotype.Service;
 
-import java.util.List;
 import java.util.Map;
 
 @Slf4j
 @Service
 public class DeviceService {
 
-    @Autowired
-    private DeviceCache deviceCache;
     @Autowired
     private DeviceRepository deviceRepository;
     @Autowired
-    private ThingModelRepository thingModelRepository;
-    @Autowired
-    private ThingModelService thingModelService;
-    @Autowired
-    private DeviceEventRepository deviceEventRepository;
-    @Autowired
     private DataOwnerService dataOwnerService;
     @Autowired
-    private IDeviceManager deviceManager;
+    private ComponentManager componentManager;
 
     public String invokeService(String deviceId, String service, Map<String, Object> args) {
         DeviceInfo device = deviceRepository.findById(deviceId)
                 .orElseThrow(() -> new NotFoundException("device not found by deviceId"));
 
         dataOwnerService.checkOwner(device);
-        return this.deviceManager.invokeService(deviceId, service, args);
+        if (!device.getState().isOnline()) {
+            throw new BizException("device is offline");
+        }
+
+        ThingService<?> thingService = ThingService.builder()
+                .mid(UniqueIdUtil.newRequestId())
+                .productKey(device.getProductKey())
+                .deviceName(device.getDeviceName())
+                .type(ThingModelMessage.TYPE_SERVICE)
+                .identifier(service)
+                .params(args)
+                .build();
+        componentManager.send(thingService);
+        return thingService.getMid();
     }
 
     public String setProperty(String deviceId, Map<String, Object> properties) {
@@ -50,26 +51,20 @@ public class DeviceService {
                 .orElseThrow(() -> new NotFoundException("device not found by deviceId"));
 
         dataOwnerService.checkOwner(device);
-        return deviceManager.setProperty(deviceId, properties);
-    }
+        if (!device.getState().isOnline()) {
+            throw new BizException("device is offline");
+        }
 
-    public void unbindDevice(String deviceId) {
-        deviceManager.unbind(deviceId);
+        ThingService<?> thingService = ThingService.builder()
+                .mid(UniqueIdUtil.newRequestId())
+                .productKey(device.getProductKey())
+                .deviceName(device.getDeviceName())
+                .type(ThingModelMessage.TYPE_PROPERTY)
+                .identifier("set")
+                .params(properties)
+                .build();
+        componentManager.send(thingService);
+        return thingService.getMid();
     }
 
-    public List<DeviceInfo> findDevices(DeviceInfo form) {
-        return deviceRepository.findAll(Example.of(form));
-    }
-
-    public long count(DeviceInfo form) {
-        return deviceRepository.count(Example.of(form));
-    }
-
-    @Data
-    @NoArgsConstructor
-    @AllArgsConstructor
-    private static class CmdRequest {
-        private String id;
-        private Object params;
-    }
 }

manager/src/main/java/cc/iotkit/manager/service/SpaceDeviceService.java

@@ -41,7 +41,7 @@ public class SpaceDeviceService {
                     .name(sd.getName())
                     .picUrl(product.getImg())
                     .spaceName(sd.getSpaceName())
-                    .online(deviceInfo.getState().getOnline())
+                    .online(deviceInfo.getState().isOnline())
                     .property(deviceInfo.getProperty())
                     .productKey(deviceInfo.getProductKey())
                     .build());

model/src/main/java/cc/iotkit/model/PagingData.java

@@ -1,18 +0,0 @@
-package cc.iotkit.model;
-
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-
-import java.util.List;
-
-@Data
-@NoArgsConstructor
-@AllArgsConstructor
-public class PagingData<T> {
-
-    private long count;
-
-    private List<T> data;
-
-}

model/src/main/java/cc/iotkit/model/device/DeviceInfo.java

@@ -51,11 +51,12 @@ public class DeviceInfo implements Owned {
     @AllArgsConstructor
     public static class State {
 
-        private Boolean online;
+        private boolean online;
 
         private Long onlineTime;
 
         private Long offlineTime;
 
     }
+
 }

model/src/main/java/cc/iotkit/model/device/message/DeviceProperty.java

@@ -0,0 +1,29 @@
+package cc.iotkit.model.device.message;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.elasticsearch.annotations.Document;
+import org.springframework.data.elasticsearch.annotations.Field;
+import org.springframework.data.elasticsearch.annotations.FieldType;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+@Document(indexName = "device_property")
+public class DeviceProperty {
+
+    @Id
+    private String mid;
+
+    private String deviceId;
+
+    private String name;
+
+    private Object value;
+
+    @Field(type = FieldType.Date)
+    private Long time;
+
+}

model/src/main/java/cc/iotkit/model/device/message/ThingModelMessage.java

@@ -21,22 +21,28 @@ import java.util.Map;
 @Document(indexName = "thing_model_messages")
 public class ThingModelMessage {
 
-    public static final String TYPE_PROPERTY="property";
-    public static final String TYPE_EVENT="event";
-    public static final String TYPE_SERVICE="service";
+    public static final String TYPE_LIFETIME = "lifetime";
+    public static final String TYPE_STATE = "state";
+    public static final String TYPE_PROPERTY = "property";
+    public static final String TYPE_EVENT = "event";
+    public static final String TYPE_SERVICE = "service";
 
-    public static final String ID_PROPERTY_GET="get";
-    public static final String ID_PROPERTY_SET="set";
+    public static final String ID_PROPERTY_GET = "get";
+    public static final String ID_PROPERTY_SET = "set";
 
     @Id
     private String mid;
 
+    private String deviceId;
+
     private String productKey;
 
     private String deviceName;
 
     /**
      * 消息类型
+     * lifetime:生命周期
+     * state:状态
      * property:属性
      * event:事件
      * service:服务
@@ -45,13 +51,18 @@ public class ThingModelMessage {
 
     private String identifier;
 
-    private Map<String,Object> data;
+    /**
+     * 消息状态码
+     */
+    private int code;
+
+    private Object data;
 
     /**
      * 时间戳，设备上的事件或数据产生的本地时间
      */
     @Field(type = FieldType.Date)
-    private Long occur;
+    private Long occurred;
 
     /**
      * 消息上报时间

pom.xml

@@ -4,7 +4,6 @@
     <modelVersion>4.0.0</modelVersion>
     <packaging>pom</packaging>
     <modules>
-        <module>device-server</module>
         <module>model</module>
         <module>rule-engine</module>
         <module>common</module>
@@ -181,13 +180,13 @@
 
             <dependency>
                 <groupId>io.vertx</groupId>
-                <artifactId>vertx-web</artifactId>
+                <artifactId>vertx-mqtt</artifactId>
                 <version>4.2.6</version>
             </dependency>
 
             <dependency>
                 <groupId>io.vertx</groupId>
-                <artifactId>vertx-mqtt</artifactId>
+                <artifactId>vertx-web-proxy</artifactId>
                 <version>4.2.6</version>
             </dependency>
 
@@ -209,36 +208,12 @@
                 <version>${project.version}</version>
             </dependency>
 
-            <dependency>
-                <groupId>cc.iotkit</groupId>
-                <artifactId>device-api</artifactId>
-                <version>${project.version}</version>
-            </dependency>
-
             <dependency>
                 <groupId>cc.iotkit</groupId>
                 <artifactId>rule-engine</artifactId>
                 <version>${project.version}</version>
             </dependency>
 
-            <dependency>
-                <groupId>cc.iotkit</groupId>
-                <artifactId>gateway-client</artifactId>
-                <version>${project.version}</version>
-            </dependency>
-
-            <dependency>
-                <groupId>cc.iotkit</groupId>
-                <artifactId>decode-function</artifactId>
-                <version>${project.version}</version>
-            </dependency>
-
-            <dependency>
-                <groupId>cc.iotkit</groupId>
-                <artifactId>protocol-server</artifactId>
-                <version>${project.version}</version>
-            </dependency>
-
             <dependency>
                 <groupId>cc.iotkit</groupId>
                 <artifactId>component</artifactId>

protocol-gateway/component-server/pom.xml

@@ -3,10 +3,9 @@
          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>
+        <artifactId>protocol-gateway</artifactId>
         <groupId>cc.iotkit</groupId>
         <version>0.0.1-SNAPSHOT</version>
-        <relativePath>../../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
@@ -34,6 +33,15 @@
             <artifactId>spring-context</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.springframework.data</groupId>
+            <artifactId>spring-data-redis</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>redis.clients</groupId>
+            <artifactId>jedis</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>

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

@@ -1,14 +1,24 @@
 package cc.iotkit.comps;
 
 
+import cc.iotkit.common.exception.BizException;
+import cc.iotkit.common.utils.JsonUtil;
 import cc.iotkit.comp.IComponent;
+import cc.iotkit.comps.config.CacheKey;
 import cc.iotkit.comps.service.DeviceBehaviourService;
+import cc.iotkit.converter.DeviceMessage;
+import cc.iotkit.converter.ThingService;
+import cc.iotkit.model.device.message.ThingModelMessage;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Component;
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 
+@Slf4j
 @Component
 public class ComponentManager {
 
@@ -16,6 +26,8 @@ public class ComponentManager {
 
     @Autowired
     private DeviceBehaviourService deviceBehaviourService;
+    @Autowired
+    private StringRedisTemplate redisTemplate;
 
     public void register(String id, IComponent component) {
         components.put(id, component);
@@ -31,8 +43,10 @@ public class ComponentManager {
         if (component == null) {
             return;
         }
-        component.setHandler(new MessageHandler(script, component.getConverter(),
-                deviceBehaviourService));
+        component.setHandler(
+                new MessageHandler(this, component,
+                        script, component.getConverter(),
+                        deviceBehaviourService));
         component.start();
     }
 
@@ -44,4 +58,42 @@ public class ComponentManager {
         component.stop();
     }
 
+    public void send(ThingService<?> service) {
+        log.info("start exec device service:{}", JsonUtil.toJsonString(service));
+        if (components.size() == 0) {
+            throw new BizException("there is no components");
+        }
+
+        for (IComponent com : components.values()) {
+            if (com.exist(service.getProductKey(), service.getDeviceName())) {
+                DeviceMessage message = com.getConverter().encode(service, null);
+                if (message == null) {
+                    throw new BizException("encode send message failed");
+                }
+                //保存设备端mid与平台mid对应关系
+                redisTemplate.opsForValue().set(
+                        CacheKey.getKeyCmdMid(service.getDeviceName(), message.getMid()),
+                        service.getMid(), com.getConfig().getCmdTimeout(), TimeUnit.SECONDS);
+                com.send(message);
+
+                ThingModelMessage thingModelMessage = ThingModelMessage.builder()
+                        .mid(service.getMid())
+                        .productKey(service.getProductKey())
+                        .deviceName(service.getDeviceName())
+                        .identifier(service.getIdentifier())
+                        .type(service.getType())
+                        .data(service.getParams())
+                        .build();
+                deviceBehaviourService.reportMessage(thingModelMessage);
+
+                return;
+            }
+        }
+        throw new BizException("send destination not found");
+    }
+
+    public String getPlatformMid(String deviceName, String mid) {
+        return redisTemplate.opsForValue().get(CacheKey.getKeyCmdMid(deviceName, mid));
+    }
+
 }

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

@@ -1,13 +1,18 @@
 package cc.iotkit.comps;
 
 import cc.iotkit.common.exception.BizException;
+import cc.iotkit.common.utils.JsonUtil;
+import cc.iotkit.common.utils.UniqueIdUtil;
+import cc.iotkit.comp.IComponent;
 import cc.iotkit.comp.IMessageHandler;
-import cc.iotkit.comp.model.DeviceMessage;
-import cc.iotkit.comps.model.AuthInfo;
-import cc.iotkit.comps.model.DeviceState;
-import cc.iotkit.comps.model.RegisterInfo;
+import cc.iotkit.comp.model.AuthInfo;
+import cc.iotkit.comp.model.ReceiveResult;
+import cc.iotkit.comp.model.RegisterInfo;
+import cc.iotkit.converter.DeviceMessage;
+import cc.iotkit.comp.model.DeviceState;
 import cc.iotkit.comps.service.DeviceBehaviourService;
 import cc.iotkit.converter.IConverter;
+import cc.iotkit.model.device.message.ThingModelMessage;
 import jdk.nashorn.api.scripting.NashornScriptEngine;
 import jdk.nashorn.api.scripting.ScriptObjectMirror;
 import lombok.Data;
@@ -30,32 +35,32 @@ public class MessageHandler implements IMessageHandler {
 
     private final DeviceBehaviourService deviceBehaviourService;
 
+    private final ComponentManager componentManager;
+
+    private final IComponent component;
+
     @SneakyThrows
-    public MessageHandler(String script, IConverter converter,
+    public MessageHandler(ComponentManager componentManager,
+                          IComponent component,
+                          String script, IConverter converter,
                           DeviceBehaviourService deviceBehaviourService) {
+        this.componentManager = componentManager;
+        this.component = component;
         this.converter = converter;
         this.deviceBehaviourService = deviceBehaviourService;
         scriptObj = 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) {
+    public ReceiveResult onReceive(Map<String, Object> head, String type, String msg) {
         try {
             ScriptObjectMirror result = (ScriptObjectMirror) engine.invokeMethod(scriptObj, "onReceive", head, type, msg);
+            log.info("onReceive script result:{}", JsonUtil.toJsonString(result));
             Object rstType = result.get("type");
             if (rstType == null) {
-                return;
+                return null;
             }
             //取脚本执行后返回的数据
-            Object data = result.get("data");
+            Object data = JsonUtil.toObject((ScriptObjectMirror) result.get("data"));
             if (!(data instanceof Map)) {
                 throw new BizException("script result data is incorrect");
             }
@@ -63,24 +68,32 @@ public class MessageHandler implements IMessageHandler {
 
             if ("register".equals(rstType)) {
                 //注册数据
-                RegisterInfo regInfo = new RegisterInfo();
-                BeanUtils.populate(regInfo, dataMap);
+                RegisterInfo regInfo = RegisterInfo.from(dataMap);
+                if (regInfo == null) {
+                    return null;
+                }
                 doRegister(regInfo);
+                return new ReceiveResult(regInfo.getProductKey(), regInfo.getDeviceName(), regInfo);
             } else if ("auth".equals(rstType)) {
                 //设备认证
                 AuthInfo authInfo = new AuthInfo();
                 BeanUtils.populate(authInfo, dataMap);
                 doAuth(authInfo);
+                return new ReceiveResult(authInfo.getProductKey(), authInfo.getDeviceName(), authInfo);
             } else if ("state".equals(rstType)) {
                 //设备状态变更
-                DeviceState state = new DeviceState();
-                BeanUtils.populate(state, dataMap);
+                DeviceState state = DeviceState.from(dataMap);
+                if (state == null) {
+                    return null;
+                }
                 doStateChange(state);
+                return new ReceiveResult(state.getProductKey(), state.getDeviceName(), state);
             } else if ("report".equals(rstType)) {
                 //上报数据
                 DeviceMessage message = new DeviceMessage();
                 BeanUtils.populate(message, dataMap);
                 doReport(message);
+                return new ReceiveResult(message.getProductKey(), message.getDeviceName(), message);
             }
 
         } catch (BizException e) {
@@ -88,6 +101,7 @@ public class MessageHandler implements IMessageHandler {
         } catch (Throwable e) {
             throw new BizException("receive component message error", e);
         }
+        return null;
     }
 
     private void doRegister(RegisterInfo reg) throws ScriptException, NoSuchMethodException {
@@ -115,6 +129,7 @@ public class MessageHandler implements IMessageHandler {
 
     private void doStateChange(DeviceState state) {
         try {
+            component.onDeviceStateChange(state);
             deviceBehaviourService.deviceStateChange(state.getProductKey(),
                     state.getDeviceName(),
                     DeviceState.STATE_ONLINE.equals(state.getState()));
@@ -124,10 +139,20 @@ public class MessageHandler implements IMessageHandler {
     }
 
     private void doReport(DeviceMessage message) {
-        try {
-            deviceBehaviourService.reportMessage(message);
-        } catch (Throwable e) {
-            log.error("report device message error", e);
+        ThingModelMessage thingModelMessage = converter.decode(message);
+
+        //服务回复需要重新对应mid
+        if (thingModelMessage.getIdentifier().endsWith("_reply")) {
+            String platformMid = componentManager.getPlatformMid(message.getDeviceName(), message.getMid());
+            if (platformMid == null) {
+                platformMid = UniqueIdUtil.newRequestId();
+            }
+            thingModelMessage.setMid(platformMid);
+        } else {
+            //其它消息重新生成唯一MID
+            thingModelMessage.setMid(UniqueIdUtil.newRequestId());
         }
+
+        deviceBehaviourService.reportMessage(thingModelMessage);
     }
 }

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

@@ -0,0 +1,11 @@
+package cc.iotkit.comps.config;
+
+public class CacheKey {
+
+    private static final String KEY_CMD_MID = "str:cmd:mid:%s:%s";
+
+    public static String getKeyCmdMid(String deviceName, String downMid) {
+        return String.format(KEY_CMD_MID, deviceName, downMid);
+    }
+
+}

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

@@ -1,15 +0,0 @@
-package cc.iotkit.comps.model;
-
-import lombok.Data;
-
-@Data
-public class AuthInfo {
-
-    private String productKey;
-
-    private String deviceName;
-
-    private String productSecret;
-
-    private String deviceSecret;
-}

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

@@ -1,17 +0,0 @@
-package cc.iotkit.comps.model;
-
-import lombok.Data;
-
-@Data
-public class DeviceState {
-
-    public static final String STATE_ONLINE = "online";
-    public static final String STATE_OFFLINE = "offline";
-
-    private String productKey;
-
-    private String deviceName;
-
-    private String state;
-
-}

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

@@ -1,49 +0,0 @@
-package cc.iotkit.comps.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/component-server/src/main/java/cc/iotkit/comps/service/DeviceBehaviourService.java

@@ -3,12 +3,15 @@ package cc.iotkit.comps.service;
 import cc.iotkit.common.Constants;
 import cc.iotkit.common.exception.BizException;
 import cc.iotkit.common.utils.JsonUtil;
-import cc.iotkit.comp.model.DeviceMessage;
+import cc.iotkit.common.utils.UniqueIdUtil;
+import cc.iotkit.comp.model.RegisterInfo;
+import cc.iotkit.comp.model.DeviceState;
 import cc.iotkit.comps.config.ServerConfig;
-import cc.iotkit.comps.model.RegisterInfo;
+import cc.iotkit.dao.DeviceCache;
 import cc.iotkit.dao.DeviceRepository;
 import cc.iotkit.dao.ProductRepository;
 import cc.iotkit.model.device.DeviceInfo;
+import cc.iotkit.model.device.message.ThingModelMessage;
 import cc.iotkit.model.product.Product;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.RandomUtils;
@@ -23,7 +26,6 @@ import org.springframework.stereotype.Service;
 import javax.annotation.PostConstruct;
 import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 import java.util.Optional;
 
 @Slf4j
@@ -36,8 +38,12 @@ public class DeviceBehaviourService {
     private DeviceRepository deviceRepository;
     @Autowired
     private ServerConfig serverConfig;
+    @Autowired
+    private DeviceCache deviceCache;
+    @Autowired
+    private DeviceStateHolder deviceStateHolder;
 
-    private Producer<DeviceMessage> deviceMessageProducer;
+    private Producer<ThingModelMessage> deviceMessageProducer;
 
     @PostConstruct
     public void init() throws PulsarClientException {
@@ -45,13 +51,12 @@ public class DeviceBehaviourService {
         PulsarClient client = PulsarClient.builder()
                 .serviceUrl(serverConfig.getPulsarBrokerUrl())
                 .build();
-        deviceMessageProducer = client.newProducer(JSONSchema.of(DeviceMessage.class))
-                .topic("persistent://public/default/device_raw")
+        deviceMessageProducer = client.newProducer(JSONSchema.of(ThingModelMessage.class))
+                .topic("persistent://iotkit/default/device_thing")
                 .create();
 
     }
 
-
     public void register(RegisterInfo info) {
         try {
             DeviceInfo deviceInfo = register(null, info);
@@ -66,7 +71,17 @@ public class DeviceBehaviourService {
                                     subDevice.getTag(), null));
                 }
             }
-            //todo 产生设备注册事件
+
+            //设备注册消息
+            ThingModelMessage modelMessage = new ThingModelMessage(
+                    UniqueIdUtil.newRequestId(), "",
+                    info.getProductKey(), info.getDeviceName(),
+                    ThingModelMessage.TYPE_LIFETIME, "register",
+                    0, new HashMap<>(), System.currentTimeMillis(),
+                    System.currentTimeMillis()
+            );
+
+            reportMessage(modelMessage);
         } catch (BizException e) {
             log.error("register device error", e);
             throw e;
@@ -86,34 +101,20 @@ public class DeviceBehaviourService {
         DeviceInfo device = deviceRepository.findByProductKeyAndDeviceName(pk, info.getDeviceName());
 
         if (device != null) {
-            //更新设备信息
-            device.setParentId(parentId);
-            device.setUid(uid);
-            Map<String, Object> tag = info.getTag();
-            Map<String, Object> oldTag = device.getTag();
-
-            if (oldTag == null) {
-                oldTag = new HashMap<>();
-            }
-
-            if (tag != null) {
-                oldTag.putAll(tag);
-            }
-
-            device.setTag(oldTag);
-        } else {
-            //不存在,注册新设备
-            device = new DeviceInfo();
-            device.setId(newDeviceId(info.getDeviceName()));
-            device.setParentId(parentId);
-            device.setUid(uid);
-            device.setDeviceId(device.getId());
-            device.setProductKey(pk);
-            device.setDeviceName(info.getDeviceName());
-            device.setTag(info.getTag());
-            device.setState(new DeviceInfo.State(false, null, null));
-            device.setCreateAt(System.currentTimeMillis());
+            log.info("device already registered.");
+            return device;
         }
+        //不存在,注册新设备
+        device = new DeviceInfo();
+        device.setId(newDeviceId(info.getDeviceName()));
+        device.setParentId(parentId);
+        device.setUid(uid);
+        device.setDeviceId(device.getId());
+        device.setProductKey(pk);
+        device.setDeviceName(info.getDeviceName());
+        device.setTag(info.getTag());
+        device.setState(new DeviceInfo.State(false, null, null));
+        device.setCreateAt(System.currentTimeMillis());
 
         deviceRepository.save(device);
         log.info("device registered:{}", JsonUtil.toJsonString(device));
@@ -171,19 +172,60 @@ public class DeviceBehaviourService {
         if (device == null) {
             throw new BizException("device does not exist");
         }
+        deviceStateChange(device, online);
 
+        //可能是父设备,父设备离线，子设备也要离线
+        if (!online && device.getParentId() == null) {
+            List<DeviceInfo> subDevices = deviceRepository.findByParentId(device.getDeviceId());
+            for (DeviceInfo subDevice : subDevices) {
+                deviceStateChange(subDevice, false);
+            }
+        }
+    }
+
+    private void deviceStateChange(DeviceInfo device, boolean online) {
         if (online) {
             device.getState().setOnline(true);
             device.getState().setOnlineTime(System.currentTimeMillis());
+            deviceStateHolder.online(device.getDeviceId());
         } else {
             device.getState().setOnline(false);
             device.getState().setOfflineTime(System.currentTimeMillis());
+            deviceStateHolder.offline(device.getDeviceId());
         }
         deviceRepository.save(device);
-        //todo 产生在离线事件
+
+        //设备状态变更消息
+        ThingModelMessage modelMessage = new ThingModelMessage(
+                UniqueIdUtil.newRequestId(), "",
+                device.getProductKey(), device.getDeviceName(),
+                ThingModelMessage.TYPE_STATE,
+                online ? DeviceState.STATE_ONLINE : DeviceState.STATE_OFFLINE,
+                0,
+                new HashMap<>(), System.currentTimeMillis(),
+                System.currentTimeMillis()
+        );
+
+        reportMessage(modelMessage);
     }
 
-    public void reportMessage(DeviceMessage message) throws PulsarClientException {
-        deviceMessageProducer.send(message);
+    public void reportMessage(ThingModelMessage message) {
+        try {
+            DeviceInfo device = deviceCache.findByProductKeyAndDeviceName(message.getProductKey(),
+                    message.getDeviceName());
+            if (device == null) {
+                return;
+            }
+            if (message.getOccurred() == null) {
+                message.setOccurred(System.currentTimeMillis());
+            }
+            if (message.getTime() == null) {
+                message.setTime(System.currentTimeMillis());
+            }
+            message.setDeviceId(device.getDeviceId());
+            deviceMessageProducer.send(message);
+        } catch (PulsarClientException e) {
+            log.error("send thing model message error", e);
+        }
     }
 }

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

@@ -3,9 +3,12 @@ package cc.iotkit.comps.service;
 import cc.iotkit.common.Constants;
 import cc.iotkit.common.utils.JsonUtil;
 import cc.iotkit.comps.config.ServerConfig;
+import cc.iotkit.dao.DeviceDao;
+import cc.iotkit.dao.DevicePropertyRepository;
 import cc.iotkit.dao.ThingModelMessageRepository;
 import cc.iotkit.dao.UserInfoRepository;
 import cc.iotkit.model.UserInfo;
+import cc.iotkit.model.device.message.DeviceProperty;
 import cc.iotkit.model.device.message.ThingModelMessage;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
@@ -14,6 +17,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Collectors;
 
 @Slf4j
@@ -21,20 +25,21 @@ import java.util.stream.Collectors;
 public class DeviceMessageConsumer implements MessageListener<ThingModelMessage> {
 
     private final ServerConfig serverConfig;
-
     private final ThingModelMessageRepository messageRepository;
-
-    private final UserInfoRepository userInfoRepository;
+    private final DevicePropertyRepository propertyRepository;
+    private final DeviceDao deviceDao;
 
     @SneakyThrows
     @Autowired
     public DeviceMessageConsumer(ServerConfig serverConfig,
                                  ThingModelMessageRepository messageRepository,
-                                 UserInfoRepository userInfoRepository) {
+                                 UserInfoRepository userInfoRepository,
+                                 DevicePropertyRepository propertyRepository,
+                                 DeviceDao deviceDao) {
         this.serverConfig = serverConfig;
         this.messageRepository = messageRepository;
-        this.userInfoRepository = userInfoRepository;
-
+        this.propertyRepository = propertyRepository;
+        this.deviceDao = deviceDao;
         PulsarClient client = PulsarClient.builder()
                 .serviceUrl(this.serverConfig.getPulsarBrokerUrl())
                 .build();
@@ -56,14 +61,34 @@ public class DeviceMessageConsumer implements MessageListener<ThingModelMessage>
     @Override
     public void received(Consumer<ThingModelMessage> consumer, Message<ThingModelMessage> msg) {
         ThingModelMessage modelMessage = msg.getValue();
-        log.info("receive message:{}", JsonUtil.toJsonString(modelMessage));
+        String deviceId = modelMessage.getDeviceId();
+        log.info("save message to es:{}", JsonUtil.toJsonString(modelMessage));
+        //属性入库
+        if (ThingModelMessage.TYPE_PROPERTY.equals(modelMessage.getType())
+                && "report".equals(modelMessage.getIdentifier())) {
+            log.info("update device property,deviceId:{},property:{}",
+                    deviceId, JsonUtil.toJsonString(modelMessage.getData()));
+            deviceDao.updateProperties(deviceId, (Map<String, Object>) modelMessage.getData());
+
+            //设备属性历史数据存储
+            if (modelMessage.getData() instanceof Map) {
+                Map map = (Map) modelMessage.getData();
+                for (Object key : map.keySet()) {
+                    propertyRepository.save(
+                            new DeviceProperty(
+                                    modelMessage.getMid(),
+                                    deviceId,
+                                    key.toString(),
+                                    map.get(key),
+                                    modelMessage.getOccurred()
+                            )
+                    );
+                }
+            }
+        }
+
         //设备消息日志入库
         messageRepository.save(modelMessage);
-
-        messageRepository.findAll().forEach(m -> {
-            log.info(JsonUtil.toJsonString(m));
-        });
-
         consumer.acknowledge(msg);
     }
 

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

@@ -0,0 +1,122 @@
+package cc.iotkit.comps.service;
+
+import cc.iotkit.common.utils.ThreadUtil;
+import cc.iotkit.comps.config.ServerConfig;
+import cc.iotkit.dao.DeviceRepository;
+import cc.iotkit.model.device.DeviceInfo;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.*;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.PostConstruct;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 设备状态维持，每1分钟更新一次心跳
+ */
+@Slf4j
+@Service
+public class DeviceStateHolder implements MessageListener<DeviceStateHolder.OfflineMessage> {
+
+    private ScheduledThreadPoolExecutor stateHolderTask;
+
+    private Set<String> devices = new TreeSet<>();
+
+    @Autowired
+    private StringRedisTemplate redisTemplate;
+    @Autowired
+    private ServerConfig serverConfig;
+    @Autowired
+    private DeviceRepository deviceRepository;
+
+    private Producer<OfflineMessage> offlineMessageProducer;
+
+    @PostConstruct
+    public void init() throws PulsarClientException {
+        stateHolderTask = ThreadUtil.newScheduled(4, "thread-device-state-holder");
+        stateHolderTask.scheduleAtFixedRate(this::hold, 0, 1, TimeUnit.MINUTES);
+
+        PulsarClient client = PulsarClient.builder()
+                .serviceUrl(this.serverConfig.getPulsarBrokerUrl())
+                .build();
+
+        offlineMessageProducer = client.newProducer(Schema.JSON(OfflineMessage.class))
+                .topic("persistent://iotkit/default/holder_offline")
+                .create();
+
+        client.newConsumer(Schema.JSON(OfflineMessage.class))
+                .topic("persistent://iotkit/default/holder_offline")
+                .subscriptionName("holder_offline")
+                .consumerName("device-state-holder-consumer")
+                .messageListener(this).subscribe();
+    }
+
+    public void online(String deviceId) {
+        try {
+            devices.add(deviceId);
+            hold(deviceId);
+            //上线后先产生离线消息
+            offlineMessageProducer.send(new OfflineMessage(deviceId));
+        } catch (Throwable e) {
+            log.error("state holder online error", e);
+        }
+    }
+
+    public void offline(String deviceId) {
+        devices.remove(deviceId);
+    }
+
+    private void hold() {
+        //标识在线
+        for (String deviceId : devices) {
+            hold(deviceId);
+        }
+    }
+
+    private void hold(String deviceId) {
+        redisTemplate.opsForValue().set("str:device:state:holder:" + deviceId,
+                "1", 5, TimeUnit.SECONDS);
+    }
+
+    @SneakyThrows
+    @Override
+    public void received(Consumer<OfflineMessage> consumer, Message<OfflineMessage> msg) {
+        String deviceId = msg.getValue().getDeviceId();
+        //如果设备在线，不处理离线消息
+        String hold = redisTemplate.opsForValue().get("str:device:state:holder:" + deviceId);
+        if (hold != null) {
+            return;
+        }
+        //如果设备不在线，则将设备更新为离线
+        DeviceInfo device = deviceRepository.findByDeviceId(deviceId);
+        DeviceInfo.State state = device.getState();
+        state.setOnline(false);
+        state.setOfflineTime(System.currentTimeMillis());
+        deviceRepository.save(device);
+        log.info("device offline,deviceId:{}", deviceId);
+
+        consumer.acknowledge(msg);
+    }
+
+    @Override
+    public void reachedEndOfTopic(Consumer<OfflineMessage> consumer) {
+
+    }
+
+
+    @Data
+    @NoArgsConstructor
+    @AllArgsConstructor
+    public static class OfflineMessage {
+        private String deviceId;
+    }
+}

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

@@ -10,4 +10,15 @@ public abstract class AbstractComponent implements IComponent {
 
     protected IConverter converter;
 
+    protected CompConfig config;
+
+    @Override
+    public void create(CompConfig config) {
+        this.config=config;
+    }
+
+    @Override
+    public CompConfig getConfig() {
+        return config;
+    }
 }

protocol-gateway/gateway-client/src/main/java/cc/iotkit/protocol/Result.java → protocol-gateway/component/src/main/java/cc/iotkit/comp/CompConfig.java

@@ -1,4 +1,4 @@
-package cc.iotkit.protocol;
+package cc.iotkit.comp;
 
 import lombok.AllArgsConstructor;
 import lombok.Data;
@@ -7,10 +7,10 @@ import lombok.NoArgsConstructor;
 @Data
 @NoArgsConstructor
 @AllArgsConstructor
-public class Result {
+public class CompConfig {
 
-    private boolean success;
+    private long cmdTimeout;
 
-    private String content;
+    private String other;
 
 }

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

@@ -1,10 +1,13 @@
 package cc.iotkit.comp;
 
+import cc.iotkit.comp.model.DeviceState;
+import cc.iotkit.comp.model.RegisterInfo;
+import cc.iotkit.converter.DeviceMessage;
 import cc.iotkit.converter.IConverter;
 
 public interface IComponent {
 
-    void create(String config);
+    void create(CompConfig config);
 
     void start();
 
@@ -12,10 +15,17 @@ public interface IComponent {
 
     void destroy();
 
+    void onDeviceStateChange(DeviceState state);
+
+    void send(DeviceMessage message);
+
+    boolean exist(String productKey, String deviceName);
+
     void setHandler(IMessageHandler handler);
 
     void setConverter(IConverter converter);
 
     IConverter getConverter();
 
+    CompConfig getConfig();
 }

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

@@ -1,14 +1,10 @@
 package cc.iotkit.comp;
 
+import cc.iotkit.comp.model.ReceiveResult;
+
 import java.util.Map;
 
 public interface IMessageHandler {
 
-    void register(Map<String, Object> head, String msg);
-
-    void auth(Map<String, Object> head, String msg);
-
-    void state(Map<String, Object> head, String msg);
-
-    void onReceive(Map<String, Object> head, String type, String msg);
+    ReceiveResult onReceive(Map<String, Object> head, String type, String msg);
 }

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

@@ -9,4 +9,8 @@ public class AuthInfo {
 
     private String deviceName;
 
+    private String productSecret;
+
+    private String deviceSecret;
+
 }

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

@@ -0,0 +1,33 @@
+package cc.iotkit.comp.model;
+
+import cc.iotkit.common.utils.JsonUtil;
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.Map;
+
+@Data
+@Slf4j
+public class DeviceState {
+
+    public static final String STATE_ONLINE = "online";
+    public static final String STATE_OFFLINE = "offline";
+
+    private String productKey;
+
+    private String deviceName;
+
+    private String state;
+
+    private Parent parent;
+
+    public static DeviceState from(Map map) {
+        return JsonUtil.parse(JsonUtil.toJsonString(map), DeviceState.class);
+    }
+
+    @Data
+    public static class Parent {
+        private String productKey;
+        private String deviceName;
+    }
+}

protocol-gateway/decode-function/src/main/java/cc/iotkit/protocol/function/DeviceMessage.java → protocol-gateway/component/src/main/java/cc/iotkit/comp/model/ReceiveResult.java

@@ -1,25 +1,18 @@
-package cc.iotkit.protocol.function;
+package cc.iotkit.comp.model;
 
 import lombok.AllArgsConstructor;
-import lombok.Builder;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
-/**
- * 设备消息
- */
 @Data
 @NoArgsConstructor
 @AllArgsConstructor
-@Builder
-public class DeviceMessage {
+public class ReceiveResult {
 
     private String productKey;
 
     private String deviceName;
 
-    private String mid;
-
-    private String content;
+    private Object data;
 
 }

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

@@ -4,13 +4,17 @@ import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
 import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.beanutils.BeanUtils;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 
 /**
  * 注册信息
  */
+@Slf4j
 @Data
 @NoArgsConstructor
 @AllArgsConstructor
@@ -33,6 +37,38 @@ public class RegisterInfo {
         this.model = model;
     }
 
+    public RegisterInfo(String productKey, String deviceName, String subProductKey, String subDeviceName) {
+        this.productKey = productKey;
+        this.deviceName = deviceName;
+        if (subProductKey != null && subDeviceName != null) {
+            SubDevice subDevice = new SubDevice(subProductKey, subDeviceName, null, null);
+            subDevices = new ArrayList<>();
+            subDevices.add(subDevice);
+        }
+    }
+
+    public static RegisterInfo from(Map map) {
+        RegisterInfo bean = new RegisterInfo();
+        try {
+            BeanUtils.populate(bean, map);
+            List<SubDevice> subDevices = new ArrayList<>();
+            List<Object> sourceSubDevices = (List<Object>) map.get("subDevices");
+            if (sourceSubDevices == null) {
+                return bean;
+            }
+            for (Object sourceSubDevice : sourceSubDevices) {
+                SubDevice subDevice = new SubDevice();
+                BeanUtils.populate(subDevice, (Map<String, ? extends Object>) sourceSubDevice);
+                subDevices.add(subDevice);
+            }
+            bean.setSubDevices(subDevices);
+        } catch (Throwable e) {
+            log.error("parse bean from map error", e);
+            return null;
+        }
+        return bean;
+    }
+
     @Data
     @NoArgsConstructor
     @AllArgsConstructor
@@ -46,4 +82,4 @@ public class RegisterInfo {
 
         private Map<String, Object> tag;
     }
-}
+}

protocol-gateway/component/src/main/java/cc/iotkit/comp/model/DeviceMessage.java → protocol-gateway/converter/src/main/java/cc/iotkit/converter/DeviceMessage.java

@@ -1,4 +1,4 @@
-package cc.iotkit.comp.model;
+package cc.iotkit.converter;
 
 import lombok.Data;
 
@@ -11,5 +11,5 @@ public class DeviceMessage {
 
     private String mid;
 
-    private String content;
+    private Object content;
 }

protocol-gateway/converter/src/main/java/cc/iotkit/converter/DeviceService.java

@@ -1,20 +0,0 @@
-package cc.iotkit.converter;
-
-import lombok.Data;
-
-@Data
-public class DeviceService<T> {
-
-    private String mid;
-
-    private String productKey;
-
-    private String deviceName;
-
-    private String type;
-
-    private String identifier;
-
-    private T params;
-
-}

protocol-gateway/converter/src/main/java/cc/iotkit/converter/IConverter.java

@@ -6,8 +6,8 @@ public interface IConverter {
 
     void setScript(String script);
 
-    ThingModelMessage decode(String msg);
+    ThingModelMessage decode(DeviceMessage msg);
 
-    String encode(DeviceService<?> service, Device device);
+    DeviceMessage encode(ThingService<?> service, Device device);
 
 }

protocol-gateway/converter/src/main/java/cc/iotkit/converter/ScriptConverter.java

@@ -1,5 +1,6 @@
 package cc.iotkit.converter;
 
+import cc.iotkit.common.utils.JsonUtil;
 import cc.iotkit.model.device.message.ThingModelMessage;
 import jdk.nashorn.api.scripting.NashornScriptEngine;
 import jdk.nashorn.api.scripting.ScriptObjectMirror;
@@ -9,26 +10,26 @@ import org.apache.commons.beanutils.BeanUtils;
 
 import javax.script.ScriptEngineManager;
 import javax.script.ScriptException;
+import java.util.Map;
 
 @Slf4j
 @Data
 public class ScriptConverter implements IConverter {
     private final NashornScriptEngine engine = (NashornScriptEngine) (new ScriptEngineManager()).getEngineByName("nashorn");
 
-    private String script;
+    private Object scriptObj;
 
     public void setScript(String script) {
-        this.script = script;
         try {
-            engine.eval(script);
+            scriptObj = engine.eval(script);
         } catch (ScriptException e) {
             log.error("eval converter script error", e);
         }
     }
 
-    public ThingModelMessage decode(String msg) {
+    public ThingModelMessage decode(DeviceMessage msg) {
         try {
-            ScriptObjectMirror result = (ScriptObjectMirror) engine.invokeFunction("decode", msg);
+            ScriptObjectMirror result = (ScriptObjectMirror) engine.invokeMethod(scriptObj, "decode", msg);
             ThingModelMessage modelMessage = new ThingModelMessage();
             BeanUtils.populate(modelMessage, result);
             return modelMessage;
@@ -39,10 +40,13 @@ public class ScriptConverter implements IConverter {
     }
 
     @Override
-    public String encode(DeviceService<?> service, Device device) {
+    public DeviceMessage encode(ThingService<?> service, Device device) {
         try {
-            ScriptObjectMirror result = (ScriptObjectMirror) engine.invokeFunction("encode", service, device);
-            return result.toString();
+            ScriptObjectMirror result = (ScriptObjectMirror) engine.invokeMethod(scriptObj, "encode", service, device);
+            Map map = (Map) JsonUtil.toObject(result);
+            DeviceMessage message = new DeviceMessage();
+            BeanUtils.populate(message, map);
+            return message;
         } catch (Throwable e) {
             log.error("execute encode script error", e);
         }

protocol-gateway/gateway-client/src/main/java/cc/iotkit/protocol/DeviceMessage.java → protocol-gateway/converter/src/main/java/cc/iotkit/converter/ThingService.java

@@ -1,25 +1,26 @@
-package cc.iotkit.protocol;
+package cc.iotkit.converter;
 
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
-/**
- * 设备消息
- */
 @Data
 @NoArgsConstructor
 @AllArgsConstructor
 @Builder
-public class DeviceMessage {
+public class ThingService<T> {
+
+    private String mid;
 
     private String productKey;
 
     private String deviceName;
 
-    private String mid;
+    private String type;
+
+    private String identifier;
 
-    private String content;
+    private T params;
 
 }

protocol-gateway/decode-function/pom.xml

@@ -1,70 +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>decode-function</artifactId>
-
-    <dependencies>
-        <dependency>
-            <groupId>org.apache.pulsar</groupId>
-            <artifactId>pulsar-functions-api</artifactId>
-        </dependency>
-
-        <dependency>
-            <groupId>org.projectlombok</groupId>
-            <artifactId>lombok</artifactId>
-        </dependency>
-
-        <dependency>
-            <groupId>cc.iotkit</groupId>
-            <artifactId>common</artifactId>
-        </dependency>
-
-    </dependencies>
-
-    <build>
-        <plugins>
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-assembly-plugin</artifactId>
-                <version>3.1.0</version>
-                <configuration>
-                    <archive>
-                        <manifest>
-                            <mainClass>cc.iotkit.fun.TestFunction</mainClass>
-                        </manifest>
-                    </archive>
-                    <descriptorRefs>
-                        <descriptorRef>jar-with-dependencies</descriptorRef>
-                    </descriptorRefs>
-                </configuration>
-                <executions>
-                    <execution>
-                        <id>make-assembly</id>
-                        <phase>package</phase>
-                        <goals>
-                            <goal>single</goal>
-                        </goals>
-                    </execution>
-                </executions>
-            </plugin>
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-compiler-plugin</artifactId>
-                <configuration>
-                    <source>8</source>
-                    <target>8</target>
-                </configuration>
-            </plugin>
-        </plugins>
-    </build>
-
-</project>

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

@@ -1,51 +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.*;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Optional;
-import java.util.concurrent.ConcurrentHashMap;
-
-/**
- * 上行消息转换函数
- */
-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<>();
-
-    @Override
-    public ThingModelMessage process(DeviceMessage msg, Context context) throws Exception {
-        Optional<Object> optGateway = context.getUserConfigValue("gateway");
-        Optional<Object> optScript = context.getUserConfigValue("script");
-        if (!optGateway.isPresent() || !optScript.isPresent()) {
-            return null;
-        }
-
-        String gateway = optGateway.get().toString();
-        compiledScripts.putIfAbsent(gateway, engine.compile(optScript.get() + ".decode(msg)"));
-
-        CompiledScript script = compiledScripts.get(gateway);
-        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;
-        }
-        if (result instanceof Map) {
-            return ThingModelMessage.from((Map<?, ?>) result);
-        }
-        return null;
-    }
-
-}

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

@@ -1,62 +0,0 @@
-package cc.iotkit.protocol.function;
-
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-
-import java.util.HashMap;
-import java.util.Map;
-
-/**
- * 物模型消息
- */
-@Data
-@NoArgsConstructor
-@AllArgsConstructor
-@Builder
-public class ThingModelMessage {
-
-    private String productKey;
-
-    private String deviceName;
-
-    private String mid;
-
-    private String identifier;
-
-    private Map<String, Object> data;
-
-    /**
-     * 时间戳，设备上的事件或数据产生的本地时间
-     */
-    private Long occur;
-
-    /**
-     * 消息上报时间
-     */
-    private Long time;
-
-    public static ThingModelMessage from(Map<?,?> map) {
-        ThingModelMessage message = new ThingModelMessage();
-        message.setProductKey(getStr(map, "productKey"));
-        message.setDeviceName(getStr(map, "deviceName"));
-        message.setMid(getStr(map, "mid"));
-        message.setIdentifier(getStr(map, "identifier"));
-        Object data = map.get("data");
-        if (data instanceof Map) {
-            message.setData((Map<String, Object>) data);
-        } else {
-            message.setData(new HashMap<>());
-        }
-        return message;
-    }
-
-    private static String getStr(Map<?,?> map, String key) {
-        Object val = map.get(key);
-        if (val == null) {
-            return null;
-        }
-        return val.toString();
-    }
-}

protocol-gateway/gateway-client/pom.xml → protocol-gateway/emqx-component/pom.xml

@@ -9,33 +9,28 @@
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
-    <artifactId>gateway-client</artifactId>
-
-    <properties>
-        <maven.compiler.source>8</maven.compiler.source>
-        <maven.compiler.target>8</maven.compiler.target>
-    </properties>
+    <artifactId>emqx-component</artifactId>
 
     <dependencies>
 
         <dependency>
-            <groupId>org.springframework</groupId>
-            <artifactId>spring-web</artifactId>
+            <groupId>io.vertx</groupId>
+            <artifactId>vertx-core</artifactId>
         </dependency>
 
         <dependency>
-            <groupId>com.squareup.okhttp3</groupId>
-            <artifactId>okhttp</artifactId>
+            <groupId>io.vertx</groupId>
+            <artifactId>vertx-web-proxy</artifactId>
         </dependency>
 
         <dependency>
-            <groupId>org.apache.pulsar</groupId>
-            <artifactId>pulsar-client</artifactId>
+            <groupId>io.vertx</groupId>
+            <artifactId>vertx-mqtt</artifactId>
         </dependency>
 
         <dependency>
-            <groupId>org.projectlombok</groupId>
-            <artifactId>lombok</artifactId>
+            <groupId>cc.iotkit</groupId>
+            <artifactId>model</artifactId>
         </dependency>
 
         <dependency>
@@ -43,6 +38,11 @@
             <artifactId>common</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>cc.iotkit</groupId>
+            <artifactId>component</artifactId>
+        </dependency>
+
     </dependencies>
 
 </project>

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

@@ -0,0 +1,70 @@
+package cc.iotkit.comp.emqx;
+
+import cc.iotkit.comp.IMessageHandler;
+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 lombok.extern.slf4j.Slf4j;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Slf4j
+public class AuthVerticle extends AbstractVerticle {
+
+    private HttpServer backendServer;
+
+    private IMessageHandler executor;
+
+    private EmqxConfig config;
+
+    public void setExecutor(IMessageHandler executor) {
+        this.executor = executor;
+    }
+
+    public AuthVerticle(EmqxConfig config) {
+        this.config = config;
+    }
+
+    @Override
+    public void start() throws Exception {
+        backendServer = vertx.createHttpServer();
+        Router backendRouter = Router.router(vertx);
+
+        backendRouter.route(HttpMethod.POST, "/mqtt/auth").handler(rc -> {
+            JsonObject json = rc.getBodyAsJson();
+            try {
+                executor.onReceive(new HashMap<>(), "auth", json.toString());
+                rc.response().setStatusCode(200)
+                        .end();
+            } catch (Throwable e) {
+                rc.response().setStatusCode(500)
+                        .end();
+                log.error("mqtt auth failed", e);
+            }
+        });
+        backendRouter.route(HttpMethod.POST, "/mqtt/acl").handler(rc -> {
+            JsonObject json = rc.getBodyAsJson();
+            try {
+                Map<String, Object> head = new HashMap<>();
+                head.put("topic", json.getString("topic"));
+                executor.onReceive(head, "subscribe", json.toString());
+                rc.response().setStatusCode(200)
+                        .end();
+            } catch (Throwable e) {
+                rc.response().setStatusCode(500)
+                        .end();
+                log.error("mqtt acl failed", e);
+            }
+        });
+
+        backendServer.requestHandler(backendRouter).listen(config.getAuthPort());
+    }
+
+    @Override
+    public void stop() throws Exception {
+        backendServer.close(voidAsyncResult -> log.info("close emqx auth server..."));
+    }
+}

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

@@ -0,0 +1,147 @@
+package cc.iotkit.comp.emqx;
+
+import cc.iotkit.common.exception.BizException;
+import cc.iotkit.common.utils.JsonUtil;
+import cc.iotkit.comp.AbstractComponent;
+import cc.iotkit.comp.CompConfig;
+import cc.iotkit.comp.model.DeviceState;
+import cc.iotkit.converter.DeviceMessage;
+import io.vertx.core.Future;
+import io.vertx.core.Vertx;
+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 java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+
+@Slf4j
+public class EmqxComponent extends AbstractComponent {
+
+    private Vertx vertx;
+    private AuthVerticle authVerticle;
+    private CountDownLatch countDownLatch;
+    private String deployedId;
+    private EmqxConfig mqttConfig;
+
+    public void create(CompConfig config) {
+        super.create(config);
+        vertx = Vertx.vertx();
+        mqttConfig = JsonUtil.parse(config.getOther(), EmqxConfig.class);
+        authVerticle = new AuthVerticle(mqttConfig);
+    }
+
+    @Override
+    public void start() {
+        try {
+            authVerticle.setExecutor(getHandler());
+            countDownLatch = new CountDownLatch(1);
+            Future<String> future = vertx.deployVerticle(authVerticle);
+            future.onSuccess((s -> {
+                deployedId = s;
+                countDownLatch.countDown();
+            }));
+            future.onFailure((e) -> {
+                countDownLatch.countDown();
+                log.error("start emqx auth component failed", e);
+            });
+            countDownLatch.await();
+
+            MqttClientOptions options = new MqttClientOptions()
+                    .setClientId(mqttConfig.getClientId())
+                    .setUsername(mqttConfig.getUsername())
+                    .setPassword(mqttConfig.getPassword())
+                    .setCleanSession(true)
+                    .setKeepAliveInterval(60);
+
+            if (mqttConfig.isSsl()) {
+                options.setSsl(true)
+                        .setTrustAll(true);
+            }
+            MqttClient client = MqttClient.create(vertx, options);
+
+            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(topic, 1);
+            }
+
+            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")) {
+
+
+                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));
+
+        } catch (Throwable e) {
+            throw new BizException("start emqx auth component error", e);
+        }
+    }
+
+    @SneakyThrows
+    @Override
+    public void stop() {
+        authVerticle.stop();
+        Future<Void> future = vertx.undeploy(deployedId);
+        future.onSuccess(unused -> log.info("stop emqx auth component success"));
+    }
+
+    @Override
+    public void destroy() {
+
+    }
+
+    @Override
+    public void onDeviceStateChange(DeviceState state) {
+
+    }
+
+    @Override
+    public void send(DeviceMessage message) {
+
+    }
+
+    @Override
+    public boolean exist(String productKey, String deviceName) {
+        return false;
+    }
+
+}

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

@@ -0,0 +1,25 @@
+package cc.iotkit.comp.emqx;
+
+import lombok.Data;
+
+import java.util.List;
+
+@Data
+public class EmqxConfig {
+
+    private int authPort;
+
+    private String broker;
+
+    private int port;
+
+    private boolean ssl;
+
+    private String clientId;
+
+    private String username;
+
+    private String password;
+
+    private List<String> subscribeTopics;
+}

protocol-gateway/gateway-client/src/main/java/cc/iotkit/protocol/DeregisterInfo.java

@@ -1,17 +0,0 @@
-package cc.iotkit.protocol;
-
-import lombok.Data;
-
-
-/**
- * 注销信息
- */
-@Data
-public class DeregisterInfo {
-
-    private String productKey;
-
-    private String deviceName;
-
-    private boolean cascade;
-}

protocol-gateway/gateway-client/src/main/java/cc/iotkit/protocol/DeviceBehaviour.java

@@ -1,34 +0,0 @@
-package cc.iotkit.protocol;
-
-
-/**
- * 设备行为接口
- */
-public interface DeviceBehaviour {
-
-    /**
-     * 设备注册
-     */
-    Result register(RegisterInfo info);
-
-    /**
-     * 设备上线
-     */
-    Result online(String productKey, String deviceName);
-
-    /**
-     * 设备离线
-     */
-    Result offline(String productKey, String deviceName);
-
-    /**
-     * 设备消息上报
-     */
-    void messageReport(DeviceMessage msg);
-
-    /**
-     * OTA消息上报
-     */
-    void otaProgressReport(OtaMessage msg);
-
-}

protocol-gateway/gateway-client/src/main/java/cc/iotkit/protocol/DeviceGateway.java

@@ -1,23 +0,0 @@
-package cc.iotkit.protocol;
-
-
-import org.springframework.web.bind.annotation.RequestMapping;
-
-/**
- * 设备网关接口
- */
-public interface DeviceGateway {
-
-    /**
-     * 指令下发
-     */
-    @RequestMapping("/sendMessage")
-    Result sendMessage(DeviceMessage msg);
-
-    /**
-     * OTA升级任务下发
-     */
-    @RequestMapping("/sendOta")
-    Result sendOta(OtaInfo ota);
-
-}

protocol-gateway/gateway-client/src/main/java/cc/iotkit/protocol/OtaInfo.java

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

protocol-gateway/gateway-client/src/main/java/cc/iotkit/protocol/OtaMessage.java

@@ -1,30 +0,0 @@
-package cc.iotkit.protocol;
-
-import lombok.Data;
-
-/**
- * OTA消息
- */
-@Data
-public class OtaMessage {
-
-    private final String TYPE_PROGRESS="progress";
-    private final String TYPE_RESULT="result";
-
-    private final String PROGRESS_DOWNLOADING="downloading";
-    private final String PROGRESS_DOWNLOADED="downloaded";
-    private final String PROGRESS_UPGRADING="upgrading";
-    private final String PROGRESS_UPGRADED="upgraded";
-
-    private String productKey;
-
-    private String deviceName;
-
-    private String type;
-
-    private String jobId;
-
-    private String progress;
-
-    private String result;
-}

protocol-gateway/gateway-client/src/main/java/cc/iotkit/protocol/RegisterInfo.java

@@ -1,49 +0,0 @@
-package cc.iotkit.protocol;
-
-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/gateway-client/src/main/java/cc/iotkit/protocol/client/DeviceBehaviourClient.java

@@ -1,151 +0,0 @@
-package cc.iotkit.protocol.client;
-
-import cc.iotkit.common.exception.BizException;
-import cc.iotkit.common.utils.JsonUtil;
-import cc.iotkit.protocol.*;
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-import lombok.SneakyThrows;
-import okhttp3.*;
-import org.apache.pulsar.client.api.Producer;
-import org.apache.pulsar.client.api.PulsarClient;
-import org.apache.pulsar.client.impl.schema.JSONSchema;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-
-public class DeviceBehaviourClient implements DeviceBehaviour {
-
-    public final String gatewayServer;
-
-    private GatewayConfig gatewayConfig;
-
-    private final OkHttpClient httpClient;
-
-    private Producer<DeviceMessage> deviceMessageProducer;
-
-    private Producer<OtaMessage> otaMessageProducer;
-
-
-    @SneakyThrows
-    public DeviceBehaviourClient(String server) {
-        //初始化协议网关http客户端
-        this.gatewayServer = server.replaceAll("/$", "");
-        httpClient = new OkHttpClient.Builder()
-                .connectTimeout(10, TimeUnit.SECONDS)
-                .readTimeout(30, TimeUnit.SECONDS)
-                .build();
-
-        //获取协议网关配置
-        getConfig();
-
-        //初始化pulsar客户端
-        PulsarClient client = PulsarClient.builder()
-                .serviceUrl(gatewayConfig.getMqServiceUrl())
-                .build();
-
-        deviceMessageProducer = client.newProducer(JSONSchema.of(DeviceMessage.class))
-                .topic("persistent://public/default/device_raw")
-                .create();
-
-        otaMessageProducer = client.newProducer(JSONSchema.of(OtaMessage.class))
-                .topic("persistent://public/default/device_ota")
-                .create();
-    }
-
-    public Request buildRequest(String url, String method, String type, Map<String, Object> data) {
-        Request.Builder requestBuilder = new Request.Builder()
-                .url(url);
-
-        RequestBody requestBody;
-        if ("json".equals(type)) {
-            requestBody = RequestBody.create(MediaType.get("application/json; charset=utf-8"),
-                    JsonUtil.toJsonString(data));
-        } else {
-            FormBody.Builder builder = new FormBody.Builder();
-            data.forEach((key, val) -> builder.add(key, val.toString()));
-            requestBody = builder.build();
-        }
-        requestBuilder.method(method.toUpperCase(), requestBody);
-        return requestBuilder.build();
-    }
-
-
-    private <T> Result invoke(String path, T data) {
-        return invoke(path, "form", data);
-    }
-
-    private <T> Result invokeJson(String path, T data) {
-        return invoke(path, "json", data);
-    }
-
-    private <T> Result invoke(String path, String type, T data) {
-        Request request = buildRequest(gatewayServer + path, "post", type,
-                (data instanceof Map) ? (Map<String, Object>) data :
-                        JsonUtil.parse(JsonUtil.toJsonString(data), Map.class));
-
-        Call call = httpClient.newCall(request);
-        try {
-            Response response = call.execute();
-            if (!response.isSuccessful() || response.body() == null) {
-                return new Result(false, "Remote interface call failed:" + response.body().string());
-            }
-            String content = response.body().string();
-            return JsonUtil.parse(content, Result.class);
-        } catch (Throwable e) {
-            return new Result(false, "Interface call failed:" + e.getMessage());
-        }
-    }
-
-    private void getConfig() {
-        if (this.gatewayConfig == null) {
-            Result result = invoke("/getConfig", new HashMap<>());
-            if (!result.isSuccess()) {
-                throw new BizException("get gateway config failed");
-            }
-            this.gatewayConfig = JsonUtil.parse(result.getContent(), GatewayConfig.class);
-        }
-    }
-
-    @Override
-    public Result register(RegisterInfo info) {
-        return invokeJson("/register", info);
-    }
-
-    @Override
-    public Result online(String productKey, String deviceName) {
-        Map<String, Object> data = new HashMap<>();
-        data.put("productKey", productKey);
-        data.put("deviceName", deviceName);
-        return invoke("/online", data);
-    }
-
-    @Override
-    public Result offline(String productKey, String deviceName) {
-        Map<String, Object> data = new HashMap<>();
-        data.put("productKey", productKey);
-        data.put("deviceName", deviceName);
-        return invoke("/offline", data);
-    }
-
-    @SneakyThrows
-    @Override
-    public void messageReport(DeviceMessage msg) {
-        deviceMessageProducer.send(msg);
-    }
-
-    @Override
-    public void otaProgressReport(OtaMessage msg) {
-    }
-
-    @Data
-    @NoArgsConstructor
-    @AllArgsConstructor
-    public static class GatewayConfig {
-
-        private String mqServiceUrl;
-
-    }
-}

device-server/mqtt-client-simulator/pom.xml → protocol-gateway/mqtt-client-simulator/pom.xml

@@ -3,7 +3,7 @@
          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>device-server</artifactId>
+        <artifactId>protocol-gateway</artifactId>
         <groupId>cc.iotkit</groupId>
         <version>0.0.1-SNAPSHOT</version>
     </parent>