基于surging的木舟平台如何分布式接入设备

小岭哥

一、概述

上篇文章介绍了木舟通过基于木舟平台浅谈surging 的热点KEY的解决方法，那么此篇文章将介绍基于surging的木舟平台如何分布式接入设备.
      木舟 (Kayak) 是什么?
       木舟(Kayak)是基于.NET6.0软件环境下的surging微服务引擎进行开发的, 平台包含了微服务和物联网平台。支持异步和响应式编程开发，功能包含了物模型,设备,产品,网络组件的统一管理和微服务平台下的注册中心，服务路由，模块，中间服务等管理。还有多协议适配(TCP,MQTT,UDP,CoAP,HTTP,Grpc,websocket,rtmp,httpflv,webservice,等),通过灵活多样的配置适配能够接入不同厂家不同协议等设备。并且通过设备告警,消息通知,数据可视化等功能。能够让你能快速建立起微服务物联网平台系统。
     那么下面就为大家介绍如何从创建组件、协议、设备网关，设备到设备网关接入，再到设备数据上报，把整个流程通过此篇文章进行阐述。
      木舟kayal 平台开源地址：https://github.com/microsurging/
      surging 微服务引擎开源地址：https://github.com/fanliang11/surging（后面surging 会移动到microsurging进行维护）
二、网络组件

1.编辑创建Tcp协议的网络组件，可以选择独立配置(独立配置是集群模式). 下图是解析方式选择了自定义脚本进行解码操作。

 选择了独立配置后(共享配置不会进行注册路由)，如果架设了新的网关实例，就会在注册中心networkroute/tcp路径下添加服务节点，以consul注册中心为例，打开：http://127.0.0.1:8500

 
三、自定义协议

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

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

1. public class Demo3ProtocolSupportProvider : ProtocolSupportProvider
2. {
3.      public override IObservable<ProtocolSupport> Create(ProtocolContext context)
4.      {
6.          var support = new ComplexProtocolSupport();
7.          support.Id = "demo_3";
8.          support.Name = "演示协议3";
9.          support.Description = "演示协议3";
10.          support.AddAuthenticator(MessageTransport.Tcp, new Demo5Authenticator());
11.          support.AddDocument(MessageTransport.Tcp, "Document/document-tcp.md");
12.          support.Script = "\r\nvar decode=function(buffer)\r\n{\r\n  parser.Fixed(5).Handler(\r\n function(buffer){ \r\n      var bytes = BytesUtils.GetBytes(buffer,1,4);\r\n      var len = BytesUtils.LeStrToInt(bytes,1,4);//2. 获取消息长度.\r\n       var buf = BytesUtils.Slice(buffer,0,5); \r\n parser.Fixed(len).Result(buf); \r\n        }).Handler(function(buffer){ parser.Result(buffer).Complete();  \r\n        }\r\n )\r\n}\r\nvar encode=function(buffer)\r\n{\r\n}";      
13.          support.AddMessageCodecSupport(MessageTransport.Tcp, () => Observable.Return(new ScriptDeviceMessageCodec(support.Script)));
14.          support.AddConfigMetadata(MessageTransport.Tcp, _tcpConfig);
16.          support.AddAuthenticator(MessageTransport.Udp, new Demo5Authenticator());
17.          support.Script = "\r\nvar decode=function(buffer)\r\n{\r\n  parser.Fixed(5).Handler(\r\n function(buffer){ \r\n      var bytes = BytesUtils.GetBytes(buffer,1,4);\r\n      var len = BytesUtils.LeStrToInt(bytes,1,4);//2. 获取消息长度.\r\n       var buf = BytesUtils.Slice(buffer,0,5); \r\n parser.Fixed(len).Result(buf); \r\n        }).Handler(function(buffer){ parser.Result(buffer).Complete();  \r\n        }\r\n )\r\n}\r\nvar encode=function(buffer)\r\n{\r\n}";
18.          support.AddMessageCodecSupport(MessageTransport.Udp, () => Observable.Return(new ScriptDeviceMessageCodec(support.Script)));
19.          support.AddConfigMetadata(MessageTransport.Udp, _udpConfig);
20.          return Observable.Return(support);
21.      }
22. }

复制代码

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

1. ### 认证说明
3. CONNECT报文:
4. ```text
5. clientId: 设备ID
6. password: md5(timestamp+"|"+secureKey)
7. ```

复制代码

 
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.添加消息编解码代码  support.AddMessageCodecSupport(MessageTransport.Tcp, () => Observable.Return(new ScriptDeviceMessageCodec(support.Script)));， 可以自定义编解码，ScriptDeviceMessageCodec代码如下：
 

[code]using DotNetty.Buffers;using Jint;using Jint.Parser;using Microsoft.CodeAnalysis.Scripting;using Microsoft.Extensions.Logging;using RulesEngine.Models;using Surging.Core.CPlatform.Codecs.Core;using Surging.Core.CPlatform.Utilities;using Surging.Core.DeviceGateway.Runtime.Device.Message;using Surging.Core.DeviceGateway.Runtime.Device.Message.Event;using Surging.Core.DeviceGateway.Runtime.Device.Message.Property;using Surging.Core.DeviceGateway.Runtime.Device.MessageCodec;using Surging.Core.DeviceGateway.Runtime.RuleParser.Implementation;using Surging.Core.DeviceGateway.Utilities;using System;using System.Collections.Generic;using System.Linq;using System.Reactive.Linq;using System.Reactive.Subjects;using System.Runtime;using System.Text;using System.Text.Json;using System.Text.RegularExpressions;using System.Threading.Tasks;namespace Surging.Core.DeviceGateway.Runtime.Device.Implementation{    public class ScriptDeviceMessageCodec : DeviceMessageCodec    {        public string GlobalVariable { get; private set; }        public string EncoderScript { get; private set; }        public string DecoderScript { get; private set; }        public IObservable _rulePipePayload;        private readonly ILogger _logger;         public ScriptDeviceMessageCodec(string script) {            _logger = ServiceLocator.GetService();            RegexOptions options = RegexOptions.Singleline | RegexOptions.IgnoreCase;            string matchStr = Regex.Match(script, @"var\s*[\w$]*\s*\=.*function.*\(.*\)\s*\{[\s\S]*\}.*?v", options).Value;            if (!string.IsNullOrEmpty(matchStr))            {                DecoderScript = matchStr.TrimEnd('v');                DecoderScript= Regex.Replace(DecoderScript, @"var\s*[\w$]*\s*\=[.\r|\n|\t|\s]*?(function)\s*\([\w$]*\s*\)\s*\{", "", RegexOptions.IgnoreCase);                DecoderScript= DecoderScript.Slice(0, DecoderScript.LastIndexOf('}'));                EncoderScript = script.Replace(DecoderScript, "");                            }             var matchStr1 = Regex.Matches(script, @"(?