自定义协议开发

环境准备和开发工具

  • JDK：1.8+
  • MAVEN：3.1+

注意

maven不要使用全局仓库配置，可能导致依赖无法下载

  • 开发工具：idea

第一步 通过idea创建maven工程：demo-protocol

第二步 修改pom文件，添加依赖

  1.     <dependencies>
  2.     <!-->jetlinks 核心依赖<-->
  3.         <dependency>
  4.             <groupId>org.jetlinks</groupId>
  5.             <artifactId>jetlinks-core</artifactId>
  6.             <version>1.1.1</version>
  7.         </dependency>
  8.         <!-->jetlinks 协议解析接口包<-->
  9.         <dependency>
  10.             <groupId>org.jetlinks</groupId>
  11.             <artifactId>jetlinks-supports</artifactId>
  12.             <version>1.1.1</version>
  13.         </dependency>
  14.         <!-->lombok，需要idea安装lombok插件，否则去掉<-->
  15.         <dependency>
  16.             <groupId>org.projectlombok</groupId>
  17.             <artifactId>lombok</artifactId>
  18.             <version>1.18.10</version>
  19.         </dependency>
  20.         <!-->vertx核心包，可以用来进行网络模拟测试<-->
  21.         <dependency>
  22.             <groupId>io.vertx</groupId>
  23.             <artifactId>vertx-core</artifactId>
  24.             <version>3.8.3</version>
  25.             <scope>test</scope>
  26.         </dependency>
  27.         <!-->单元测试包<--> 
  28.         <dependency>
  29.             <groupId>org.junit.jupiter</groupId>
  30.             <artifactId>junit-jupiter</artifactId>
  31.             <version>5.5.2</version>
  32.             <scope>test</scope>
  33.         </dependency>
  34.         <!-->logback日志<-->
  35.         <dependency>
  36.             <groupId>ch.qos.logback</groupId>
  37.             <artifactId>logback-classic</artifactId>
  38.             <version>1.2.3</version>
  39.             <scope>test</scope>
  40.         </dependency>
  42.     </dependencies>
  43.     <!-->netty组件<-->
  44.     <dependencyManagement>
  45.         <dependencies>
  46.             <dependency>
  47.                 <groupId>io.netty</groupId>
  48.                 <artifactId>netty-bom</artifactId>
  49.                 <version>${netty.version}</version>
  50.                 <type>pom</type>
  51.                 <scope>import</scope>
  52.             </dependency>
  53.         </dependencies>
  54.     </dependencyManagement>
  • 修改maven pom文件properties
  1.  <properties>
  2.         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  3.         <project.build.locales>zh_CN</project.build.locales>
  4.         <java.version>1.8</java.version>
  5.         <project.build.jdk>${java.version}</project.build.jdk>
  6.         <netty.version>4.1.45.Final</netty.version>
  7.     </properties>
  • 添加maven编译规则
  1.     <build>
  2.         <plugins>
  3.             <plugin>
  4.                 <groupId>org.apache.maven.plugins</groupId>
  5.                 <artifactId>maven-compiler-plugin</artifactId>
  6.                 <version>3.1</version>
  7.                 <configuration>
  8.                     <source>${project.build.jdk}</source>
  9.                     <target>${project.build.jdk}</target>
  10.                     <encoding>${project.build.sourceEncoding}</encoding>
  11.                 </configuration>
  12.             </plugin>
  13.         </plugins>
  14.     </build>
  • 添加hsweb私服和阿里云仓库
  1.     <repositories>
  2.         <repository>
  3.             <id>hsweb-nexus</id>
  4.             <name>Nexus Release Repository</name>
  5.             <url>http://nexus.hsweb.me/content/groups/public/</url>
  6.             <snapshots>
  7.                 <enabled>true</enabled>
  8.                 <updatePolicy>always</updatePolicy>
  9.             </snapshots>
  10.         </repository>
  11.         <repository>
  12.             <id>aliyun-nexus</id>
  13.             <name>aliyun</name>
  14.             <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
  15.         </repository>
  16.     </repositories>

