基于surging 的木舟平台如何通过HTTP网络组件接入设备

李志达

一、概述

      上篇文章介绍了木舟如何上传模块热部署，那么此篇文章将介绍如何利用HTTP网络组件接入设备，那么有些人会问木舟又是什么，是什么架构为基础，能做什么呢？
       木舟 (Kayak) 是什么?
       木舟(Kayak)是基于.NET6.0软件环境下的surging微服务引擎进行开发的, 平台包含了微服务和物联网平台。支持异步和响应式编程开发，功能包含了物模型,设备,产品,网络组件的统一管理和微服务平台下的注册中心，服务路由，模块，中间服务等管理。还有多协议适配(TCP,MQTT,UDP,CoAP,HTTP,Grpc,websocket,rtmp,httpflv,webservice,等),通过灵活多样的配置适配能够接入不同厂家不同协议等设备。并且通过设备告警,消息通知,数据可视化等功能。能够让你能快速建立起微服务物联网平台系统。
     那么下面就为大家介绍如何从创建组件、协议、设备网关，设备到设备网关接入，再到设备数据上报，把整个流程通过此篇文章进行阐述。
二、网络组件

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

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

 又或者是访问一下中间服务，以上篇文章上传的Testapi 模块为例：

 三、自定义协议

• 如何创建自定义协议模块
  

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

1.   public class Demo5ProtocolSupportProvider : ProtocolSupportProvider
2.     {
3.         public override IObservable<ProtocolSupport> Create(ProtocolContext context)
4.         {<br>   　　       var support = new ComplexProtocolSupport();<br>              support.Id = "demo5";<br>              support.Name = "演示协议5";<br>              support.Description = "演示协议5";
5.               support.AddDocument(MessageTransport.Http, "Document/document-http.md");
6.        　　　　support.AddAuthenticator(MessageTransport.Http, new Demo5Authenticator());
7.       　　　　 support.AddRoutes(MessageTransport.Http, new List<BasicMessageCodec>() {
8.          　　 BasicMessageCodec.DeviceOnline,
9.           　　 BasicMessageCodec.ReportProperty,
10.            　　BasicMessageCodec.WriteProperty,
11.             　　BasicMessageCodec.ReadProperty,
12.               BasicMessageCodec.Event
13.        }.Select(p => HttpDescriptor.Instance(p.Pattern)
14.            .GroupName(p.Route.GroupName())
15.            .HttpMethod(p.Route.HttpMethod())
16.            .Path(p.Pattern)
17.            .ContentType(MediaType.ToString(MediaType.ApplicationJson))
18.            .Description(p.Route.Description())
19.            .Example(p.Route.Example())
20.            ).ToList());
21.        support.AddMessageCodecSupport(MessageTransport.Http, () => Observable.Return(new HttpDeviceMessageCodec()));
22.        support.AddConfigMetadata(MessageTransport.Http, _httpConfig);
23.        return Observable.Return(support);
24.                
25.         }
26.      
27.      }               

复制代码

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

1. ### 使用HTTP推送设备数据
3. 上报属性例子:
5. POST /{productId}/{deviceId}/properties/report
6. Authorization:{产品或者设备中配置的Token}
7. Content-Type: application/json
9. {
10. "properties":{
11.    "temp":11.5
12. }
13. }
15. 上报事件例子:
17. POST /{productId}/{deviceId}/event/{eventId}
18. Authorization:{产品或者设备中配置的Token}
19. Content-Type: application/json
21. {
22. "data":{
23.    "createtime": ""
24. }
25. }

复制代码

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

 

1.        public class Demo5Authenticator : IAuthenticator
2.        {
3.            public IObservable<AuthenticationResult> Authenticate(IAuthenticationRequest request, IDeviceOperator deviceOperator)
4.            {
5.                var result = Observable.Return<AuthenticationResult>(default);
6.                if (request is DefaultAuthRequest)
7.                {
8.                    var authRequest = request as DefaultAuthRequest;
9.                    deviceOperator.GetConfig(authRequest.GetTransport()==MessageTransport.Http?"token": "key").Subscribe(  config =>
10.                    {
11.                        var password = config.Convert<string>();
12.                        if (authRequest.Password.Equals(password))
13.                        {
14.                            result= result.Publish(AuthenticationResult.Success(authRequest.DeviceId));
15.                        }
16.                        else
17.                        {
18.                            result= result.Publish(AuthenticationResult.Failure(StatusCode.CUSTOM_ERROR, "验证失败,密码错误"));
19.                        }
20.                    });
21.                }
22.                else
23.                result = Observable.Return<AuthenticationResult>(AuthenticationResult.Failure(StatusCode.CUSTOM_ERROR, "不支持请求参数类型"));
24.                return result;
25.            }
27.            public IObservable<AuthenticationResult> Authenticate(IAuthenticationRequest request, IDeviceRegistry registry)
28.            {
29.                var result = Observable.Return<AuthenticationResult>(default);
30.                var authRequest = request as DefaultAuthRequest;
31.                registry
32.                  .GetDevice(authRequest.DeviceId)
33.                  .Subscribe(async p => {
35.                     var config=  await p.GetConfig(authRequest.GetTransport() == MessageTransport.Http ? "token" : "key");
36.                      var password= config.Convert<string>();
37.                     if(authRequest.Password.Equals(password))
38.                      {
39.                          result= result.Publish(AuthenticationResult.Success(authRequest.DeviceId));
40.                      }
41.                      else
42.                      {
43.                          result= result.Publish(AuthenticationResult.Failure(StatusCode.CUSTOM_ERROR, "验证失败,密码错误"));
44.                      }
45.                  });
46.                return result;
47.            }
48.        }

