线上测试木舟物联网平台之如何通过HTTP网络组件接入设备

news/2025/3/12 8:40:09/文章来源:https://www.cnblogs.com/fanliang11/p/18766635

一、概述

木舟 (Kayak) 是什么?

木舟(Kayak)是基于.NET6.0软件环境下的surging微服务引擎进行开发的, 平台包含了微服务和物联网平台。支持异步和响应式编程开发，功能包含了物模型,设备,产品,网络组件的统一管理和微服务平台下的注册中心，服务路由，模块，中间服务等管理。还有多协议适配(TCP,MQTT,UDP,CoAP,HTTP,Grpc,websocket,rtmp,httpflv,webservice,等),通过灵活多样的配置适配能够接入不同厂家不同协议等设备。并且通过设备告警,消息通知,数据可视化等功能。能够让你能快速建立起微服务物联网平台系统。

那么下面就为大家介绍如何从创建组件、协议、设备网关，设备到设备网关接入，再到设备数据上报，把整个流程通过此篇文章进行阐述。

木舟物联网平台：http://117.72.121.2:3100

surging 微服务引擎开源地址：https://github.com/fanliang11/surging（后面surging 会移动到microsurging进行维护）

二、木舟物联网平台管理平台界面

用户名：fanly 密码：123456

三、链路跟踪监控

地址：http://117.72.121.2:8080/ （这是社区版，仅支持v6.0 ，企业版本支持v8.0）

四、网络组件

1.编辑创建HTTP协议的网络组件，可以选择共享配置和独立配置(独立配置是集群模式)，然后可以选择开启swagger和webservice.

开启成功后，可以看看swagger 是否可以访问

地址：http://117.72.121.2:281/swagger/index.html

开启成功后，可以看看webservice是否可以访问

webservice:http://117.72.121.2:168/devicedata/devicedata/changedevicestage.asmx?servicekey=WebService

五、自定义协议

如何创建自定义协议模块

如果是网络编程开发，必然会涉及到协议报文的编码解码处理，那么对于平台也是做到了灵活处理，首先是协议模块创建，通过以下代码看出协议模块可以添加协议说明md文档, 身份鉴权处理，HTTP路由,消息编解码，元数据配置。下面一一介绍如何进行编写

  public class Demo5ProtocolSupportProvider : ProtocolSupportProvider{public override IObservable<ProtocolSupport> Create(ProtocolContext context){
   　　       var support = new ComplexProtocolSupport();
              support.Id = "demo5";
              support.Name = "演示协议5";
              support.Description = "演示协议5";support.AddDocument(MessageTransport.Http, "Document/document-http.md");support.AddAuthenticator(MessageTransport.Http, new Demo5Authenticator()); support.AddRoutes(MessageTransport.Http, new List<BasicMessageCodec>() {BasicMessageCodec.DeviceOnline,BasicMessageCodec.ReportProperty,BasicMessageCodec.WriteProperty,BasicMessageCodec.ReadProperty,BasicMessageCodec.Event}.Select(p => HttpDescriptor.Instance(p.Pattern).GroupName(p.Route.GroupName()).HttpMethod(p.Route.HttpMethod()).Path(p.Pattern).ContentType(MediaType.ToString(MediaType.ApplicationJson)).Description(p.Route.Description()).Example(p.Route.Example())).ToList());support.AddMessageCodecSupport(MessageTransport.Http, () => Observable.Return(new HttpDeviceMessageCodec()));support.AddConfigMetadata(MessageTransport.Http, _httpConfig);return Observable.Return(support);}}

1. 添加协议说明文档如代码: support.AddDocument(MessageTransport.Http, "Document/document-http.md");，文档仅支持 markdown文件，如下所示

### 使用HTTP推送设备数据上报属性例子: POST /{productId}/{deviceId}/properties/report
Authorization:{产品或者设备中配置的Token}
Content-Type: application/json{"properties":{"temp":11.5}
}上报事件例子:POST /{productId}/{deviceId}/event/{eventId}
Authorization:{产品或者设备中配置的Token}
Content-Type: application/json{"data":{"createtime": ""}
}