第三步 协议开发

  • 新建packag：org.jetlinks.demo.protocol

  • 创建协议编码解码类：DemoDeviceMessageCodec

  1.   import io.netty.buffer.ByteBuf;
  2.   import io.netty.buffer.ByteBufUtil;
  3.   import io.netty.buffer.Unpooled;
  4.   import lombok.AllArgsConstructor;
  5.   import lombok.extern.slf4j.Slf4j;
  6.   import org.apache.commons.codec.DecoderException;
  7.   import org.apache.commons.codec.binary.Hex;
  8.   import org.jetlinks.core.message.DeviceMessage;
  9.   import org.jetlinks.core.message.DeviceOnlineMessage;
  10.   import org.jetlinks.core.message.Message;
  11.   import org.jetlinks.core.message.codec.*;
  12.   import org.jetlinks.core.message.function.FunctionInvokeMessage;
  13.   import org.jetlinks.core.message.function.FunctionInvokeMessageReply;
  14.   import org.jetlinks.core.server.session.DeviceSession;
  15.   import org.jetlinks.wt.protocol.message.NbIotMessage;
  16.   import org.jetlinks.wt.protocol.message.data.enums.DataIdEnum;
  17.   import org.jetlinks.wt.protocol.message.enums.ControlEnum;
  18.   import org.reactivestreams.Publisher;
  19.   import reactor.core.publisher.Flux;
  20.   import reactor.core.publisher.Mono;
  22.   @AllArgsConstructor
  23.   @Slf4j
  24.   public class DemoDeviceMessageCodec implements DeviceMessageCodec {
  25.       // 传输协议定义
  26.       @Override
  27.       public Transport getSupportTransport() {
  28.           return DefaultTransport.TCP;
  29.       }
  31.       // 把tcp消息解码为平台消息，多用于设备上报消息到平台
  32.       @Override
  33.       public Mono<? extends Message> decode(MessageDecodeContext context) {
  34.           return Mono.empty();
  35.       }
  37.       // 把平台消息编码为协议传输消息，多用于平台命令下发到设备
  38.       @Override
  39.       public Publisher<? extends EncodedMessage> encode(MessageEncodeContext context) {
  40.           retrun Mono.empty();
  41.       }
  42.   }
  • 创建协议入口类： DemoProtocolSupportProvider
  1.   import org.jetlinks.core.ProtocolSupport;
  2.   import org.jetlinks.core.Value;
  3.   import org.jetlinks.core.defaults.CompositeProtocolSupport;
  4.   import org.jetlinks.core.device.AuthenticationResponse;
  5.   import org.jetlinks.core.device.DeviceRegistry;
  6.   import org.jetlinks.core.device.MqttAuthenticationRequest;
  7.   import org.jetlinks.core.message.codec.DefaultTransport;
  8.   import org.jetlinks.core.metadata.DefaultConfigMetadata;
  9.   import org.jetlinks.core.metadata.types.PasswordType;
  10.   import org.jetlinks.core.metadata.types.StringType;
  11.   import org.jetlinks.core.spi.ProtocolSupportProvider;
  12.   import org.jetlinks.core.spi.ServiceContext;
  13.   import org.jetlinks.demo.protocol.tcp.DemoTcpMessageCodec;
  14.   import org.jetlinks.supports.official.JetLinksDeviceMetadataCodec;
  15.   import reactor.core.publisher.Mono;
  17.   public class DemoProtocolSupportProvider implements ProtocolSupportProvider {
  19.       @Override
  20.       public Mono<? extends ProtocolSupport> create(ServiceContext context) {
  21.           CompositeProtocolSupport support = new CompositeProtocolSupport();
  22.           // 协议ID
  23.           support.setId("demo-v1");
  24.           // 协议名称
  25.           support.setName("演示协议v1");
  26.           // 协议说明
  27.           support.setDescription("演示协议");
  28.           // 物模型编解码，固定为JetLinksDeviceMetadataCodec
  29.           support.setMetadataCodec(new JetLinksDeviceMetadataCodec());
  30.           //TCP消息编解码器
  31.           DemoDeviceMessageCodec codec = new DemoDeviceMessageCodec();
  32.           // 两个参数，协议支持和编解码类DemoDeviceMessageCodec中保持一致，第二个参数定义使用的编码解码类
  33.           support.addMessageCodecSupport(DefaultTransport.TCP, () -> Mono.just(codec));
  34.           return Mono.just(support);
  35.       }
  36.   }

