- A+
所属分类：.NET技术
前言
Surging 发展已经有快6年的时间，经过这些年的发展，功能框架也趋于成熟，但是针对于商业化需求还需要不断的打磨，前段时间客户找到我想升级成平台化，针对他的需求我们讨论了很久，他的一些需求想法还是非常先进的，针对于现在surging 还有很多需要研发打磨，后面因为时间成本的关系我拒绝了他，我也打算后面慢慢的将surging升级为平台化以便支持物联网，流媒体等业务场景，可以通过统一定义，统一管理来配置构建各种业务场景，而web网关，设备网关，流媒体网关等各种类型的网关可以通过配置添加删除网络协议组件，以便可以支持多种协议适配，设配管理和配置规则引擎等业务场景。 今天所要讲的是以TCP组件为范本如何进行优化构建成平台化多协议适配。
构建TCP组件
首先来看看如何创建配置启动协议组件，以TCP协议组件为例，以下代码用于创建具有配置属性的TCP网络组件，并配置规则引擎脚本以解码消息
var config =new Dictionary<string, object>(); config.Add("script", @"parser.Fixed(4).Handler( function(buffer){ var buf = BytesUtils.Slice(buffer,1,4); parser.Fixed(buffer.ReadableBytes).Result(buf); }).Handler( function(buffer){parser.Fixed(8).Result(buffer);} ).Handler(function(buffer){ parser.Result('处理完成','gb2312').Complete(); } )"); var network= serviceProvider.ServiceProvoider.GetInstances<INetworkProvider<TcpServerProperties>>().CreateNetwork(new TcpServerProperties { ParserType = PayloadParserType.Script, PayloadType = PayloadType.String, Host = "127.0.0.1", Port = 322, ParserConfiguration = config }); network.StartAsync();
以下代码用于创建具有配置属性的TCP网络组件，并配置direct解码消息
var network1 = serviceProvider.ServiceProvoider.GetInstances<INetworkProvider<TcpServerProperties>>().CreateNetwork(new TcpServerProperties { ParserType = PayloadParserType.Direct, PayloadType = PayloadType.String, Host = "127.0.0.1", Port = 321 }); network1.StartAsync();
还有两种类型的payload parser：fixedlength和delimited，这里就不依次说明了，直接通过添加更改ParserType属性就能配置相关的parse
创建Tcp服务
public class TcpService : TcpBehavior, ITcpService { private readonly IDeviceProvider _deviceProvider; public TcpService(IDeviceProvider deviceProvider) { _deviceProvider = deviceProvider; } public override void Load(string clientId,TcpServerProperties tcpServerProperties) { var deviceStatus =_deviceProvider.IsConnected(clientId); this.Parser.HandlePayload().Subscribe(buffer=>ParserBuffer(buffer)); } public override void DeviceStatusProcess(DeviceStatus status, string clientId, TcpServerProperties tcpServerProperties) { //throw new NotImplementedException(); } public async Task ParserBuffer(IByteBuffer buffer) { List<string> result = new List<string>(); while (buffer.ReadableBytes > 0) { result.Add(buffer.ReadString(this.Parser.GetNextFixedRecordLength(), Encoding.GetEncoding("gb2312"))); } // var str= buffer.ReadString(buffer.ReadableBytes, Encoding.UTF8); var byteBuffer= Unpooled.Buffer(); byteBuffer.WriteString("rn", Encoding.UTF8); byteBuffer.WriteString("处理完成", Encoding.GetEncoding("gb2312")); await Sender.SendAndFlushAsync(byteBuffer); buffer.Release(); // await Sender.SendAndFlushAsync("消息已接收",Encoding.GetEncoding("gb2312")); this.Parser.Close(); } }
测试
规则引擎脚本解码测试
direct 解码测试
总结
最近忙着对于rtsp 进行bug 的修复和测试， 这样流媒体业务场景除了httpflv,rtmp 协议外，也能针对于rtsp协议的支持，协议之间互相也能进行转发， 对于rtsp 完成后，surging 将转继续优化构建平台化，年底也会陆续推出后台系统，基于Node-Red 可视化数据流的服务规则引擎编排。