2. 添加身份鉴权如代码: support.AddAuthenticator(MessageTransport.Http, new Demo5Authenticator()) ，自定义身份鉴权Demo5Authenticator 代码如下：

       public class Demo5Authenticator : IAuthenticator{public IObservable<AuthenticationResult> Authenticate(IAuthenticationRequest request, IDeviceOperator deviceOperator){var result = Observable.Return<AuthenticationResult>(default);if (request is DefaultAuthRequest){var authRequest = request as DefaultAuthRequest;deviceOperator.GetConfig(authRequest.GetTransport()==MessageTransport.Http?"token": "key").Subscribe(  config =>{var password = config.Convert<string>();if (authRequest.Password.Equals(password)){result= result.Publish(AuthenticationResult.Success(authRequest.DeviceId));}else{result= result.Publish(AuthenticationResult.Failure(StatusCode.CUSTOM_ERROR, "验证失败,密码错误"));}});}elseresult = Observable.Return<AuthenticationResult>(AuthenticationResult.Failure(StatusCode.CUSTOM_ERROR, "不支持请求参数类型"));return result;}public IObservable<AuthenticationResult> Authenticate(IAuthenticationRequest request, IDeviceRegistry registry){var result = Observable.Return<AuthenticationResult>(default);var authRequest = request as DefaultAuthRequest;registry.GetDevice(authRequest.DeviceId).Subscribe(async p => {var config=  await p.GetConfig(authRequest.GetTransport() == MessageTransport.Http ? "token" : "key");var password= config.Convert<string>();if(authRequest.Password.Equals(password)){result= result.Publish(AuthenticationResult.Success(authRequest.DeviceId));}else{result= result.Publish(AuthenticationResult.Failure(StatusCode.CUSTOM_ERROR, "验证失败,密码错误"));}});return result;}}

3. 添加Http路由代码support.AddRoutes，那么如何配置呢，代码如下：

    public static BasicMessageCodec ReportProperty =>new BasicMessageCodec("/*/properties/report", typeof(ReadPropertyMessage), route => route.GroupName("属性上报").HttpMethod("Post").Description("上报物模型属性数据").Example("{\"properties\":{\"属性ID\":\"属性值\"}}"));

4.添加消息编解码代码 support.AddMessageCodecSupport(MessageTransport.Http, () => Observable.Return(new HttpDeviceMessageCodec()))，可以自定义编解码，HttpDeviceMessageCodec代码如下：

  public class HttpDeviceMessageCodec : DeviceMessageCodec{private readonly MessageTransport _transport;public HttpDeviceMessageCodec() : this(MessageTransport.Http){}private static DefaultHttpResponseMessage Unauthorized(String msg){return new DefaultHttpResponseMessage().ContentType(MediaType.ApplicationJson).Body("{\"success\":false,\"code\":\"unauthorized\",\"message\":\"" + msg + "\"}").Status(HttpStatus.AuthorizationFailed);}private static DefaultHttpResponseMessage BadRequest(){return new DefaultHttpResponseMessage().ContentType(MediaType.ApplicationJson).Body("{\"success\":false,\"code\":\"bad_request\"}").Status(HttpStatus.RequestError);}public HttpDeviceMessageCodec(MessageTransport transport){_transport = transport;}public override IObservable<IDeviceMessage> Decode(MessageDecodeContext context){if (context.GetMessage() is HttpRequestMessage){return DecodeHttpRequestMessage(context);}return Observable.Return<IDeviceMessage>(default);}public override  IObservable<IEncodedMessage> Encode(MessageEncodeContext context){return Observable.Return<IEncodedMessage>(default);}private IObservable<IDeviceMessage> DecodeHttpRequestMessage(MessageDecodeContext context){var result = Observable.Return<IDeviceMessage>(default);var message = (HttpExchangeMessage)context.GetMessage();Header? header = message.Request.GetHeader("Authorization");if (header == null || header.Value == null || header.Value.Length == 0){message.Response(Unauthorized("Authorization header is required")).ToObservable().Subscribe(p => result = result.Publish(default));return result;}var httpToken = header.Value[0];var paths = message.Path.Split("/");if (paths.Length == 0){message.Response(BadRequest()).ToObservable().Subscribe(p => result = result.Publish(default));return result;}String deviceId = paths[1];context.GetDevice(deviceId).Subscribe(async deviceOperator =>{var config = deviceOperator==null?null: await deviceOperator.GetConfig("token");var token = config?.Convert<string>();if (token == null || !httpToken.Equals(token)){await message.Response(Unauthorized("Device not registered or authentication failed"));}else{var deviceMessage = await DecodeBody(message, deviceId);if (deviceMessage != null){await message.Success("{\"success\":true,\"code\":\"success\"}");result = result.Publish(deviceMessage);}else{await message.Response(BadRequest());}}});return result;}private async Task<IDeviceMessage> DecodeBody(HttpExchangeMessage message,string deviceId){byte[] body = new byte[message.Payload.ReadableBytes];message.Payload.ReadBytes(body);var deviceMessage = await TopicMessageCodec.Dodecode(message.Path, body);deviceMessage.DeviceId = deviceId;return deviceMessage;}}

5.添加元数据配置代码 support.AddConfigMetadata(MessageTransport.Http, _httpConfig); _httpConfig代码如下

        private readonly DefaultConfigMetadata _httpConfig = new DefaultConfigMetadata("Http认证配置", "token为http认证令牌").Add("token", "token", "http令牌", StringType.Instance);