复制代码

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

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

复制代码

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

1.   public class HttpDeviceMessageCodec : DeviceMessageCodec
2.   {
3.       private readonly MessageTransport _transport;
5.       public HttpDeviceMessageCodec() : this(MessageTransport.Http)
6.       {
7.       }
9.       private static DefaultHttpResponseMessage Unauthorized(String msg)
10.       {
11.           return new DefaultHttpResponseMessage()
12.                   .ContentType(MediaType.ApplicationJson)
13.                   .Body("{"success":false,"code":"unauthorized","message":"" + msg + ""}")
14.                   .Status(HttpStatus.AuthorizationFailed);
15.       }
17.       private static DefaultHttpResponseMessage BadRequest()
18.       {
19.           return new DefaultHttpResponseMessage()
20.                   .ContentType(MediaType.ApplicationJson)
21.                   .Body("{"success":false,"code":"bad_request"}")
22.                   .Status(HttpStatus.RequestError);
23.       }
25.       public HttpDeviceMessageCodec(MessageTransport transport)
26.       {
27.           _transport = transport;
28.       }
29.       public override IObservable<IDeviceMessage> Decode(MessageDecodeContext context)
30.       {
31.           if (context.GetMessage() is HttpRequestMessage)
32.           {
33.               return DecodeHttpRequestMessage(context);
34.           }
35.           return Observable.Return<IDeviceMessage>(default);
36.       }
39.       public override  IObservable<IEncodedMessage> Encode(MessageEncodeContext context)
40.       {
41.           return Observable.Return<IEncodedMessage>(default);
42.       }
46.       private IObservable<IDeviceMessage> DecodeHttpRequestMessage(MessageDecodeContext context)
47.       {
48.           var result = Observable.Return<IDeviceMessage>(default);
49.           var message = (HttpExchangeMessage)context.GetMessage();
51.           Header? header = message.Request.GetHeader("Authorization");
52.           if (header == null || header.Value == null || header.Value.Length == 0)
53.           {
54.               message
55.                    .Response(Unauthorized("Authorization header is required")).ToObservable()
56.                    .Subscribe(p => result = result.Publish(default));
58.               return result;
59.           }
60.           var httpToken = header.Value[0];
62.           var paths = message.Path.Split("/");
63.           if (paths.Length == 0)
64.           {
65.               message.Response(BadRequest()).ToObservable()
66.                  .Subscribe(p => result = result.Publish(default));
67.               return result;
68.           }
69.           String deviceId = paths[1];
70.           context.GetDevice(deviceId).Subscribe(async deviceOperator =>
71.           {
72.               var config = deviceOperator==null?null: await deviceOperator.GetConfig("token");
73.               var token = config?.Convert<string>();
74.               if (token == null || !httpToken.Equals(token))
75.               {
76.                   await message
77.                        .Response(Unauthorized("Device not registered or authentication failed"));
78.               }
79.               else
80.               {
81.                   var deviceMessage = await DecodeBody(message, deviceId);
82.                   if (deviceMessage != null)
83.                   {
84.                       await message.Success("{"success":true,"code":"success"}");
85.                       result = result.Publish(deviceMessage);
86.                   }
87.                   else
88.                   {
89.                       await message.Response(BadRequest());
90.                   }
91.               }
92.           });
93.           return result;
94.       }
96.       private async Task<IDeviceMessage> DecodeBody(HttpExchangeMessage message,string deviceId)
97.       {
99.           byte[] body = new byte[message.Payload.ReadableBytes];
100.           message.Payload.ReadBytes(body);
101.           var deviceMessage = await TopicMessageCodec.Dodecode(message.Path, body);
102.           deviceMessage.DeviceId = deviceId;
103.           return deviceMessage;
104.       }
105.   }

复制代码

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

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

复制代码

• 如何加载协议模块，协议模块包含了协议模块支持添加引用加载和上传热部署加载。                           
  

   引用加载模块

 上传热部署协议模块

 四、设备网关

创建设备网关

 五、产品管理

以下是添加产品。

 设备接入

 六、设备管理

添加设备

 HTTP认证配置

 创建告警阈值

 七、测试

 利用Postman 进行测试，以调用http://127.0.0.1:168/{productid}/{deviceid}/properties/report 为例，Authorization设置：123456
1.正常数据测试

 
 
 2. 如果是选用Get方式调用，会因为找不到ServiceRoute而返回错误。

 3. 把Authorization改成1111，会返回错误Device not registered or authentication failed，从而上报数据失败

 以上上传的数据可以在设备信息-》运行状态中查看

 告警信息可以在超临界数据中查看

  七、总结

 以上是基于HTTP网络组件设备接入，现有平台网络组件可以支持TCP,MQTT,UDP,CoAP,HTTP,Grpc,websocket,rtmp,httpflv,webservice,tcpclient, 而设备接入支持TCP，UDP，HTTP网络组件，后面会陆续添加支持所有网络组件接入，后面我也会陆续介绍其它网路组件设备接入 ，  然后定于11月20日发布1.0测试版平台。也请大家到时候关注捧场。
 

来源:https://www.cnblogs.com/fanliang11/p/18527947
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容，谢谢合作！