平台从第三方或者设备主动拉取数据

场景：设备无法推送数据到平台,需要平台主动去调用第三方平台接口或者去调用设备的tcp服务等。

流程

  1. 通过实现自定义协议的DeviceStateChecker来自定义处理设备状态获取逻辑,比如通过调用第三方平台获取设备信息.
  2. 通过实现自定义协议的DeviceMessageSenderInterceptor.afterSent来拦截消息发送,替换掉默认处理方式.在这里使用WebClient或者Vertx请求第三方或者设备.
  3. 请求后解析数据为对应的消息,调用DecodedClientMessageHandler.handleMessage(device,message)完成默认消息处理之后,返回消息.

例子一,通过http到第三方平台获取数据.

第一步定义消息编码解码器

  1. public class HttpMessageCodec implements DeviceMessageCodec {
  3.     // 定义一个通用的响应，用于收到请求后响应
  4.     private static final SimpleHttpResponseMessage response = SimpleHttpResponseMessage
  5.             .builder()
  6.             .payload(Unpooled.wrappedBuffer("{success:true}".getBytes()))
  7.             .contentType(MediaType.APPLICATION_JSON)
  8.             .status(200)
  9.             .build();
  11.     @Override
  12.     public Transport getSupportTransport() {
  13.         return DefaultTransport.HTTP;
  14.     }
  16.     @Nonnull
  17.     @Override
  18.     public Publisher<? extends Message> decode(@Nonnull MessageDecodeContext context){
  19.         // 这里用于别的平台请求/通知jetlinks的请求处理
  20.         // 把消息转换为http消息
  21.          HttpExchangeMessage message = (HttpExchangeMessage) context.getMessage();
  22.         String url = message.getUrl();
  23.         // 这里通常需要判断是不是自己需要的请求，如果不是直接返回/响应，防止非法请求
  24.         if (!url.endsWith("/eventRcv")) {
  25.             return message.response(response).then(Mono.empty());
  26.         }
  27.         // 获取具体消息类型
  28.         ByteBuf payload = message.getPayload();
  29.         String string = payload.toString(StandardCharsets.UTF_8);
  30.         // 通常来说，云平台通知的定义为事件消息（也可以定义成别的消息）
  31.         EventMessage eventMessage = new EventMessage();
  32.         eventMessage.setEvent("test");
  33.         eventMessage.setDeviceId(string);
  34.         eventMessage.setData(string);
  35.         eventMessage.setMessageId(String.valueOf(System.currentTimeMillis()));
  36.         eventMessage.setTimestamp(System.currentTimeMillis());
  37.         return message.response(response).thenMany(Flux.just(eventMessage));
  38.     }
  41.     @Nonnull
  42.     @Override
  43.     public Publisher<? extends EncodedMessage> encode(@Nonnull MessageEncodeContext context) {
  44.         // 对接其他云平台，命令发起不在这里处理，所以这里返回空就可以了
  45.         return Mono.empty();
  46.     }
  48. }

