add jetlinks主题 透传主题、日志主题、时间同步主题

jetlinks-components/network-component/network-core/pom.xml

@@ -47,5 +47,11 @@
             <scope>compile</scope>
         </dependency>
 
+        <dependency>
+            <groupId>cn.hutool</groupId>
+            <artifactId>hutool-json</artifactId>
+            <version>5.5.0</version>
+        </dependency>
+
     </dependencies>
 </project>

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/support/JetLinksExtendMqttDeviceMessageCodec.java → jetlinks-components/network-component/network-core/src/main/java/org/jetlinks/community/support/JetLinksExtendMqttDeviceMessageCodec.java

@@ -1,14 +1,20 @@
-package org.jetlinks.community.standalone.support;
+package org.jetlinks.community.support;
 
+import cn.hutool.core.util.EnumUtil;
+import cn.hutool.core.util.HexUtil;
+import cn.hutool.core.util.StrUtil;
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
 import io.netty.buffer.Unpooled;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.jetlinks.community.support.message.DirectMessage;
 import org.jetlinks.core.device.DeviceConfigKey;
 import org.jetlinks.core.message.*;
 import org.jetlinks.core.message.codec.*;
 import reactor.core.publisher.Mono;
 import javax.annotation.Nonnull;
-import java.nio.charset.StandardCharsets;
+import java.util.regex.Pattern;
 
 /**
  * @author lifang
@@ -17,10 +23,12 @@ import java.nio.charset.StandardCharsets;
  * @Description TODO
  * @createTime 2021年08月27日 08:30:00
  */
