Merge remote-tracking branch 'origin/master'

jetlinks-components/common-component/src/main/java/org/jetlinks/community/configuration/CommonConfiguration.java

@@ -0,0 +1,31 @@
+package org.jetlinks.community.configuration;
+
+import com.alibaba.fastjson.JSON;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import org.apache.commons.beanutils.BeanUtilsBean;
+import org.apache.commons.beanutils.Converter;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class CommonConfiguration {
+
+    static {
+        BeanUtilsBean.getInstance().getConvertUtils().register(new Converter() {
+            @Override
+            public <T> T convert(Class<T> aClass, Object o) {
+                if (o instanceof String) {
+                    o = ((String) o).getBytes();
+                }
+                if (o instanceof byte[]) {
+                    o = Unpooled.wrappedBuffer(((byte[]) o));
+                }
+                if (o instanceof ByteBuf) {
+                    return (T) o;
+                }
+                return convert(aClass, JSON.toJSONBytes(o));
+            }
+        }, ByteBuf.class);
+    }
+
+}

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/service/DefaultElasticSearchService.java

@@ -4,6 +4,7 @@ import com.alibaba.fastjson.JSON;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections4.CollectionUtils;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.bulk.BulkRequest;
 import org.elasticsearch.action.bulk.BulkResponse;
@@ -189,6 +190,7 @@ public class DefaultElasticSearchService implements ElasticSearchService {
                         }));
             })
             .collectList()