第二步 定义一个消息拦截器

  1. @Slf4j
  2. @AllArgsConstructor
  3. @Getter
  4. @Setter
  5. public class HttpMessageSenderInterceptor implements DeviceMessageSenderInterceptor{
  6.     // 通过构造器注入一个编码消息处理器，用于消息的持久化
  7.     private DecodedClientMessageHandler handler;
  9.     private static final WebClient webclient=WebClient.builder().build();
  10.    /**
  11.      * 在消息发送后触发.
  12.      *
  13.      * @param device  设备操作接口
  14.      * @param message 源消息
  15.      * @param reply   回复的消息
  16.      * @param <R>     回复的消息类型
  17.      * @return 新的回复结果
  18.      */
  19.     public <R extends DeviceMessage> Flux<R> afterSent(DeviceOperator device, DeviceMessage message, Flux<R> reply) {
  20.         return Flux.from(
  21.             // 从配置中获取url等各种请求所需参数
  22.             device.getConfigs("url")
  23.                         .flatMap(values->{
  24.                             String url=values.getValue("url").map(Value::asString).orElse(null);
  25.                             // 通常发起请求都是通过方法调用
  26.                             FunctionInvokeMessage invokeMessage = (FunctionInvokeMessage) message;
  27.                             // 从命令发起的上下文中获取消息体
  28.                           List<FunctionParameter> inputs = invokeMessage.getInputs();
  29.                             Map< String, Object> body=iputs
  30.                                 .stream()
  31.                                 .collect(Collectors
  32.                                          .toMap(FunctionParameter::getName,             FunctionParameter::getValue));
  33.                             return webclient  // 构造WebClient
  34.                                 .post()  // 指定请求类型
  35.                                 .uri(url) // 请求路径
  36.                                 .bodyValue(body) // 请求参数
  37.                                    .retrieve() // 发起请求
  38.                                    .bodyToMono(String.class) // 响应参数
  39.                                    .flatMap(s -> {
  40.                                     // 响应参数包装为功能回复参数
  41.                                        FunctionInvokeMessageReply reply1 = new FunctionInvokeMessageReply();
  42.                                        reply1.setSuccess(true);
  43.                                        reply1.setMessage(s);
  44.                                        reply1.setDeviceId(message.getDeviceId());
  45.                                        reply1.setMessageId(message.getMessageId());
  46.                                        reply1.setTimestamp(System.currentTimeMillis());
  47.                                        reply1.setOutput(s);
  48.                                        reply1.setFunctionId(((FunctionInvokeMessage) message).getFunctionId());
  49.                                return Mono.just(reply1)
  50.                                        .map(deviceMessage->(R)deviceMessage);
  51.                            })
  52.                            // 消息持久化
  53.                            .flatMap(msg->handler.handleMessage(device,msg)
  54.                                    .thenReturn(msg));
  55.                         })
  56.                 );
  57.     }
  58. }

第三步 定义一个设备状态检测器

  1. /**
  2. * 这个接口会在进入设备详情页面和刷新设备状态时调用
  3. */
  4. @Slf4j
  5. public class HttpDeviceStateChecker implements DeviceStateChecker {
  6.     @Override
  7.     public @NotNull Mono<Byte> checkState(@NotNull DeviceOperator device) {
  8.         // 如果第三方平台有提供设备状态查询接口，则调用接口确定设备状态，否则设置为设备在线，方便发起功能或者属性查询
  9.         return Mono.just(DeviceState.online);
  10.     }
  11. }

第四步 定义协议处理器

  1. public class HttpProtocolSupportProvider implements ProtocolSupportProvider{
  3.         private static final DefaultConfigMetadata httpRequest = new DefaultConfigMetadata(
  4.             "Http请求配置"
  5.             , "")
  6.             .add("url", "url", " http请求地址", new StringType());
  8.     @Override
  9.     public Mono<? extends ProtocolSupport> create(ServiceContext serviceContext){
  10.         CompositeProtocolSupport support = new CompositeProtocolSupport();
  11.         support.setId("http-demo-v1");
  12.         support.setName("http调用第三方接口DEMO");
  13.         support.setDescription("http调用第三方接口DEMO");
  14.         support.setMetadataCodec(new JetLinksDeviceMetadataCodec());
  15.         // 设置一个编解码入口
  16.         HttpMessageCodec codec = new HttpMessageCodec();
  17.         support.addMessageCodecSupport(DefaultTransport.HTTP, () -> Mono.just(codec));
  18.         // 添加配置项定义
  19.         support.addConfigMetadata(DefaultTransport.HTTP, httpRequest);
  20.         HttpDeviceStateChecker httpDeviceStateChecker = new HttpDeviceStateChecker();
  21.         // 设置设备状态检查接口
  22.         support.setDeviceStateChecker(httpDeviceStateChecker);
  23.         // 设置HTTP消息拦截器，用于发送HTTP消息
  24.         serviceContext.getService(DecodedClientMessageHandler.class)
  25.                 .ifPresent(handler -> support.addMessageSenderInterceptor(new HttpMessageSenderInterceptor(handler)));
  26.         return Mono.just(support);
  27.     }
  28. }

例子二,通过tcp短链接从设备拉取数据.

TODO