第四步 设备上报消息解码

  1. @AllArgsConstructor
  2. @Slf4j
  3. public class DemoTcpMessageCodec implements DeviceMessageCodec {
  5.        ....
  7.         // 把tcp消息解码为平台消息，多用于设备上报消息到平台
  8.         @Override
  9.         public Publisher<? extends Message> decode(MessageDecodeContext context) {
  10.             return Flux.defer(() -> {
  11.                 // 消息上下文
  12.                 FromDeviceMessageContext ctx = ((FromDeviceMessageContext) context);
  13.                 // 从上下文中获取消息字节数组
  14.                 ByteBuf byteBuf = context.getMessage().getPayload();
  15.                 byte[] payload = ByteBufUtil.getBytes(byteBuf, 0, byteBuf.readableBytes(), false);
  16.                 // 把字节流转换为字符串，根据不同设备不同协议进行解析，
  17.                 String text=new String(payload);
  18.                 ReportPropertyMessage message = new ReportPropertyMessage();
  19.                 // 设置消息ID为我们获得的消息内容
  20.                 message.setDeviceId(text);
  21.                 // 以当前时间戳为消息时间
  22.                 long time= LocalDateTime.now().toInstant(ZoneOffset.of("+8")).toEpochMilli();
  23.                 message.setTimestamp(time);
  24.                 // 构造上报属性
  25.                 Map<String, Object> properties = new HashMap<>();
  26.                 properties.put("text",text);
  27.                 // 设置上报属性
  28.                 message.setProperties(properties);
  30.                 // 获取设备会话信息
  31.                 DeviceSession session = ctx.getSession();
  32.                 // 如果session中没有设备信息，则为设备首次上线
  33.                 if (session.getOperator() == null) {
  34.                     DeviceOnlineMessage onlineMessage = new DeviceOnlineMessage();
  35.                     onlineMessage.setDeviceId(text);
  36.                     onlineMessage.setTimestamp(System.currentTimeMillis());
  37.                     // 返回到平台上线消息
  38.                     return Flux.just(message,onlineMessage);
  39.                 }
  40.                 // 返回到平台属性上报消息
  41.                 return Mono.just(message);
  42.             });
  43.         }
  44. .....
  46.   }

第五步 平台发送消息到设备（消息编码）

  1. @AllArgsConstructor
  2. @Slf4j
  3. public class DemoTcpMessageCodec implements DeviceMessageCodec {
  4.     ..........
  5.     // 把平台消息编码为协议传输消息，多用于平台命令下发到设备
  6.     @Override
  7.     public Publisher<? extends EncodedMessage> encode(MessageEncodeContext context) {
  8.          // 从平台消息上下文中获取消息内容
  9.         CommonDeviceMessage message = (CommonDeviceMessage) context.getMessage();
  10.         EncodedMessage encodedMessage = EncodedMessage.simple(Unpooled.wrappedBuffer(message.toString().getBytes()));
  11.         // 根据消息类型的不同，构造不同的消息
  12.         if (message instanceof ReadPropertyMessage) {
  13.             ReadPropertyMessage readPropertyMessage = (ReadPropertyMessage) message;
  14.             // 获取需要传输的字节
  15.             byte[] bytes = readPropertyMessage.toString().getBytes();
  16.             // 构造为平台传输到设备的消息体
  17.             encodedMessage = EncodedMessage.simple(Unpooled.wrappedBuffer(bytes));
  18.         }
  19.         return Mono.just(encodedMessage);
  20.     }
  21. }

第六步 打成jar包上传到平台，并进行测试

以TCP服务网关接入设备为例。

进行协议上传和 配置，并使用TCP工具进行测试。

注意

建议在编写协议包时，先用单元测试进行测试。最后再发布到平台进行模拟设备测试。

调试(debug)协议

将协议包工程放到和jetlinks相同到工程目录里,即可使用IDE进行debug.

参照目录:

  2. --jetlinks
  3. ----|--dev
  4. ----|---|-- demo-protocl    # 开发中的协议
  5. ----|--jetlinks-components 
  6. ----|--jetlinks-standalone
  7. ----|-- ....

注意

在调试过程中修改代码可以进行热加载,但是重启服务后会失效,需要重新发布协议.