+@Slf4j
 public class JetLinksExtendMqttDeviceMessageCodec extends JetlinksExtendTopicMessageCodec  implements DeviceMessageCodec {
 
     private Transport transport;
 
+    private final static Pattern pattern=Pattern.compile("^\\/[a-z1-9A-Z]+\\/[a-z1-9A-Z]+\\/direct");
 
     public JetLinksExtendMqttDeviceMessageCodec(Transport transport) {
         this.transport = transport;
@@ -64,11 +72,19 @@ public class JetLinksExtendMqttDeviceMessageCodec extends JetlinksExtendTopicMes
 
     @Nonnull
     @Override
+    @SneakyThrows
     public Mono<Message> decode(@Nonnull MessageDecodeContext context) {
         return Mono.fromSupplier(() -> {
             MqttMessage message = (MqttMessage) context.getMessage();
             String topic = message.getTopic();
-            String jsonData = message.getPayload().toString(StandardCharsets.UTF_8);
+            byte[] bytes = message.getPayload().array();
+            if(pattern.matcher(topic).matches()){
+                DirectMessage directMessage = new DirectMessage();
+                directMessage.setPayload(HexUtil.encodeHexStr(bytes));
+                return directMessage;
+            }
+            String jsonData = StrUtil.utf8Str(bytes);
+
 
             JSONObject object = JSON.parseObject(jsonData, JSONObject.class);
             if (object == null) {
@@ -77,4 +93,8 @@ public class JetLinksExtendMqttDeviceMessageCodec extends JetlinksExtendTopicMes
             return decode(topic, object).getMessage();
         });
     }
+
+    public static void main(String[] args) {
+        String pattern="/**/**/direct";
+    }
 }

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/support/JetLinksExtendProtocolSupportProvider.java → jetlinks-components/network-component/network-core/src/main/java/org/jetlinks/community/support/JetLinksExtendProtocolSupportProvider.java

@@ -1,4 +1,4 @@
-package org.jetlinks.community.standalone.support;
+package org.jetlinks.community.support;
 
 import lombok.extern.slf4j.Slf4j;
 import org.jetlinks.core.defaults.CompositeProtocolSupport;

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/support/JetlinksExtendTopicMessageCodec.java → jetlinks-components/network-component/network-core/src/main/java/org/jetlinks/community/support/JetlinksExtendTopicMessageCodec.java

@@ -1,12 +1,12 @@
-package org.jetlinks.community.standalone.support;
+package org.jetlinks.community.support;
 
 import com.alibaba.fastjson.JSONObject;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.Setter;
-import org.jetlinks.community.standalone.support.time.LogMessage;
-import org.jetlinks.community.standalone.support.time.TimeSyncMessage;
-import org.jetlinks.community.standalone.support.time.TimeSyncReplyMessage;
+import org.jetlinks.community.support.message.LogMessage;
+import org.jetlinks.community.support.message.TimeSyncMessage;
+import org.jetlinks.community.support.message.TimeSyncReplyMessage;
 import org.jetlinks.core.message.*;
 import org.jetlinks.core.message.event.EventMessage;
 import org.jetlinks.core.message.firmware.*;
@@ -23,7 +23,7 @@ import java.util.Optional;
  * @author lifang
  * @version 1.0.0
  * @ClassName JetlinksExtendTopicMessageCodec.java
- * @Description TODO
+ * @Description 增加了透传和
  * @createTime 2021年08月27日 08:33:00
  */
 public class JetlinksExtendTopicMessageCodec {
@@ -152,6 +152,12 @@ public class JetlinksExtendTopicMessageCodec {
             mqttData.put("messageId", message.getMessageId());
             mqttData.put("deviceId", deviceId);
             return new JetlinksExtendTopicMessageCodec.EncodedTopic(topic, mqttData);
+        } else if (message instanceof TimeSyncReplyMessage) {
+            String topic = "/" .concat(deviceId).concat("/time-sync/reply");
+            JSONObject mqttData = new JSONObject();
+            mqttData.put("messageId", message.getMessageId());
+            mqttData.put("timestamp", message.getTimestamp());
+            return new JetlinksExtendTopicMessageCodec.EncodedTopic(topic, mqttData);
         } else if (message instanceof RequestFirmwareMessageReply) {
             String topic = "/" .concat(deviceId).concat("/firmware/pull/reply");
             RequestFirmwareMessageReply firmwareMessage = ((RequestFirmwareMessageReply) message);

jetlinks-components/network-component/network-core/src/main/java/org/jetlinks/community/support/message/DirectMessage.java

@@ -0,0 +1,32 @@
+package org.jetlinks.community.support.message;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+import org.jetlinks.core.message.CommonDeviceMessage;
+import org.jetlinks.core.message.MessageType;
+
+import javax.annotation.Nonnull;
+
+/**
+ * @author lifang
+ * @version 1.0.0
+ * @ClassName DirectMessage.java
+ * @Description TODO
+ * @createTime 2021年08月27日 10:22:00
+ */
+@Data
+@EqualsAndHashCode(callSuper = false)
+@NoArgsConstructor
+public class DirectMessage extends CommonDeviceMessage {
+    /**
+     * 存储十六进制
+     */
+    @Nonnull
+    private String payload;
+
+    @Override
+    public MessageType getMessageType() {
+        return MessageType.DIRECT;
+    }
+}

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/support/time/LogMessage.java → jetlinks-components/network-component/network-core/src/main/java/org/jetlinks/community/support/message/LogMessage.java

@@ -1,4 +1,4 @@
-package org.jetlinks.community.standalone.support.time;
+package org.jetlinks.community.support.message;
 
 import org.jetlinks.core.message.CommonDeviceMessage;
 import org.jetlinks.core.message.MessageType;

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/support/time/TimeSyncMessage.java → jetlinks-components/network-component/network-core/src/main/java/org/jetlinks/community/support/message/TimeSyncMessage.java

@@ -1,6 +1,8 @@
-package org.jetlinks.community.standalone.support.time;
+package org.jetlinks.community.support.message;
 
+import cn.hutool.core.util.EnumUtil;
 import org.jetlinks.core.message.CommonDeviceMessage;
+import org.jetlinks.core.message.MessageType;
 import org.jetlinks.core.message.RepayableDeviceMessage;
 
 /**
@@ -11,6 +13,12 @@ import org.jetlinks.core.message.RepayableDeviceMessage;
  * @createTime 2021年08月27日 09:10:00
  */
 public class TimeSyncMessage extends CommonDeviceMessage implements RepayableDeviceMessage<TimeSyncReplyMessage> {
+
+    private static MessageType messageType= EnumUtil.likeValueOf(MessageType.class, "timeSync");
+    @Override
+    public MessageType getMessageType() {
+        return messageType;
+    }
     @Override
     public TimeSyncReplyMessage newReply() {
         return new TimeSyncReplyMessage().from(this);

jetlinks-components/network-component/network-core/src/main/java/org/jetlinks/community/support/message/TimeSyncReplyMessage.java

@@ -0,0 +1,21 @@
+package org.jetlinks.community.support.message;
+
+import cn.hutool.core.util.EnumUtil;
+import org.jetlinks.core.message.CommonDeviceMessageReply;
+import org.jetlinks.core.message.MessageType;
+
+/**
+ * @author lifang
+ * @version 1.0.0
+ * @ClassName TimeSyncReplyMessage.java
+ * @Description TODO
+ * @createTime 2021年08月27日 09:11:00
+ */
+public class TimeSyncReplyMessage extends CommonDeviceMessageReply<TimeSyncReplyMessage> {
+    private static MessageType messageType= EnumUtil.likeValueOf(MessageType.class, "timeSyncReply");
+    @Override
+    public MessageType getMessageType() {
+        return messageType;
+    }
+
+}

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/configuration/DeviceManagerConfiguration.java

@@ -8,14 +8,17 @@ import org.jetlinks.core.event.EventBus;
 import org.jetlinks.core.server.MessageHandler;
 import org.jetlinks.core.server.session.DeviceSessionManager;
 import org.jetlinks.supports.cluster.redis.RedisClusterManager;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.DependsOn;
 
 @Configuration
 public class DeviceManagerConfiguration {
 
     @Bean
+    @DependsOn("enumsExtend")
     public DeviceMessageConnector deviceMessageConnector(EventBus eventBus,
                                                          MessageHandler messageHandler,
                                                          DeviceSessionManager sessionManager,

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

@@ -1,5 +1,6 @@
 package org.jetlinks.community.device.message;
 
+import cn.hutool.core.util.EnumUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.jetlinks.community.PropertyConstants;
 import org.jetlinks.core.Values;
@@ -50,6 +51,10 @@ public class DeviceMessageConnector implements DecodedClientMessageHandler {
             EventMessage event = ((EventMessage) message);
             builder.append("/message/event/").append(event.getEvent());
         });
+        //时间同步
+        createFastBuilder(EnumUtil.likeValueOf(MessageType.class, "timeSync"), "/time-sync");
+        //时间同步回复
+        createFastBuilder(EnumUtil.likeValueOf(MessageType.class, "timeSyncReply"), "/time-sync/reply");
 
         //上报属性
         createFastBuilder(MessageType.REPORT_PROPERTY, "/message/property/report");

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

@@ -5,6 +5,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.jetlinks.community.device.message.DeviceMessageConnector;
 import org.jetlinks.community.device.service.data.DeviceDataService;
 import org.jetlinks.community.gateway.annotation.Subscribe;
+import org.jetlinks.community.support.message.TimeSyncReplyMessage;
 import org.jetlinks.core.device.DeviceOperator;
 import org.jetlinks.core.device.DeviceRegistry;
 import org.jetlinks.core.event.EventBus;
@@ -73,6 +74,27 @@ public class TimeSeriesMessageWriterConnector {
         return dataService.saveDeviceMessage(message);
     }
 
+
+    /**
+     * 时间同步
+     *
+     * @param message 设备消息
+     * @return void
+     */
+    @Subscribe(topics = "/device/**/*/time-sync", id = "device-time-sync")
+    public Mono<Void> timeSync(DeviceMessage message) {
+        return registry.getDevice(message.getDeviceId())
+            .map(DeviceOperator::messageSender)
+            .flatMap(sender->{
+                TimeSyncReplyMessage reply = new TimeSyncReplyMessage();
+                reply.setDeviceId(message.getDeviceId());
+                reply.setMessageId(message.getMessageId());
+                return sender.sendAndForget(reply);
+            })
+            .then();
+    }
+
+
 //    @PostConstruct
 //    public void clusterSubscribe(){
 //        //上行主题数据处理

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/JetLinksApplication.java

@@ -6,16 +6,19 @@ import org.hswebframework.web.authorization.events.AuthorizingHandleBeforeEvent;
 import org.hswebframework.web.crud.annotation.EnableEasyormRepository;
 import org.hswebframework.web.logging.aop.EnableAccessLogger;
 import org.hswebframework.web.logging.events.AccessLoggerAfterEvent;
+import org.jetlinks.community.support.message.TimeSyncMessage;
+import org.jetlinks.community.standalone.utils.EnumsUtils;
+import org.jetlinks.core.message.MessageType;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.boot.autoconfigure.elasticsearch.ElasticsearchRestClientAutoConfiguration;
-import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
 import org.springframework.cache.annotation.EnableCaching;
+import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Profile;
 import org.springframework.context.event.EventListener;
 import org.springframework.stereotype.Component;
-import reactor.core.publisher.Hooks;
+
 import javax.annotation.PostConstruct;
+import java.util.function.Supplier;
 
 
 @SpringBootApplication(scanBasePackages = "org.jetlinks.community", exclude = {
@@ -30,13 +33,7 @@ import javax.annotation.PostConstruct;
 public class JetLinksApplication {
 
     public static void main(String[] args) {
-        try {
-            SpringApplication.run(JetLinksApplication.class, args);
-        }catch (Exception e){
-            e.printStackTrace();
-        }
-
-
+        SpringApplication.run(JetLinksApplication.class, args);
     }
 
     @Profile("dev")
@@ -64,6 +61,27 @@ public class JetLinksApplication {
             log.info("{}=>{} {}-{}", event.getLogger().getIp(), event.getLogger().getUrl(), event.getLogger().getDescribe(), event.getLogger().getAction());
 
         }
+
+        @Bean
+        public Class<Void>  enumsExtend(){
+            //添加时间同步主题
+            EnumsUtils.addEnum(MessageType.class,"timeSync", new Class<?>[]{Supplier.class},new Object[]{new Supplier<TimeSyncMessage>() {
+
+                @Override
+                public TimeSyncMessage get() {
+                    return new TimeSyncMessage();
+                }
+            }});
+
+            //添加时间回复主题
+            EnumsUtils.addEnum(MessageType.class,"timeSyncReply", new Class<?>[]{Supplier.class},new Object[]{new Supplier<TimeSyncMessage>() {
+                @Override
+                public TimeSyncMessage get() {
+                    return new TimeSyncMessage();
+                }
+            }});
+            return Void.TYPE;
+        }
     }
 
 

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/support/time/TimeSyncReplyMessage.java

@@ -1,15 +0,0 @@
-package org.jetlinks.community.standalone.support.time;
-
-import org.jetlinks.core.message.CommonDeviceMessage;
-import org.jetlinks.core.message.CommonDeviceMessageReply;
-
-/**
- * @author lifang
- * @version 1.0.0
- * @ClassName TimeSyncReplyMessage.java
- * @Description TODO
- * @createTime 2021年08月27日 09:11:00
- */
-public class TimeSyncReplyMessage extends CommonDeviceMessageReply<TimeSyncReplyMessage> {
-
-}

jetlinks-standalone/src/main/java/org/jetlinks/community/standalone/utils/EnumsUtils.java

@@ -0,0 +1,125 @@
+package org.jetlinks.community.standalone.utils;
+
+import sun.reflect.ConstructorAccessor;
+import sun.reflect.FieldAccessor;
+import sun.reflect.ReflectionFactory;
+
+import java.lang.reflect.Array;
+import java.lang.reflect.Modifier;
+import java.util.*;
+import java.lang.reflect.AccessibleObject;
+import java.lang.reflect.Field;
+
+/**
+ * @author lifang
+ * @version 1.0.0
+ * @ClassName EnumsUtils.java
+ * @Description TODO
+ * @createTime 2021年08月27日 11:32:00
+ */
+public class EnumsUtils {
+    @SuppressWarnings("unchecked")
+    public static <T extends Enum<?>> void addEnum(Class<T> enumType, String enumName,Class<?>[] paramClass,Object[] paramValue) {
+
+        // 0. Sanity checks
+        if (!Enum.class.isAssignableFrom(enumType)) {
+            throw new RuntimeException("class " + enumType + " is not an instance of Enum");
+        }
+
+        // 1. Lookup "$VALUES" holder in enum class and get previous enum instances
+        Field valuesField = null;
+        Field[] fields = enumType.getDeclaredFields();
+        for (Field field : fields) {
+            if (field.getName().contains("$VALUES")) {
+                valuesField = field;
+                break;
+            }
+        }
+        AccessibleObject.setAccessible(new Field[] { valuesField }, true);
+
+        try {
+
+            // 2. Copy it
+            T[] previousValues = (T[]) valuesField.get(enumType);
+            List<T> values = new ArrayList<T>(Arrays.asList(previousValues));
+
+            // 3. build new enum
+            T newValue = (T) makeEnum(enumType, // The target enum class
+                enumName, // THE NEW ENUM INSTANCE TO BE DYNAMICALLY ADDED
+                values.size(),
+                //new Class<?>[] {}, // could be used to pass values to the enum constuctor if needed
+                paramClass,
+                //new Object[] {}
+                paramValue
+            ); // could be used to pass values to the enum constuctor if needed
+
+            // 4. add new value
+            values.add(newValue);
+            Object object=values.toArray((T[]) Array.newInstance(enumType, 0));
+            // 5. Set new values field
+            setFailsafeFieldValue(valuesField, null, object);
+
+            // 6. Clean enum cache
+            cleanEnumCache(enumType);
+
+        } catch (Exception e) {
+            e.printStackTrace();
+            throw new RuntimeException(e.getMessage(), e);
+        }
+    }
+
+    private static Object makeEnum(Class<?> enumClass, String value, int ordinal, Class<?>[] additionalTypes,
+                                   Object[] additionalValues) throws Exception {
+        Object[] parms = new Object[additionalValues.length + 2];
+        parms[0] = value;
+        parms[1] = Integer.valueOf(ordinal);
+        System.arraycopy(additionalValues, 0, parms, 2, additionalValues.length);
+        return enumClass.cast(getConstructorAccessor(enumClass, additionalTypes).newInstance(parms));
+    }
+
+    private static ReflectionFactory reflectionFactory = ReflectionFactory.getReflectionFactory();
+    private static void setFailsafeFieldValue(Field field, Object target, Object value) throws NoSuchFieldException,
+        IllegalAccessException {
+
+        // let's make the field accessible
+        field.setAccessible(true);
+        // next we change the modifier in the Field instance to
+        // not be final anymore, thus tricking reflection into
+        // letting us modify the static final field
+        Field modifiersField = Field.class.getDeclaredField("modifiers");
+        modifiersField.setAccessible(true);
+        int modifiers = modifiersField.getInt(field);
+
+        // blank out the final bit in the modifiers int
+        modifiers &= ~Modifier.FINAL;
+        modifiersField.setInt(field, modifiers);
+
+        FieldAccessor fa = reflectionFactory.newFieldAccessor(field, false);
+        fa.set(target, value);
+    }
+    private static void blankField(Class<?> enumClass, String fieldName) throws NoSuchFieldException,
+        IllegalAccessException {
+        for (Field field : Class.class.getDeclaredFields()) {
+            if (field.getName().contains(fieldName)) {
+                AccessibleObject.setAccessible(new Field[] { field }, true);
+                setFailsafeFieldValue(field, enumClass, null);
+                break;
+            }
+        }
+    }
+
+    private static void cleanEnumCache(Class<?> enumClass) throws NoSuchFieldException, IllegalAccessException {
+        blankField(enumClass, "enumConstantDirectory"); // Sun (Oracle?!?) JDK 1.5/6
+        blankField(enumClass, "enumConstants"); // IBM JDK
+    }
+
+    private static ConstructorAccessor getConstructorAccessor(Class<?> enumClass, Class<?>[] additionalParameterTypes)
+        throws NoSuchMethodException {
+        Class<?>[] parameterTypes = new Class[additionalParameterTypes.length + 2];
+        parameterTypes[0] = String.class;
+        parameterTypes[1] = int.class;
+        System.arraycopy(additionalParameterTypes, 0, parameterTypes, 2, additionalParameterTypes.length);
+        return reflectionFactory.newConstructorAccessor(enumClass.getDeclaredConstructor(parameterTypes));
+
+    }
+}

jetlinks-standalone/src/main/resources/META-INF/services/org.jetlinks.core.spi.ProtocolSupportProvider

@@ -1 +1 @@
-org.jetlinks.community.standalone.support.JetLinksExtendProtocolSupportProvider
+org.jetlinks.community.support.JetLinksExtendProtocolSupportProvider