+            .filter(CollectionUtils::isNotEmpty)
             .flatMap(lst -> {
                 BulkRequest request = new BulkRequest();
                 lst.forEach(request::add);

jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/MessageGateway.java

@@ -74,6 +74,9 @@ public interface MessageGateway {
         return subscribe(Stream.of(topics).map(Subscription::new).collect(Collectors.toList()), false);
     }
 
+    Flux<TopicMessage> subscribe(Collection<Subscription> subscription, String id, boolean shareCluster);
+
+
     /**
      * 注册一个消息连接器,用于进行真实的消息收发
      *

jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/Subscription.java

@@ -2,6 +2,10 @@ package org.jetlinks.community.gateway;
 
 import lombok.*;
 
+import java.util.Collection;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
 /**
  * 订阅信息.支持通配符**(匹配多层目录)和*(匹配单层目录).
  *
@@ -23,4 +27,10 @@ public class Subscription {
         this.topic = topic;
     }
 
+    public static Collection<Subscription> asList(String... sub) {
+        return Stream.of(sub)
+            .map(Subscription::new)
+            .collect(Collectors.toList());
+    }
+
 }

jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/annotation/Subscribe.java

@@ -23,6 +23,8 @@ public @interface Subscribe {
     @AliasFor("topics")
     String[] value() default {};
 
+    String id() default "";
+
     boolean shareCluster() default false;
 
 }

jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/spring/SpringMessageConnector.java

@@ -12,6 +12,7 @@ import org.springframework.stereotype.Component;
 import org.springframework.util.ClassUtils;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.ReflectionUtils;
+import org.springframework.util.StringUtils;
 import reactor.core.publisher.EmitterProcessor;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.FluxSink;
@@ -55,9 +56,12 @@ public class SpringMessageConnector implements MessageConnector, BeanPostProcess
             if (CollectionUtils.isEmpty(subscribes)) {
                 return;
             }
+            String id = subscribes.getString("id");
+            if (!StringUtils.hasText(id)) {
+                id = type.getSimpleName().concat(".").concat(method.getName());
+            }
             SpringMessageConnection connection = new SpringMessageConnection(
-                type.getSimpleName().concat(".").concat(method.getName())
-                , Stream.of(subscribes.getStringArray("value")).map(Subscription::new).collect(Collectors.toList())
+                id, Stream.of(subscribes.getStringArray("value")).map(Subscription::new).collect(Collectors.toList())
                 , new ProxyMessageListener(bean, method),
                 subscribes.getBoolean("shareCluster")
             );

jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/supports/DefaultMessageGateway.java

@@ -54,8 +54,13 @@ public class DefaultMessageGateway implements MessageGateway {
 
     @Override
     public Flux<TopicMessage> subscribe(Collection<Subscription> subscriptions, boolean shareCluster) {
+        return subscribe(subscriptions, "local:".concat(IDGenerator.SNOW_FLAKE_STRING.generate()), shareCluster);
+    }
+
+    @Override
+    public Flux<TopicMessage> subscribe(Collection<Subscription> subscriptions, String id, boolean shareCluster) {
         return Flux.defer(() -> {
-            LocalMessageConnection networkConnection = localGatewayConnector.addConnection("local:" + IDGenerator.SNOW_FLAKE_STRING.generate(), shareCluster);
+            LocalMessageConnection networkConnection = localGatewayConnector.addConnection(id, shareCluster);
             return networkConnection
                 .onLocalMessage()
                 .doOnSubscribe(sub -> subscriptions.forEach(networkConnection::addSubscription))
@@ -182,7 +187,7 @@ public class DefaultMessageGateway implements MessageGateway {
             //加载会话已有的订阅信息
             session.getSubscriptions()
                 .map(Subscription::getTopic)
-                .flatMap(topic -> root.find(topic))
+                .flatMap(topic -> root.get(topic))
                 .subscribe(part -> part.addSessionId(getId()));
         }
 

jetlinks-components/io-component/src/main/java/org/jetlinks/community/io/excel/DefaultImportExportService.java

@@ -18,10 +18,15 @@ import java.io.InputStream;
 @Component
 public class DefaultImportExportService implements ImportExportService {
 
+    private final WebClient.Builder builder;
+
+    public DefaultImportExportService(WebClient.Builder builder) {
+        this.builder = builder;
+    }
 
     public <T> Flux<RowResult<T>> doImport(Class<T> clazz, String fileUrl) {
         return getInputStream(fileUrl)
-                .flatMapMany(inputStream -> ExcelReadDataListener.of(inputStream, clazz));
+            .flatMapMany(inputStream -> ExcelReadDataListener.of(inputStream, clazz));
     }
 
     @Override
@@ -31,16 +36,17 @@ public class DefaultImportExportService implements ImportExportService {
 
     public Mono<InputStream> getInputStream(String fileUrl) {
 
-        return Mono.defer(()->{
+        return Mono.defer(() -> {
             if (fileUrl.startsWith("http")) {
-               return WebClient.create().get()
+                return builder.build()
+                    .get()
                     .uri(fileUrl)
                     .accept(MediaType.APPLICATION_OCTET_STREAM)
                     .exchange()
                     .flatMap(clientResponse -> clientResponse.bodyToMono(Resource.class))
                     .flatMap(resource -> Mono.fromCallable(resource::getInputStream));
             } else {
-                return Mono.fromCallable(()->new FileInputStream(fileUrl));
+                return Mono.fromCallable(() -> new FileInputStream(fileUrl));
             }
         });
 

jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/session/MqttConnectionSession.java

@@ -9,6 +9,10 @@ import org.jetlinks.core.server.session.DeviceSession;
 import org.jetlinks.community.network.mqtt.server.MqttConnection;
 import reactor.core.publisher.Mono;
 
+import java.net.InetSocketAddress;
+import java.time.Duration;
+import java.util.Optional;
+
 public class MqttConnectionSession implements DeviceSession {
 
     @Getter
@@ -23,11 +27,11 @@ public class MqttConnectionSession implements DeviceSession {
     @Getter
     private MqttConnection connection;
 
-    public MqttConnectionSession(String id,DeviceOperator operator,Transport transport,MqttConnection connection){
-        this.id=id;
-        this.operator=operator;
-        this.transport=transport;
-        this.connection=connection;
+    public MqttConnectionSession(String id, DeviceOperator operator, Transport transport, MqttConnection connection) {
+        this.id = id;
+        this.operator = operator;
+        this.transport = transport;
+        this.connection = connection;
     }
 
     private long connectTime = System.currentTimeMillis();
@@ -60,7 +64,12 @@ public class MqttConnectionSession implements DeviceSession {
 
     @Override
     public void ping() {
+        connection.keepAlive();
+    }
 
+    @Override
+    public void setKeepAliveTimeout(Duration timeout) {
+        connection.setKeepAliveTimeout(timeout);
     }
 
     @Override
@@ -70,6 +79,11 @@ public class MqttConnectionSession implements DeviceSession {
 
     @Override
     public void onClose(Runnable call) {
+        connection.onClose(c -> call.run());
+    }
 
+    @Override
+    public Optional<InetSocketAddress> getClientAddress() {
+        return Optional.ofNullable(connection.getClientAddress());
     }
 }

jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/server/MqttConnection.java

@@ -6,6 +6,8 @@ import org.jetlinks.core.server.mqtt.MqttAuth;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import java.net.InetSocketAddress;
+import java.time.Duration;
 import java.util.Optional;
 import java.util.function.Consumer;
 
@@ -117,4 +119,10 @@ public interface MqttConnection {
     Mono<Void> close();
 
     long getLastPingTime();
+
+    void keepAlive();
+
+    void setKeepAliveTimeout(Duration duration);
+
+    InetSocketAddress getClientAddress();
 }

jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/server/vertx/VertxMqttConnection.java

@@ -5,6 +5,7 @@ import io.netty.buffer.Unpooled;
 import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
 import io.netty.handler.codec.mqtt.MqttQoS;
 import io.vertx.core.buffer.Buffer;
+import io.vertx.core.net.SocketAddress;
 import io.vertx.mqtt.MqttEndpoint;
 import io.vertx.mqtt.MqttTopicSubscription;
 import io.vertx.mqtt.messages.MqttPublishMessage;
@@ -26,7 +27,9 @@ import reactor.core.publisher.FluxSink;
 import reactor.core.publisher.Mono;
 
 import javax.annotation.Nonnull;
+import java.net.InetSocketAddress;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.Optional;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -36,7 +39,7 @@ import java.util.stream.Collectors;
 class VertxMqttConnection implements MqttConnection {
 
     private MqttEndpoint endpoint;
-    private long keepAliveTimeout;
+    private long keepAliveTimeoutMs;
     @Getter
     private long lastPingTime = System.currentTimeMillis();
     private volatile boolean closed = false, accepted = false, autoAckSub = true, autoAckUnSub = true, autoAckMsg = true;
@@ -50,7 +53,7 @@ class VertxMqttConnection implements MqttConnection {
 
     public VertxMqttConnection(MqttEndpoint endpoint) {
         this.endpoint = endpoint;
-        this.keepAliveTimeout = (endpoint.keepAliveTimeSeconds() + 10) * 1000L;
+        this.keepAliveTimeoutMs = (endpoint.keepAliveTimeSeconds() + 10) * 1000L;
     }
 
     private final Consumer<MqttConnection> defaultListener = mqttConnection -> {
@@ -58,6 +61,7 @@ class VertxMqttConnection implements MqttConnection {
         subscription.onComplete();
         unsubscription.onComplete();
         messageProcessor.onComplete();
+
     };
 
     private Consumer<MqttConnection> disconnectConsumer = defaultListener;
@@ -113,6 +117,11 @@ class VertxMqttConnection implements MqttConnection {
         return this;
     }
 
+    @Override
+    public void keepAlive() {
+        ping();
+    }
+
     void ping() {
         lastPingTime = System.currentTimeMillis();
     }
@@ -184,6 +193,20 @@ class VertxMqttConnection implements MqttConnection {
             });
     }
 
+    @Override
+    public void setKeepAliveTimeout(Duration duration) {
+        keepAliveTimeoutMs = duration.toMillis();
+    }
+
+    @Override
+    public InetSocketAddress getClientAddress() {
+
+        SocketAddress address = endpoint.remoteAddress();
+        if (address != null) {
+            return new InetSocketAddress(address.host(), address.port());
+        }
+        return null;
+    }
 
     @Override
     public String getClientId() {
@@ -236,7 +259,7 @@ class VertxMqttConnection implements MqttConnection {
 
     @Override
     public boolean isAlive() {
-        return endpoint.isConnected() && ((System.currentTimeMillis() - lastPingTime) < keepAliveTimeout);
+        return endpoint.isConnected() && (keepAliveTimeoutMs < 0 || ((System.currentTimeMillis() - lastPingTime) < keepAliveTimeoutMs));
     }
 
     @Override
@@ -255,6 +278,7 @@ class VertxMqttConnection implements MqttConnection {
         }
         closed = true;
         disconnectConsumer.accept(this);
+        disconnectConsumer = defaultListener;
     }
 
     @AllArgsConstructor

jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/client/TcpClient.java

@@ -6,6 +6,7 @@ import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.net.InetSocketAddress;
+import java.time.Duration;
 
 /**
  * TCP 客户端
@@ -43,4 +44,10 @@ public interface TcpClient extends Network {
      */
     void keepAlive();
 
+    /**
+     * 设置客户端心跳超时时间
+     *
+     * @param timeout 超时时间
+     */
+    void setKeepAliveTimeout(Duration timeout);
 }

jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/client/VertxTcpClient.java

@@ -33,7 +33,7 @@ public class VertxTcpClient extends AbstractTcpClient {
     private String id;
 
     @Setter
-    private long keepAliveTimeout = Duration.ofMinutes(10).toMillis();
+    private long keepAliveTimeoutMs = Duration.ofMinutes(10).toMillis();
 
     private volatile long lastKeepAliveTime = System.currentTimeMillis();
 
@@ -44,9 +44,14 @@ public class VertxTcpClient extends AbstractTcpClient {
         lastKeepAliveTime = System.currentTimeMillis();
     }
 
+    @Override
+    public void setKeepAliveTimeout(Duration timeout) {
+        keepAliveTimeoutMs = timeout.toMillis();
+    }
+
     @Override
     public boolean isAlive() {
-        return socket != null && (System.currentTimeMillis() - lastKeepAliveTime < keepAliveTimeout);
+        return socket != null && (keepAliveTimeoutMs < 0 || System.currentTimeMillis() - lastKeepAliveTime < keepAliveTimeoutMs);
     }
 
     @Override

jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/client/VertxTcpClientProvider.java

@@ -59,7 +59,7 @@ public class VertxTcpClientProvider implements NetworkProvider<TcpClientProperti
         client.setRecordParser(payloadParserBuilder.build(properties.getParserType(), Values.of(properties.getParserConfiguration())));
         NetClient netClient = vertx.createNetClient(properties.getOptions());
         client.setClient(netClient);
-        client.setKeepAliveTimeout(properties.getLong("keepAliveTimeout").orElse(Duration.ofMinutes(10).toMillis()));
+        client.setKeepAliveTimeoutMs(properties.getLong("keepAliveTimeout").orElse(Duration.ofMinutes(10).toMillis()));
         netClient.connect(properties.getPort(), properties.getHost(), result -> {
             if (result.succeeded()) {
                 log.debug("connect tcp [{}:{}] success", properties.getHost(), properties.getPort());

jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/device/TcpDeviceSession.java

@@ -9,6 +9,10 @@ import org.jetlinks.community.network.tcp.TcpMessage;
 import org.jetlinks.community.network.tcp.client.TcpClient;
 import reactor.core.publisher.Mono;
 
+import java.net.InetSocketAddress;
+import java.time.Duration;
+import java.util.Optional;
+
 class TcpDeviceSession implements DeviceSession {
 
     @Getter
@@ -64,11 +68,22 @@ class TcpDeviceSession implements DeviceSession {
         client.keepAlive();
     }
 
+
+    @Override
+    public void setKeepAliveTimeout(Duration timeout) {
+        client.setKeepAliveTimeout(timeout);
+    }
+
     @Override
     public boolean isAlive() {
         return client.isAlive();
     }
 
+    @Override
+    public Optional<InetSocketAddress> getClientAddress() {
+        return Optional.ofNullable(client.getRemoteAddress());
+    }
+
     @Override
     public void onClose(Runnable call) {
         client.onDisconnect(call);

jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/server/VertxTcpServer.java

@@ -57,7 +57,7 @@ public class VertxTcpServer extends AbstractTcpServer implements TcpServer {
 
     protected void acceptTcpConnection(NetSocket socket) {
         VertxTcpClient client = new VertxTcpClient(id + "_" + socket.remoteAddress());
-        client.setKeepAliveTimeout(keepAliveTimeout);
+        client.setKeepAliveTimeoutMs(keepAliveTimeout);
         try {
             socket.exceptionHandler(err -> {
                 log.error("tcp server client [{}] error", socket.remoteAddress(), err);

jetlinks-manager/authentication-manager/src/main/java/org/jetlinks/community/auth/web/OrganizationController.java

@@ -0,0 +1,69 @@
+package org.jetlinks.community.auth.web;
+
+import org.hswebframework.web.api.crud.entity.PagerResult;
+import org.hswebframework.web.api.crud.entity.QueryParamEntity;
+import org.hswebframework.web.api.crud.entity.TreeSupportEntity;
+import org.hswebframework.web.authorization.annotation.*;
+import org.hswebframework.web.system.authorization.api.entity.DimensionEntity;
+import org.hswebframework.web.system.authorization.defaults.service.DefaultDimensionService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.*;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+@RequestMapping("/organization")
+@RestController
+@Resource(id = "organization", name = "机构管理")
+public class OrganizationController {
+
+    static String orgDimensionTypeId = "org";
+
+    @Autowired
+    private DefaultDimensionService dimensionService;
+
+    @GetMapping("/_all/tree")
+    @Authorize(merge = false)
+    public Flux<DimensionEntity> getAllOrgTree() {
+        return getAllOrg()
+            .collectList()
+            .flatMapIterable(list -> TreeSupportEntity.list2tree(list, DimensionEntity::setChildren));
+    }
+
+    @GetMapping("/_all")
+    @Authorize(merge = false)
+    public Flux<DimensionEntity> getAllOrg() {
+        return dimensionService
+            .createQuery()
+            .where(DimensionEntity::getTypeId, orgDimensionTypeId)
+            .fetch();
+    }
+
+    @GetMapping("/_query")
+    @QueryAction
+    public Mono<PagerResult<DimensionEntity>> queryDimension(QueryParamEntity entity) {
+        return entity
+            .toNestQuery(q -> q.where(DimensionEntity::getTypeId, orgDimensionTypeId))
+            .execute(Mono::just)
+            .as(dimensionService::queryPager);
+    }
+
+    @PatchMapping
+    @SaveAction
+    public Mono<Void> saveOrg(@RequestBody Flux<DimensionEntity> entityFlux) {
+        return entityFlux
+            .doOnNext(entity -> entity.setTypeId(orgDimensionTypeId))
+            .as(dimensionService::save)
+            .then();
+    }
+
+    @DeleteMapping
+    @DeleteAction
+    public Mono<Void> deleteOrg(@RequestBody Flux<DimensionEntity> entityFlux) {
+        return entityFlux
+            .doOnNext(entity -> entity.setTypeId(orgDimensionTypeId))
+            .as(dimensionService::save)
+            .then();
+    }
+
+
+}

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/measurements/message/DeviceMessageMeasurementProvider.java

@@ -8,34 +8,39 @@ import org.jetlinks.community.device.message.DeviceMessageUtils;
 import org.jetlinks.community.device.timeseries.DeviceTimeSeriesMetric;
 import org.jetlinks.community.gateway.MessageGateway;
 import org.jetlinks.community.gateway.TopicMessage;
+import org.jetlinks.community.gateway.annotation.Subscribe;
 import org.jetlinks.community.micrometer.MeterRegistryManager;
 import org.jetlinks.community.timeseries.TimeSeriesManager;
 import org.jetlinks.core.message.DeviceMessage;
 import org.springframework.stereotype.Component;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
 import java.time.Duration;
 
 @Component
 public class DeviceMessageMeasurementProvider extends StaticMeasurementProvider {
 
+    MeterRegistry registry;
+
     public DeviceMessageMeasurementProvider(MessageGateway messageGateway,
                                             MeterRegistryManager registryManager,
                                             TimeSeriesManager timeSeriesManager) {
         super(DeviceDashboardDefinition.instance, DeviceObjectDefinition.message);
-        addMeasurement(new DeviceMessageMeasurement(messageGateway, timeSeriesManager));
-
-
-        MeterRegistry registry = registryManager.getMeterRegister(DeviceTimeSeriesMetric.deviceMetrics().getId(),
+        registry = registryManager.getMeterRegister(DeviceTimeSeriesMetric.deviceMetrics().getId(),
             "target", "msgType", "productId");
 
-        //订阅设备消息,用于统计设备消息量
-        messageGateway.subscribe("/device/*/message/**")
-            .map(this::convertTags)
-            .subscribe(tags -> registry
-                .counter("message-count", tags)
-                .increment());
+        addMeasurement(new DeviceMessageMeasurement(messageGateway, timeSeriesManager));
+
+    }
 
+    @Subscribe("/device/*/message/**")
+    public Mono<Void> incrementMessage(TopicMessage message) {
+        return Mono.fromRunnable(() -> {
+            registry
+                .counter("message-count", convertTags(message))
+                .increment();
+        });
     }
 
     static final String[] empty = new String[0];

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/measurements/status/DeviceStatusMeasurementProvider.java

@@ -10,9 +10,11 @@ import org.jetlinks.community.device.service.LocalDeviceInstanceService;
 import org.jetlinks.community.device.timeseries.DeviceTimeSeriesMetric;
 import org.jetlinks.community.gateway.MessageGateway;
 import org.jetlinks.community.gateway.TopicMessage;
+import org.jetlinks.community.gateway.annotation.Subscribe;
 import org.jetlinks.community.micrometer.MeterRegistryManager;
 import org.jetlinks.community.timeseries.TimeSeriesManager;
 import org.springframework.stereotype.Component;
+import reactor.core.publisher.Mono;
 
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -22,6 +24,19 @@ import java.util.function.Function;
 @Component
 public class DeviceStatusMeasurementProvider extends StaticMeasurementProvider {
 
+    private MeterRegistry registry;
+
+    Map<String, LongAdder> productCounts = new ConcurrentHashMap<>();
+
+    Function<String, LongAdder> counterAdder = productId ->
+        productCounts.computeIfAbsent(productId, __id -> {
+            LongAdder adder = new LongAdder();
+            Gauge.builder("online-count", adder, LongAdder::sum)
+                .tag("productId", __id)
+                .register(registry);
+            return adder;
+        });
+
     public DeviceStatusMeasurementProvider(MeterRegistryManager registryManager,
                                            LocalDeviceInstanceService instanceService,
                                            TimeSeriesManager timeSeriesManager,
@@ -32,37 +47,30 @@ public class DeviceStatusMeasurementProvider extends StaticMeasurementProvider {
 
         addMeasurement(new DeviceStatusRecordMeasurement(instanceService, timeSeriesManager));
 
-        MeterRegistry registry = registryManager.getMeterRegister(DeviceTimeSeriesMetric.deviceMetrics().getId(),
+        registry = registryManager.getMeterRegister(DeviceTimeSeriesMetric.deviceMetrics().getId(),
             "target", "msgType", "productId");
-        Map<String, LongAdder> productCounts = new ConcurrentHashMap<>();
+    }
 
-        Function<String, LongAdder> counterAdder = productId ->
-            productCounts.computeIfAbsent(productId, __id -> {
-                LongAdder adder = new LongAdder();
-                Gauge.builder("online-count", adder, LongAdder::sum)
-                    .tag("productId", __id)
-                    .register(registry);
-                return adder;
-            });
+    @Subscribe("/device/*/online")
+    public Mono<Void> incrementOnline(TopicMessage msg){
+        return Mono.fromRunnable(()->{
+            String productId = parseProductId(msg);
+            counterAdder.apply(productId).increment();
+            registry
+                .counter("online", "productId", productId)
+                .increment();
+        });
+    }
 
-        //上线
-        messageGateway.subscribe("/device/*/online")
-            .map(this::parseProductId)
-            .subscribe(productId -> {
-                counterAdder.apply(productId).increment();
-                registry
-                    .counter("online", "productId", productId)
-                    .increment();
-            });
-        //下线
-        messageGateway.subscribe("/device/*/offline")
-            .map(this::parseProductId)
-            .subscribe(productId -> {
-                counterAdder.apply(productId).decrement();
-                registry
-                    .counter("offline", "productId", productId)
-                    .increment();
-            });
+    @Subscribe("/device/*/offline")
+    public Mono<Void> incrementOffline(TopicMessage msg){
+        return Mono.fromRunnable(()->{
+            String productId = parseProductId(msg);
+            counterAdder.apply(productId).increment();
+            registry
+                .counter("offline", "productId", productId)
+                .increment();
+        });
     }
 
     private String parseProductId(TopicMessage msg) {

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/message/DeviceMessageConnector.java

@@ -1,6 +1,7 @@
 package org.jetlinks.community.device.message;
 
 import lombok.extern.slf4j.Slf4j;
+import org.jetlinks.core.Values;
 import org.jetlinks.core.device.DeviceRegistry;
 import org.jetlinks.core.message.*;
 import org.jetlinks.core.message.event.EventMessage;
@@ -14,6 +15,7 @@ import reactor.core.publisher.FluxSink;
 import reactor.core.publisher.Mono;
 
 import javax.annotation.Nonnull;
+import java.util.HashMap;
 import java.util.function.Function;
 
 /**
@@ -89,7 +91,6 @@ public class DeviceMessageConnector
             .then();
     }
 
-
     public Mono<String> getTopic(Message message) {
         if (message instanceof DeviceMessage) {
             DeviceMessage deviceMessage = ((DeviceMessage) message);
@@ -101,43 +102,16 @@ public class DeviceMessageConnector
                 .getDevice(deviceId)
                 //获取设备配置是可能存在的性能瓶颈
                 .flatMap(operator -> operator.getSelfConfigs(appendConfigHeader))
+                .switchIfEmpty(Mono.fromSupplier(() -> Values.of(new HashMap<>())))
                 .flatMap(configs -> {
                     configs.getAllValues().forEach(deviceMessage::addHeader);
-                    String topic;
-
-                    // TODO: 2019/12/28 自定义topic支持?
-
-                    if (message instanceof EventMessage) {   //事件
-                        EventMessage event = ((EventMessage) message);
-                        topic = "/device/" + deviceMessage.getDeviceId() + "/message/event/".concat(event.getEvent());
-                    } else if (message instanceof ReportPropertyMessage) {   //上报属性
-                        topic = "/device/" + deviceMessage.getDeviceId() + "/message/property/report";
-                    } else if (message instanceof DeviceOnlineMessage) {   //设备上线
-                        topic = "/device/" + deviceMessage.getDeviceId() + "/online";
-                    } else if (message instanceof DeviceOfflineMessage) {   //设备离线
-                        topic = "/device/" + deviceMessage.getDeviceId() + "/offline";
-                    } else if (message instanceof ChildDeviceMessage) { //子设备消息
-                        topic = "/device/" + deviceMessage.getDeviceId() + "/message/children";
+                    String topic = "/device/".concat(deviceId).concat(createDeviceMessageTopic(message));
+                    if (message instanceof ChildDeviceMessage) { //子设备消息
                         return onMessage(((ChildDeviceMessage) message).getChildDeviceMessage())
                             .thenReturn(topic);
                     } else if (message instanceof ChildDeviceMessageReply) { //子设备消息
-                        topic = "/device/" + deviceMessage.getDeviceId() + "/message/children/reply";
                         return onMessage(((ChildDeviceMessageReply) message).getChildDeviceMessage())
                             .thenReturn(topic);
-                    } else if (message instanceof ReadPropertyMessage) { //读取属性
-                        topic = "/device/" + deviceMessage.getDeviceId() + "/message/property/read";
-                    } else if (message instanceof WritePropertyMessage) { //修改属性
-                        topic = "/device/" + deviceMessage.getDeviceId() + "/message/property/write";
-                    } else if (message instanceof FunctionInvokeMessage) { //调用功能
-                        topic = "/device/" + deviceMessage.getDeviceId() + "/message/function/reply";
-                    } else if (message instanceof ReadPropertyMessageReply) { //读取属性回复
-                        topic = "/device/" + deviceMessage.getDeviceId() + "/message/property/read/reply";
-                    } else if (message instanceof WritePropertyMessageReply) { //修改属性回复
-                        topic = "/device/" + deviceMessage.getDeviceId() + "/message/property/write/reply";
-                    } else if (message instanceof FunctionInvokeMessageReply) { //调用功能回复
-                        topic = "/device/" + deviceMessage.getDeviceId() + "/message/function/reply";
-                    } else {
-                        topic = "/device/" + deviceMessage.getDeviceId() + "/message/unknown";
                     }
                     return Mono.just(topic);
                 });
@@ -146,6 +120,49 @@ public class DeviceMessageConnector
         return Mono.just("/device/unknown/message/unknown");
     }
 
+    public String createDeviceMessageTopic(Message message) {
+        if (message instanceof EventMessage) {   //事件
+            EventMessage event = ((EventMessage) message);
+            return "/message/event/".concat(event.getEvent());
+        } else if (message instanceof ReportPropertyMessage) {   //上报属性
+            return "/message/property/report";
+        } else if (message instanceof DeviceOnlineMessage) {   //设备上线
+            return "/online";
+        } else if (message instanceof DeviceOfflineMessage) {   //设备离线
+            return "/offline";
+        } else if (message instanceof ChildDeviceMessage) { //子设备消息
+            Message msg = ((ChildDeviceMessage) message).getChildDeviceMessage();
+            if (msg instanceof DeviceMessage) {
+                return "/message/children/".concat(((DeviceMessage) msg).getDeviceId()).concat(createDeviceMessageTopic(msg));
+            }
+            return "/message/children/".concat(createDeviceMessageTopic(message));
+        } else if (message instanceof ChildDeviceMessageReply) { //子设备消息
+            Message msg = ((ChildDeviceMessageReply) message).getChildDeviceMessage();
+            if (msg instanceof DeviceMessage) {
+                return "/message/children/reply/".concat(((DeviceMessage) msg).getDeviceId()).concat(createDeviceMessageTopic(msg));
+            }
+            return "/message/children/reply/".concat(createDeviceMessageTopic(message));
+        } else if (message instanceof ReadPropertyMessage) { //读取属性
+            return "/message/property/read";
+        } else if (message instanceof WritePropertyMessage) { //修改属性
+            return "/message/property/write";
+        } else if (message instanceof FunctionInvokeMessage) { //调用功能
+            return "/message/function/reply";
+        } else if (message instanceof ReadPropertyMessageReply) { //读取属性回复
+            return "/message/property/read/reply";
+        } else if (message instanceof WritePropertyMessageReply) { //修改属性回复
+            return "/message/property/write/reply";
+        } else if (message instanceof FunctionInvokeMessageReply) { //调用功能回复
+            return "/message/function/reply";
+        } else if (message instanceof DeviceRegisterMessage) { //注册
+            return "/register";
+        } else if (message instanceof DeviceUnRegisterMessage) { //注销
+            return "/unregister";
+        } else {
+            return "/message/unknown";
+        }
+    }
+
     @Nonnull
     @Override
     public Flux<MessageConnection> onConnection() {

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/service/LocalDeviceInstanceService.java

@@ -4,6 +4,7 @@ import com.alibaba.excel.EasyExcel;
 import com.alibaba.excel.ExcelWriter;
 import com.alibaba.excel.write.metadata.WriteSheet;
 import com.alibaba.fastjson.JSON;
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections4.CollectionUtils;
 import org.hswebframework.ezorm.core.dsl.Query;
@@ -19,12 +20,12 @@ import org.hswebframework.web.exception.NotFoundException;
 import org.hswebframework.web.logger.ReactiveLogger;
 import org.jetlinks.community.device.entity.DeviceOperationLogEntity;
 import org.jetlinks.community.device.message.DeviceMessageUtils;
+import org.jetlinks.community.gateway.Subscription;
+import org.jetlinks.community.gateway.annotation.Subscribe;
 import org.jetlinks.core.device.DeviceConfigKey;
 import org.jetlinks.core.device.DeviceOperator;
 import org.jetlinks.core.device.DeviceRegistry;
-import org.jetlinks.core.message.DeviceMessage;
-import org.jetlinks.core.message.DeviceOfflineMessage;
-import org.jetlinks.core.message.DeviceOnlineMessage;
+import org.jetlinks.core.message.*;
 import org.jetlinks.core.metadata.DataType;
 import org.jetlinks.core.metadata.DeviceMetadata;
 import org.jetlinks.core.metadata.EventMetadata;
@@ -100,6 +101,7 @@ public class LocalDeviceInstanceService extends GenericReactiveCrudService<Devic
             .doOnNext(instance -> instance.setState(null))
             .as(super::save);
     }
+
     /**
      * 获取设备所有信息
      *
@@ -112,7 +114,7 @@ public class LocalDeviceInstanceService extends GenericReactiveCrudService<Devic
             .zipWhen(instance -> deviceProductService.findById(instance.getProductId()), DeviceInfo::of) //产品型号信息
             .switchIfEmpty(Mono.error(NotFoundException::new))
             .zipWhen(deviceInfo -> getDeviceRunRealInfo(id), DeviceAllInfoResponse::of) //设备运行状态
-            .zipWhen(info -> getDeviceLatestProperties( id).collectList(), DeviceAllInfoResponse::ofProperties) //设备属性
+            .zipWhen(info -> getDeviceLatestProperties(id).collectList(), DeviceAllInfoResponse::ofProperties) //设备属性
             .zipWhen(info -> {
                     DeviceMetadata deviceMetadata = new JetLinksDeviceMetadata(JSON.parseObject(info.getDeviceInfo().getDeriveMetadata()));
                     return getEventCounts(deviceMetadata.getEvents(), id, info.getDeviceInfo().getProductId()); //事件数量统计
@@ -169,8 +171,7 @@ public class LocalDeviceInstanceService extends GenericReactiveCrudService<Devic
                     .id(instance.getId())
                     .productId(instance.getProductId())
                     .build()
-                    .addConfig(DeviceConfigKey.parentGatewayId, instance.getParentId())
-                )
+                    .addConfig(DeviceConfigKey.parentGatewayId, instance.getParentId()))
                     //设置其他配置信息
                     .flatMap(deviceOperator -> deviceOperator.getState()
                         .flatMap(r -> {
@@ -204,6 +205,7 @@ public class LocalDeviceInstanceService extends GenericReactiveCrudService<Devic
 
     /**
      * 取消发布(取消激活),取消后,设备无法再连接到服务. 注册中心也无法再获取到该设备信息.
+     *
      * @param id 设备ID
      * @return 取消结果
      */
@@ -326,13 +328,13 @@ public class LocalDeviceInstanceService extends GenericReactiveCrudService<Devic
 
         //订阅设备上下线
         FluxUtils.bufferRate(messageGateway
-            .subscribe("/device/*/online", "/device/*/offline")
+            .subscribe(Subscription.asList("/device/*/online", "/device/*/offline"), "device-state-synchronizer", false)
             .flatMap(message -> Mono.justOrEmpty(DeviceMessageUtils.convert(message))
                 .map(DeviceMessage::getDeviceId)), 800, 200, Duration.ofSeconds(2))
-            .flatMap(list -> syncStateBatch(Flux.just(list), false).count())
+            .publishOn(Schedulers.parallel())
+            .concatMap(list -> syncStateBatch(Flux.just(list), false).reduce(Math::addExact))
             .onErrorContinue((err, obj) -> log.error(err.getMessage(), err))
             .subscribe((i) -> log.info("同步设备状态成功:{}", i));
-
     }
 
     public Mono<DeviceInfo> getDeviceInfoById(String id) {
@@ -343,130 +345,90 @@ public class LocalDeviceInstanceService extends GenericReactiveCrudService<Devic
     }
 
     public Flux<Integer> syncStateBatch(Flux<List<String>> batch, boolean force) {
+
         return batch
-            .flatMap(list -> Flux.fromIterable(list)
-                .flatMap(registry::getDevice)
+            .concatMap(list -> Flux.fromIterable(list)
                 .publishOn(Schedulers.parallel())
+                .flatMap(registry::getDevice)
                 .flatMap(operation -> {
                     Mono<Byte> state = force ? operation.checkState() : operation.getState();
                     return Mono.zip(
-                        state,//状态
+                        state.defaultIfEmpty(org.jetlinks.core.device.DeviceState.offline),//状态
                         Mono.just(operation.getDeviceId()), //设备id
-                        operation.getConfig(DeviceConfigKey.isGatewayDevice)//是否为网关设备
+                        operation.getConfig(DeviceConfigKey.isGatewayDevice).defaultIfEmpty(false)//是否为网关设备
                     );
                 })
-                .groupBy(Tuple2::getT1, Function.identity())
+                .collect(Collectors.groupingBy(Tuple2::getT1))
+                .flatMapIterable(Map::entrySet)
+                .publishOn(Schedulers.single())
                 .flatMap(group -> {
-                    @SuppressWarnings("all")
-                    DeviceState state = group.key() == null ? DeviceState.offline : DeviceState.of(group.key());
-                    return group
-                        .collectList()
-                        .flatMap(idList -> Mono.zip(
-                            //修改设备状态
-                            getRepository()
-                                .createUpdate()
-                                .set(DeviceInstanceEntity::getState, state)
-                                .where()
-                                .in(DeviceInstanceEntity::getId, idList.stream().map(Tuple3::getT2).collect(Collectors.toList()))
-                                .execute(),
-                            //修改子设备状态
-                            Flux.fromIterable(idList)
-                                .filter(Tuple3::getT3)
-                                .map(Tuple3::getT2)
-                                .collectList()
-                                .filter(CollectionUtils::isNotEmpty)
-                                .flatMap(parents ->
-                                    getRepository()
-                                        .createUpdate()
-                                        .set(DeviceInstanceEntity::getState, state)
-                                        .where()
-                                        .in(DeviceInstanceEntity::getParentId, parents)
-                                        .execute()
-                                ).defaultIfEmpty(0)
-                            , Math::addExact));
+                    DeviceState state = DeviceState.of(group.getKey());
+                    return Mono.zip(
+                        //修改设备状态
+                        getRepository()
+                            .createUpdate()
+                            .set(DeviceInstanceEntity::getState, state)
+                            .where()
+                            .in(DeviceInstanceEntity::getId, group.getValue().stream().map(Tuple3::getT2).collect(Collectors.toList()))
+                            .execute()
+                            .defaultIfEmpty(0),
+                        Flux.fromIterable(group.getValue())
+                            .filter(Tuple3::getT3)
+                            .map(Tuple3::getT2)
+                            .collectList()
+                            .filter(CollectionUtils::isNotEmpty)
+                            .flatMap(parents ->
+                                getRepository()
+                                    .createUpdate()
+                                    .set(DeviceInstanceEntity::getState, state)
+                                    .where()
+                                    .in(DeviceInstanceEntity::getParentId, parents)
+                                    .execute()
+                            ).defaultIfEmpty(0), Math::addExact);
                 }));
     }
 
-    public Flux<ImportDeviceInstanceResult> doBatchImport(String fileUrl) {
-        return deviceProductService
-            .createQuery()
-            .fetch()
-            .collectList()
-            .flatMapMany(productEntities -> {
-                Map<String, String> productNameMap = productEntities.stream()
-                    .collect(Collectors.toMap(DeviceProductEntity::getName, DeviceProductEntity::getId, (_1, _2) -> _1));
-                return importExportService
-                    .doImport(DeviceInstanceImportExportEntity.class, fileUrl)
-                    .map(result -> {
-                        try {
-                            DeviceInstanceImportExportEntity importExportEntity = result.getResult();
-                            DeviceInstanceEntity entity = FastBeanCopier.copy(importExportEntity, new DeviceInstanceEntity());
-                            String productId = productNameMap.get(importExportEntity.getProductName());
-                            if (StringUtils.isEmpty(productId)) {
-                                throw new BusinessException("设备型号不存在");
-                            }
-                            if (StringUtils.isEmpty(entity.getId())) {
-                                throw new BusinessException("设备ID不能为空");
-                            }
 
-                            entity.setProductId(productId);
-                            entity.setState(DeviceState.notActive);
-                            return entity;
-                        } catch (Throwable e) {
-                            throw new BusinessException("第" +
-                                (result.getRowIndex() + 2)
-                                + "行:" + e.getMessage());
-                        }
-                    });
-            })
-            .buffer(50)
-            .flatMap(list -> this.save(Flux.fromIterable(list)))
-            .map(ImportDeviceInstanceResult::success)
-            .onErrorResume(err -> Mono.just(ImportDeviceInstanceResult.error(err)))
-            .doOnEach(ReactiveLogger.on(SignalType.CANCEL, (ctx, signal) -> {
-                log.warn("用户取消导入设备实例:{}", fileUrl);
-            }))
-            ;
+    @Subscribe("/device/*/message/children/*/register")
+    public Mono<Void> autoBindChildrenDevice(ChildDeviceMessage message) {
+        String childId = message.getChildDeviceId();
+        Message childMessage = message.getChildDeviceMessage();
+        if (childMessage instanceof DeviceRegisterMessage) {
+            return registry.getDevice(message.getDeviceId())
+                .flatMap(DeviceOperator::getState)
+                .flatMap(state -> createUpdate()
+                    .set(DeviceInstanceEntity::getParentId, message.getDeviceId())
+                    .set(DeviceInstanceEntity::getState, DeviceState.of(state))
+                    .where(DeviceInstanceEntity::getId, childId)
+                    .execute()
+                    .then(registry
+                        .getDevice(childId)
+                        .flatMap(dev -> dev.setConfig(DeviceConfigKey.parentGatewayId, message.getDeviceId())))
+                    .then());
+        }
+        return Mono.empty();
     }
 
-    private DataBufferFactory bufferFactory = new DefaultDataBufferFactory();
-
-    public Mono<Void> doExport(ServerHttpResponse response, QueryParam queryParam, String fileName) throws IOException {
-        response.getHeaders()
-            .set(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename=".concat(
-                URLEncoder.encode(fileName, StandardCharsets.UTF_8.displayName())));
-        queryParam.setPaging(false);
+    @Subscribe("/device/*/message/children/*/unregister")
+    public Mono<Void> autoUnbindChildrenDevice(ChildDeviceMessage message) {
+        String childId = message.getChildDeviceId();
+        Message childMessage = message.getChildDeviceMessage();
+        if (childMessage instanceof DeviceUnRegisterMessage) {
+
+            return registry.getDevice(childId)
+                .flatMap(dev -> dev
+                    .removeConfig(DeviceConfigKey.parentGatewayId.getKey())
+                    .then(dev.checkState()))
+                .flatMap(state -> createUpdate()
+                    .setNull(DeviceInstanceEntity::getParentId)
+                    .set(DeviceInstanceEntity::getState, DeviceState.of(state))
+                    .where(DeviceInstanceEntity::getId, childId)
+                    .execute()
+                    .then());
 
-        return response.writeWith(Flux.create(sink -> {
-                OutputStream outputStream = new OutputStream() {
-                    @Override
-                    public void write(byte[] b) {
-                        sink.next(bufferFactory.wrap(b));
-                    }
 
-                    @Override
-                    public void write(int b) {
-                        sink.next(bufferFactory.wrap(new byte[]{(byte) b}));
-                    }
-
-                    @Override
-                    public void close() {
-                        sink.complete();
-                    }
-                };
-                ExcelWriter excelWriter = EasyExcel.write(outputStream, DeviceInstanceImportExportEntity.class).build();
-                WriteSheet writeSheet = EasyExcel.writerSheet().build();
-                sink.onCancel(query(queryParam)
-                    .map(entity -> {
-                        return  FastBeanCopier.copy(entity, new DeviceInstanceImportExportEntity());
-                    })
-                    .buffer(100)
-                    .doOnNext(list -> excelWriter.write(list, writeSheet))
-                    .doFinally(s -> excelWriter.finish())
-                    .doOnError(sink::error)
-                    .subscribe());
-            })
-        );
+        }
+        return Mono.empty();
     }
 
 

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/web/DeviceInstanceController.java

@@ -1,7 +1,11 @@
 package org.jetlinks.community.device.web;
 
+import com.alibaba.excel.EasyExcel;
+import com.alibaba.excel.ExcelWriter;
+import com.alibaba.excel.write.metadata.WriteSheet;
 import io.swagger.annotations.ApiOperation;
 import lombok.Getter;
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.hswebframework.ezorm.core.param.QueryParam;
 import org.hswebframework.ezorm.core.param.TermType;
@@ -9,12 +13,19 @@ import org.hswebframework.ezorm.rdb.exception.DuplicateKeyException;
 import org.hswebframework.web.api.crud.entity.PagerResult;
 import org.hswebframework.web.api.crud.entity.QueryParamEntity;
 import org.hswebframework.web.authorization.Authentication;
+import org.hswebframework.web.authorization.Dimension;
 import org.hswebframework.web.authorization.annotation.Authorize;
 import org.hswebframework.web.authorization.annotation.QueryAction;
 import org.hswebframework.web.authorization.annotation.Resource;
 import org.hswebframework.web.authorization.annotation.SaveAction;
+import org.hswebframework.web.bean.FastBeanCopier;
 import org.hswebframework.web.crud.web.reactive.ReactiveServiceCrudController;
 import org.hswebframework.web.exception.BusinessException;
+import org.jetlinks.community.device.entity.DeviceProductEntity;
+import org.jetlinks.community.device.entity.excel.DeviceInstanceImportExportEntity;
+import org.jetlinks.community.device.enums.DeviceState;
+import org.jetlinks.community.device.service.LocalDeviceProductService;
+import org.jetlinks.community.io.excel.ImportExportService;
 import org.jetlinks.core.device.DeviceConfigKey;
 import org.jetlinks.core.device.DeviceOperator;
 import org.jetlinks.core.device.DeviceRegistry;
@@ -26,14 +37,25 @@ import org.jetlinks.community.device.service.LocalDeviceInstanceService;
 import org.jetlinks.community.device.timeseries.DeviceTimeSeriesMetric;
 import org.jetlinks.community.timeseries.TimeSeriesManager;
 import org.jetlinks.community.timeseries.TimeSeriesMetric;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.core.io.buffer.DefaultDataBufferFactory;
+import org.springframework.http.HttpHeaders;
 import org.springframework.http.MediaType;
 import org.springframework.http.server.reactive.ServerHttpResponse;
+import org.springframework.util.StringUtils;
 import org.springframework.web.bind.annotation.*;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.io.OutputStream;
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 @RestController
 @RequestMapping({"/device-instance", "/device/instance"})
@@ -50,10 +72,16 @@ public class DeviceInstanceController implements
 
     private final DeviceRegistry registry;
 
-    public DeviceInstanceController(LocalDeviceInstanceService service, TimeSeriesManager timeSeriesManager, DeviceRegistry registry) {
+    private final LocalDeviceProductService productService;
+
+    private final ImportExportService importExportService;
+
+    public DeviceInstanceController(LocalDeviceInstanceService service, TimeSeriesManager timeSeriesManager, DeviceRegistry registry, LocalDeviceProductService productService, ImportExportService importExportService) {
         this.service = service;
         this.timeSeriesManager = timeSeriesManager;
         this.registry = registry;
+        this.productService = productService;
+        this.importExportService = importExportService;
     }
 
     @GetMapping({
@@ -132,8 +160,10 @@ public class DeviceInstanceController implements
         return service
             .query(query.includes("id"))
             .map(DeviceInstanceEntity::getId)
-            .buffer(50)
-            .as(flux -> service.syncStateBatch(flux, true));
+            .buffer(200)
+            .publishOn(Schedulers.single())
+            .concatMap(flux -> service.syncStateBatch(Flux.just(flux), true))
+            .defaultIfEmpty(0);
     }
 
     //已废弃
@@ -176,7 +206,7 @@ public class DeviceInstanceController implements
     @QueryAction
     public Mono<PagerResult<DeviceOperationLogEntity>> queryDeviceLog(@PathVariable String deviceId,
                                                                       QueryParamEntity entity) {
-        return service.queryDeviceLog(deviceId,entity);
+        return service.queryDeviceLog(deviceId, entity);
     }
 
     //已废弃
@@ -194,15 +224,70 @@ public class DeviceInstanceController implements
 
         return Authentication
             .currentReactive()
-            .flatMapMany(auth -> service.doBatchImport(fileUrl));
+            .flatMapMany(auth -> productService
+                .createQuery()
+                .fetch()
+                .collectList()
+                .flatMapMany(productEntities -> {
+                    Map<String, String> productNameMap = productEntities.stream()
+                        .collect(Collectors.toMap(DeviceProductEntity::getName, DeviceProductEntity::getId, (_1, _2) -> _1));
+                    return importExportService
+                        .doImport(DeviceInstanceImportExportEntity.class, fileUrl)
+                        .map(result -> {
+                            try {
+                                DeviceInstanceImportExportEntity importExportEntity = result.getResult();
+                                DeviceInstanceEntity entity = FastBeanCopier.copy(importExportEntity, new DeviceInstanceEntity());
+                                String productId = productNameMap.get(importExportEntity.getProductName());
+                                if (StringUtils.isEmpty(productId)) {
+                                    throw new BusinessException("设备型号不存在");
+                                }
+                                if (StringUtils.isEmpty(entity.getId())) {
+                                    throw new BusinessException("设备ID不能为空");
+                                }
+
+                                entity.setProductId(productId);
+                                entity.setState(DeviceState.notActive);
+                                return entity;
+                            } catch (Throwable e) {
+                                throw new BusinessException("第" +
+                                    (result.getRowIndex() + 2)
+                                    + "行:" + e.getMessage());
+                            }
+                        });
+                })
+                .buffer(20)
+                .publishOn(Schedulers.single())
+                .concatMap(list -> service.save(Flux.fromIterable(list)))
+                .map(ImportDeviceInstanceResult::success));
     }
 
+    DataBufferFactory bufferFactory = new DefaultDataBufferFactory();
+
     @PostMapping("/export")
-    public Mono<Void> export(ServerHttpResponse response, QueryParam parameter) throws IOException {
+    @QueryAction
+    @SneakyThrows
+    public Mono<Void> export(ServerHttpResponse response, QueryParam parameter) {
+        response.getHeaders().set(HttpHeaders.CONTENT_DISPOSITION,
+            "attachment; filename=".concat(URLEncoder.encode("设备实例.xlsx", StandardCharsets.UTF_8.displayName())));
         return Authentication
             .currentReactive()
             .flatMap(auth -> {
-                return Mono.fromCallable(() -> service.doExport(response, parameter, "设备实例.xlsx"));
-            }).then();
+                parameter.setPaging(false);
+                return response.writeWith(Mono.create(sink -> {
+                    ByteArrayOutputStream out = new ByteArrayOutputStream();
+                    ExcelWriter excelWriter = EasyExcel.write(out, DeviceInstanceImportExportEntity.class).build();
+                    WriteSheet writeSheet = EasyExcel.writerSheet().build();
+                    service.query(parameter)
+                        .map(entity -> FastBeanCopier.copy(entity, new DeviceInstanceImportExportEntity()))
+                        .buffer(100)
+                        .doOnNext(list -> excelWriter.write(list, writeSheet))
+                        .doFinally(s -> {
+                            excelWriter.finish();
+                            sink.success(bufferFactory.wrap(out.toByteArray()));
+                        })
+                        .doOnError(sink::error)
+                        .subscribe();
+                }));
+            });
     }
 }

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/web/DeviceMessageController.java

@@ -157,6 +157,9 @@ public class DeviceMessageController {
 
     private static <R extends DeviceMessageReply, T> Function<R, T> mapReply(Function<R, T> function) {
         return reply -> {
+            if (ErrorCode.REQUEST_HANDLING.name().equals(reply.getCode())) {
+                throw new DeviceOperationException(ErrorCode.REQUEST_HANDLING, reply.getMessage());
+            }
             if (!reply.isSuccess()) {
                 throw new BusinessException(reply.getMessage(), reply.getCode());
             }

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/web/GatewayDeviceController.java

@@ -0,0 +1,170 @@
+package org.jetlinks.community.device.web;
+
+import org.apache.commons.collections4.CollectionUtils;
+import org.hswebframework.web.api.crud.entity.PagerResult;
+import org.hswebframework.web.api.crud.entity.QueryParamEntity;
+import org.hswebframework.web.authorization.annotation.QueryAction;
+import org.hswebframework.web.authorization.annotation.Resource;
+import org.hswebframework.web.authorization.annotation.SaveAction;
+import org.jetlinks.core.device.DeviceConfigKey;
+import org.jetlinks.core.device.DeviceRegistry;
+import org.jetlinks.community.device.entity.DeviceInstanceEntity;
+import org.jetlinks.community.device.entity.DeviceProductEntity;
+import org.jetlinks.community.device.enums.DeviceType;
+import org.jetlinks.community.device.service.LocalDeviceInstanceService;
+import org.jetlinks.community.device.service.LocalDeviceProductService;
+import org.jetlinks.community.device.web.response.GatewayDeviceInfo;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.*;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * 网关设备接口
+ *
+ * @author zhouhao
+ * @since 1.0
+ */
+@RestController
+@RequestMapping("/device/gateway")
+@Resource(id = "device-gateway", name = "网关设备管理")
+public class GatewayDeviceController {
+
+    @Autowired
+    private LocalDeviceInstanceService instanceService;
+
+    @Autowired
+    private LocalDeviceProductService productService;
+
+    @Autowired
+    private DeviceRegistry registry;
+
+    @SuppressWarnings("all")
+    private Mono<List<String>> getGatewayProductList() {
+        return productService
+            .createQuery()
+            .select(DeviceProductEntity::getId)
+            .where(DeviceProductEntity::getDeviceType, DeviceType.gateway)
+            .fetch()
+            .map(DeviceProductEntity::getId)
+            .collectList()
+            .filter(CollectionUtils::isNotEmpty);
+    }
+
+    @GetMapping("/_query")
+    @QueryAction
+    public Mono<PagerResult<GatewayDeviceInfo>> queryGatewayDevice(QueryParamEntity param) {
+        return getGatewayProductList()
+            .flatMap(productIdList ->
+                param.toNestQuery(q -> q.in(DeviceInstanceEntity::getProductId, productIdList))
+                    .execute(Mono::just)
+                    .as(instanceService::queryPager)
+                    .filter(r -> r.getTotal() > 0)
+                    .flatMap(result -> {
+                        Map<String, DeviceInstanceEntity> mapping =
+                            result.getData()
+                                .stream()
+                                .collect(Collectors.toMap(DeviceInstanceEntity::getId, Function.identity()));
+
+                        //查询所有子设备
+                        return instanceService.createQuery()
+                            .where()
+                            .in(DeviceInstanceEntity::getParentId, mapping.keySet())
+                            .fetch()
+                            .groupBy(DeviceInstanceEntity::getParentId)
+                            .flatMap(group -> {
+                                String parentId = group.key();
+                                return group
+                                    .collectList()
+                                    .map(children -> GatewayDeviceInfo.of(mapping.get(parentId), children));
+                            })
+                            .collectMap(GatewayDeviceInfo::getId)//收集所有有子设备的网关设备信息
+                            .defaultIfEmpty(Collections.emptyMap())
+                            .flatMapMany(map -> Flux.fromIterable(mapping.values())
+                                .flatMap(ins -> Mono.justOrEmpty(map.get(ins.getId()))
+                                    //处理没有子设备的网关信息
+                                    .switchIfEmpty(Mono.fromSupplier(() -> GatewayDeviceInfo.of(ins, Collections.emptyList())))))
+                            .collectList()
+                            .map(list -> PagerResult.of(result.getTotal(), list, param));
+                    }))
+            .defaultIfEmpty(PagerResult.empty());
+    }
+
+    @GetMapping("/{id}")
+    @QueryAction
+    public Mono<GatewayDeviceInfo> getGatewayInfo(@PathVariable String id) {
+        return Mono.zip(
+            instanceService.findById(id),
+            instanceService.createQuery()
+                .where()
+                .is(DeviceInstanceEntity::getParentId, id)
+                .fetch()
+                .collectList()
+                .defaultIfEmpty(Collections.emptyList()),
+            GatewayDeviceInfo::of);
+    }
+
+
+    @PostMapping("/{gatewayId}/bind/{deviceId}")
+    @SaveAction
+    public Mono<GatewayDeviceInfo> bindDevice(@PathVariable String gatewayId,
+                                              @PathVariable String deviceId) {
+        return instanceService
+            .createUpdate()
+            .set(DeviceInstanceEntity::getParentId, gatewayId)
+            .where(DeviceInstanceEntity::getId, deviceId)
+            .execute()
+            .then(registry
+                .getDevice(deviceId)
+                .flatMap(operator -> operator.setConfig(DeviceConfigKey.parentGatewayId, gatewayId)))
+            .then(getGatewayInfo(gatewayId));
+    }
+
+    @PostMapping("/{gatewayId}/bind")
+    @SaveAction
+    public Mono<GatewayDeviceInfo> bindDevice(@PathVariable String gatewayId,
+                                              @RequestBody Flux<String> deviceId) {
+
+        return deviceId
+            .collectList()
+            .filter(CollectionUtils::isNotEmpty)
+            .flatMap(deviceIdList -> instanceService
+                .createUpdate()
+                .set(DeviceInstanceEntity::getParentId, gatewayId)
+                .where()
+                .in(DeviceInstanceEntity::getId, deviceIdList)
+                .execute()
+                .then(Flux
+                    .fromIterable(deviceIdList)
+                    .flatMap(id -> registry
+                        .getDevice(id)
+                        .flatMap(operator -> operator.setConfig(DeviceConfigKey.parentGatewayId, gatewayId))).then()
+                )
+            ).then(getGatewayInfo(gatewayId));
+
+    }
+
+    @PostMapping("/{gatewayId}/unbind/{deviceId}")
+    @SaveAction
+    public Mono<GatewayDeviceInfo> unBindDevice(@PathVariable String gatewayId,
+                                                @PathVariable String deviceId) {
+        return instanceService
+            .createUpdate()
+            .setNull(DeviceInstanceEntity::getParentId)
+            .where(DeviceInstanceEntity::getId, deviceId)
+            .and(DeviceInstanceEntity::getParentId, gatewayId)
+            .execute()
+            .filter(i -> i > 0)
+            .flatMap(i -> registry
+                .getDevice(deviceId)
+                .flatMap(operator -> operator.removeConfig(DeviceConfigKey.parentGatewayId.getKey())))
+            .then(getGatewayInfo(gatewayId));
+    }
+
+}

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/web/ProtocolSupportController.java

@@ -1,16 +1,24 @@
 package org.jetlinks.community.device.web;
 
+import com.alibaba.fastjson.JSON;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.Setter;
+import org.hswebframework.utils.StringUtils;
 import org.hswebframework.web.authorization.annotation.Authorize;
 import org.hswebframework.web.authorization.annotation.QueryAction;
 import org.hswebframework.web.authorization.annotation.Resource;
 import org.hswebframework.web.authorization.annotation.SaveAction;
 import org.hswebframework.web.crud.web.reactive.ReactiveServiceCrudController;
+import org.jetlinks.community.device.web.protocol.ProtocolDetail;
+import org.jetlinks.community.device.web.protocol.ProtocolInfo;
+import org.jetlinks.community.device.web.protocol.TransportInfo;
+import org.jetlinks.community.device.web.request.ProtocolDecodeRequest;
+import org.jetlinks.community.device.web.request.ProtocolEncodeRequest;
 import org.jetlinks.core.ProtocolSupport;
 import org.jetlinks.core.ProtocolSupports;
+import org.jetlinks.core.message.Message;
 import org.jetlinks.core.message.codec.DefaultTransport;
 import org.jetlinks.core.message.codec.Transport;
 import org.jetlinks.core.metadata.ConfigMetadata;
@@ -18,6 +26,8 @@ import org.jetlinks.core.metadata.unit.ValueUnit;
 import org.jetlinks.core.metadata.unit.ValueUnits;
 import org.jetlinks.community.device.entity.ProtocolSupportEntity;
 import org.jetlinks.community.device.service.LocalProtocolSupportService;
+import org.jetlinks.supports.protocol.management.ProtocolSupportDefinition;
+import org.jetlinks.supports.protocol.management.ProtocolSupportLoader;
 import org.jetlinks.supports.protocol.management.ProtocolSupportLoaderProvider;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.*;
@@ -43,6 +53,9 @@ public class ProtocolSupportController implements
     @Autowired
     private List<ProtocolSupportLoaderProvider> providers;
 
+    @Autowired
+    private ProtocolSupportLoader supportLoader;
+
     @PostMapping("/{id}/_deploy")
     @SaveAction
     public Mono<Boolean> deploy(@PathVariable String id) {
@@ -88,37 +101,53 @@ public class ProtocolSupportController implements
             .map(TransportInfo::of);
     }
 
-    @GetMapping("/units")
-    @Authorize(merge = false)
-    public Flux<ValueUnit> allUnits() {
-        return Flux.fromIterable(ValueUnits.getAllUnit());
+    @PostMapping("/convert")
+    @QueryAction
+    public Mono<ProtocolDetail> convertToDetail(@RequestBody Mono<ProtocolSupportEntity> entity) {
+        return entity.map(ProtocolSupportEntity::toDeployDefinition)
+            .doOnNext(def -> def.setId("_debug"))
+            .flatMap(def -> supportLoader.load(def))
+            .flatMap(ProtocolDetail::of);
     }
 
-    @Getter
-    @Setter
-    @AllArgsConstructor(staticName = "of")
-    @NoArgsConstructor
-    public static class TransportInfo {
-        private String id;
-
-        private String name;
-
-        static TransportInfo of(Transport support) {
-            return of(support.getId(), support.getName());
-        }
+    @PostMapping("/decode")
+    @SaveAction
+    public Mono<String> decode(@RequestBody Mono<ProtocolDecodeRequest> entity) {
+        return entity
+            .<Object>flatMapMany(request -> {
+                ProtocolSupportDefinition supportEntity = request.getEntity().toDeployDefinition();
+                supportEntity.setId("_debug");
+                return supportLoader.load(supportEntity)
+                    .flatMapMany(protocol -> request
+                        .getRequest()
+                        .doDecode(protocol, null));
+            })
+            .collectList()
+            .map(JSON::toJSONString)
+            .onErrorResume(err-> Mono.just(StringUtils.throwable2String(err)));
     }
 
-    @Getter
-    @Setter
-    @AllArgsConstructor(staticName = "of")
-    @NoArgsConstructor
-    public static class ProtocolInfo {
-        private String id;
-
-        private String name;
+    @PostMapping("/encode")
+    @SaveAction
+    public  Mono<String> encode(@RequestBody Mono<ProtocolEncodeRequest> entity) {
+        return entity
+            .flatMapMany(request -> {
+                ProtocolSupportDefinition supportEntity = request.getEntity().toDeployDefinition();
+                supportEntity.setId("_debug");
+                return supportLoader.load(supportEntity)
+                    .flatMapMany(protocol -> request
+                        .getRequest()
+                        .doEncode(protocol, null));
+            })
+            .collectList()
+            .map(JSON::toJSONString)
+            .onErrorResume(err-> Mono.just(StringUtils.throwable2String(err)));
+    }
 
-        static ProtocolInfo of(ProtocolSupport support) {
-            return of(support.getId(), support.getName());
-        }
+    @GetMapping("/units")
+    @Authorize(merge = false)
+    public Flux<ValueUnit> allUnits() {
+        return Flux.fromIterable(ValueUnits.getAllUnit());
     }
+
 }

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/web/protocol/ProtocolDetail.java

@@ -0,0 +1,32 @@
+package org.jetlinks.community.device.web.protocol;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.Setter;
+import org.jetlinks.core.ProtocolSupport;
+import reactor.core.publisher.Mono;
+
+import java.util.List;
+
+@Getter
+@Setter
+@AllArgsConstructor
+public class ProtocolDetail {
+    private String id;
+
+    private String name;
+
+    private List<TransportDetail> transports;
+
+    public static Mono<ProtocolDetail> of(ProtocolSupport support) {
+        return support
+            .getSupportedTransport()
+            .flatMap(trans -> TransportDetail.of(support, trans))
+            .collectList()
+            .map(details -> new ProtocolDetail(support.getId(), support.getName(), details));
+    }
+}
+
+
+
+

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/web/protocol/ProtocolInfo.java

@@ -0,0 +1,21 @@
+package org.jetlinks.community.device.web.protocol;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+import org.jetlinks.core.ProtocolSupport;
+
+@Getter
+@Setter
+@AllArgsConstructor(staticName = "of")
+@NoArgsConstructor
+public class ProtocolInfo {
+    private String id;
+
+    private String name;
+
+    public static ProtocolInfo of(ProtocolSupport support) {
+        return of(support.getId(), support.getName());
+    }
+}

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/web/protocol/TransportDetail.java

@@ -0,0 +1,22 @@
+package org.jetlinks.community.device.web.protocol;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.Setter;
+import org.jetlinks.core.ProtocolSupport;
+import org.jetlinks.core.message.codec.Transport;
+import reactor.core.publisher.Mono;
+
+
+@Getter
+@Setter
+@AllArgsConstructor
+public class TransportDetail {
+    private String id;
+
+    private String name;
+
+    public static Mono<TransportDetail> of(ProtocolSupport support, Transport transport) {
+        return Mono.just(new TransportDetail(transport.getId(), transport.getName()));
+    }
+}

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/web/protocol/TransportInfo.java

@@ -0,0 +1,21 @@
+package org.jetlinks.community.device.web.protocol;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+import org.jetlinks.core.message.codec.Transport;
+
+@Getter
+@Setter
+@AllArgsConstructor(staticName = "of")
+@NoArgsConstructor
+public class TransportInfo {
+    private String id;
+
+    private String name;
+
+    public  static TransportInfo of(Transport support) {
+        return of(support.getId(), support.getName());
+    }
+}

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/web/protocol/TransportSupportType.java

@@ -0,0 +1,18 @@
+package org.jetlinks.community.device.web.protocol;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import org.hswebframework.web.dict.EnumDict;
+
+@AllArgsConstructor
+@Getter
+public enum TransportSupportType implements EnumDict<String> {
+    ENCODE("编码"), DECODE("解码");
+    private String text;
+
+    @Override
+    public String getValue() {
+        return name();
+    }
+
+}

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/web/request/ProtocolDecodePayload.java

@@ -0,0 +1,57 @@
+package org.jetlinks.community.device.web.request;
+
+import com.alibaba.fastjson.JSON;
+import lombok.Getter;
+import lombok.Setter;
+import org.hswebframework.web.bean.FastBeanCopier;
+import org.jetlinks.core.ProtocolSupport;
+import org.jetlinks.core.device.DeviceOperator;
+import org.jetlinks.core.message.Message;
+import org.jetlinks.core.message.codec.*;
+import org.jetlinks.core.server.session.DeviceSession;
+import org.jetlinks.rule.engine.executor.PayloadType;
+import org.reactivestreams.Publisher;
+
+import javax.annotation.Nullable;
+
+@Getter
+@Setter
+public class ProtocolDecodePayload {
+
+    private DefaultTransport transport;
+
+    private PayloadType payloadType = PayloadType.STRING;
+
+    private String payload;
+
+    public EncodedMessage toEncodedMessage() {
+        if (transport == DefaultTransport.MQTT || transport == DefaultTransport.MQTT_TLS) {
+            SimpleMqttMessage message = FastBeanCopier.copy(JSON.parseObject(payload), new SimpleMqttMessage());
+            message.setPayloadType(MessagePayloadType.of(payloadType.getId()));
+            return message;
+        }
+        return EncodedMessage.simple(payloadType.write(payload));
+    }
+
+    public Publisher<? extends Message> doDecode(ProtocolSupport support, DeviceOperator deviceOperator) {
+        return support
+            .getMessageCodec(getTransport())
+            .flatMapMany(codec -> codec.decode(new FromDeviceMessageContext() {
+                @Override
+                public EncodedMessage getMessage() {
+                    return toEncodedMessage();
+                }
+
+                @Override
+                public DeviceSession getSession() {
+                    return null;
+                }
+
+                @Nullable
+                @Override
+                public DeviceOperator getDevice() {
+                    return deviceOperator;
+                }
+            }));
+    }
+}

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/web/request/ProtocolDecodeRequest.java

@@ -0,0 +1,15 @@
+package org.jetlinks.community.device.web.request;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.jetlinks.community.device.entity.ProtocolSupportEntity;
+
+@Getter
+@Setter
+public class ProtocolDecodeRequest {
+
+    ProtocolSupportEntity entity;
+
+    ProtocolDecodePayload request;
+
+}

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/web/request/ProtocolEncodePayload.java

@@ -0,0 +1,58 @@
+package org.jetlinks.community.device.web.request;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
+import lombok.Getter;
+import lombok.Setter;
+import org.jetlinks.core.ProtocolSupport;
+import org.jetlinks.core.device.DeviceOperator;
+import org.jetlinks.core.message.Message;
+import org.jetlinks.core.message.MessageType;
+import org.jetlinks.core.message.codec.DefaultTransport;
+import org.jetlinks.core.message.codec.MessageEncodeContext;
+import org.jetlinks.core.message.codec.MqttMessage;
+import org.jetlinks.rule.engine.executor.PayloadType;
+import org.reactivestreams.Publisher;
+
+import javax.annotation.Nullable;
+
+@Getter
+@Setter
+public class ProtocolEncodePayload {
+
+    private DefaultTransport transport;
+
+    private String payload;
+
+    private PayloadType payloadType = PayloadType.STRING;
+
+    public Message toDeviceMessage() {
+        return MessageType.convertMessage(JSON.parseObject(payload))
+            .orElseThrow(() -> new IllegalArgumentException("无法识别的消息"));
+    }
+
+    public Publisher<Object> doEncode(ProtocolSupport support, DeviceOperator operator) {
+        return support.getMessageCodec(getTransport())
+            .flatMapMany(codec -> codec.encode(new MessageEncodeContext() {
+                @Override
+                public Message getMessage() {
+                    return toDeviceMessage();
+                }
+
+                @Nullable
+                @Override
+                public DeviceOperator getDevice() {
+                    return operator;
+                }
+            }))
+            .map(msg -> {
+                if (msg instanceof MqttMessage) {
+                    JSONObject obj = (JSONObject) JSON.toJSON(msg);
+                    obj.put("payload", payloadType.read(msg.getPayload()));
+                    obj.remove("bytes");
+                    return obj;
+                }
+                return getPayloadType().read(msg.getPayload());
+            });
+    }
+}

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/web/request/ProtocolEncodeRequest.java

@@ -0,0 +1,16 @@
+package org.jetlinks.community.device.web.request;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.jetlinks.community.device.entity.ProtocolSupportEntity;
+
+@Getter
+@Setter
+public class ProtocolEncodeRequest {
+
+    ProtocolSupportEntity entity;
+
+    ProtocolEncodePayload request;
+
+
+}

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/web/response/ChildrenDeviceInfo.java

@@ -0,0 +1,29 @@
+package org.jetlinks.community.device.web.response;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.jetlinks.community.device.entity.DeviceInstanceEntity;
+import org.jetlinks.community.device.enums.DeviceState;
+
+@Getter
+@Setter
+public class ChildrenDeviceInfo {
+
+    private String id;
+
+    private String name;
+
+    private String description;
+
+    private DeviceState state;
+
+    public static ChildrenDeviceInfo of(DeviceInstanceEntity instance){
+        ChildrenDeviceInfo deviceInfo=new ChildrenDeviceInfo();
+        deviceInfo.setId(instance.getId());
+        deviceInfo.setName(instance.getName());
+        deviceInfo.setState(instance.getState());
+        deviceInfo.setDescription(instance.getDescribe());
+
+        return deviceInfo;
+    }
+}

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/web/response/GatewayDeviceInfo.java

@@ -0,0 +1,40 @@
+package org.jetlinks.community.device.web.response;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.jetlinks.community.device.entity.DeviceInstanceEntity;
+import org.jetlinks.community.device.enums.DeviceState;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * 网关设备信息
+ */
+@Getter
+@Setter
+public class GatewayDeviceInfo {
+
+    private String id;
+
+    private String name;
+
+    private String description;
+
+    private DeviceState state;
+
+    private List<ChildrenDeviceInfo> children;
+
+
+    public static GatewayDeviceInfo of(DeviceInstanceEntity gateway, List<DeviceInstanceEntity> children) {
+
+        GatewayDeviceInfo info = new GatewayDeviceInfo();
+        info.setId(gateway.getId());
+        info.setName(gateway.getName());
+        info.setDescription(gateway.getDescribe());
+        info.setState(gateway.getState());
+        info.setChildren(children.stream().map(ChildrenDeviceInfo::of).collect(Collectors.toList()));
+
+        return info;
+    }
+}

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/configuration/DefaultDeviceSessionManager.java

@@ -121,7 +121,7 @@ public class DefaultDeviceSessionManager implements DeviceSessionManager {
                         return session
                             .getOperator()
                             .online(serverId, session.getId())
-                            .then(Mono.fromRunnable(()-> registerListener.next(session)));
+                            .then(Mono.fromRunnable(() -> registerListener.next(session)));
                     })
                     .flatMap(ignore -> session.getOperator().online(serverId, session.getId()))
                     .thenReturn(false);
@@ -283,7 +283,7 @@ public class DefaultDeviceSessionManager implements DeviceSessionManager {
 
         //注册中心上线
         session.getOperator()
-            .online(session.getServerId().orElse(serverId), session.getId())
+            .online(session.getServerId().orElse(serverId), session.getId(), session.getClientAddress().map(String::valueOf).orElse(null))
             .doFinally(s -> {
                 //通知
                 if (onDeviceRegister.hasDownstreams()) {

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/configuration/ErrorControllerAdvice.java

@@ -2,10 +2,12 @@ package org.jetlinks.community.standalone.configuration;
 
 import lombok.extern.slf4j.Slf4j;
 import org.hswebframework.web.crud.web.ResponseMessage;
+import org.jetlinks.core.enums.ErrorCode;
 import org.jetlinks.core.exception.DeviceOperationException;
 import org.springframework.core.Ordered;
 import org.springframework.core.annotation.Order;
 import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.ExceptionHandler;
 import org.springframework.web.bind.annotation.ResponseStatus;
 import org.springframework.web.bind.annotation.RestControllerAdvice;
@@ -17,9 +19,37 @@ import reactor.core.publisher.Mono;
 public class ErrorControllerAdvice {
 
     @ExceptionHandler
-    @ResponseStatus(HttpStatus.INTERNAL_SERVER_ERROR)
-    public Mono<ResponseMessage<?>> handleException(DeviceOperationException e) {
-        return Mono.just(ResponseMessage.error(e.getCode().name().toLowerCase(), e.getMessage()));
+    public Mono<ResponseEntity<ResponseMessage<Object>>> handleException(DeviceOperationException e) {
+
+        //202
+        if (e.getCode() == ErrorCode.REQUEST_HANDLING) {
+            return Mono.just(ResponseEntity
+                .status(HttpStatus.ACCEPTED)
+                .body(ResponseMessage.error(202,
+                    e.getCode().name().toLowerCase(),
+                    e.getMessage())
+                    .result("消息已发往设备,处理中...")));
+        }
+        if (e.getCode() == ErrorCode.FUNCTION_UNDEFINED
+            || e.getCode() == ErrorCode.PARAMETER_UNDEFINED) {
+            //404
+            return Mono.just(ResponseEntity
+                .status(HttpStatus.NOT_FOUND)
+                .body(ResponseMessage.error(e.getCode().name().toLowerCase(), e.getMessage())));
+        }
+
+        if (e.getCode() == ErrorCode.PARAMETER_UNDEFINED) {
+            //400
+            return Mono.just(ResponseEntity
+                .status(HttpStatus.BAD_REQUEST)
+                .body(ResponseMessage.error(e.getCode().name().toLowerCase(), e.getMessage())));
+        }
+
+        //500
+        return Mono.just(ResponseEntity
+            .status(HttpStatus.INTERNAL_SERVER_ERROR)
+            .body(ResponseMessage.error(e.getCode().name().toLowerCase(), e.getMessage())));
     }
 
+
 }