基于surging的木舟IOT平台如何添加网络组件

室雅人和

  一 、 概述

          为了弥补代码的遗失，木舟IOT平台正在加班加点进行研发，后面不只是针对于IOT设备接入上报,告警，视频管理，组态数据可视化大屏，后面还会有快速搭建微服务平台，利用surging.cli工具根据数据库表生成微服务，中间服务，能让程序员快速完成BOSS交给的任务，从而在这个内卷的社会能占有一席之地。这些都是没有完成任务的空话，现在发此篇的目的是作者有能力开发出优秀的IOT平台，先介绍一个比较突出的功能，就是可以基于共享或者独立配置添加网络组件， 下面来介绍一下如何添加网络组件。
     一键运行打包成品下载：https://pan.baidu.com/s/11hcf9ieCkJxlGrzvIuxeQA?pwd=ajsr
    测试用户：fanly
    测试密码：123456
      为了让大家节约时间，能尽快运行产品看到效果，上面有 一键运行打包成品可以进行下载测试运行。
  二、如何测试运行

以下是目录结构,
IDE:consul 注册中心
kayak.client: 网关
kayak.server:微服务
apache-skywalking-apm：skywalking链路跟踪

 以上是目录结构，大家不需要一个个运行，只需要打开运行startup.bat,如果需要测试skywalking ，只需要apache-skywalking-apm\bin\startup.bat  文件就可以了,以下是运行的界面
三、如何添加组件

1.添加http服务组件，

打开平台界面，然后点击设备接入->网络组件，然后可以看到如下界面

 再点击新增组件或者编辑组件，完成后注意启动状态是关闭状态，此时并不能对于该组件功能进行访问调用，只有把启动状态打开，才能访问调用

 以上是http服务组件，启动完成后,如果设置了webservice和swagger,你可以访问webservice和swagger，看是否可以访问

 

2.添加/编辑Tcp服务组件

当添加/编辑Tcp组件时，设置Host:127.0.0.1 ,port:248并且还有解析方式选项，选项里面有不处理，固定长度，分隔符，自定义脚本，下面我们就来看自定义脚本

 添加脚本如下：

1. parser.Fixed(4).Handler(
2.                   function(buffer){
3.                     var buf = BytesUtils.Slice(buffer,1,4);
4.                     parser.Fixed(buffer.ReadableBytes).Result(buf);
5.              }).Handler(
6.                     function(buffer){parser.Fixed(8).Result(buffer);}
7.             ).Handler(function(buffer){
8.                     parser.Result('处理完成','gb2312').Complete();
9.                  }
10.              )

复制代码

而基于TCP服务代码如下，需要继承于TcpBehavior

1.     internal class TcpDeviceDataService : TcpBehavior, ITcpDeviceDataService
2.     {
3.         private readonly IDeviceProvider _deviceProvider;
4.         public TcpDeviceDataService(IDeviceProvider deviceProvider)
5.         {
6.             _deviceProvider = deviceProvider;
7.         }
9.         public override void Load(string clientId, NetworkProperties tcpServerProperties)
10.         {
11.             var deviceStatus = _deviceProvider.IsConnected(clientId);
12.             this.Parser.HandlePayload().Subscribe(async buffer => await ParserBuffer(buffer));
13.         }
15.         public override void DeviceStatusProcess(DeviceStatus status, string clientId, NetworkProperties tcpServerProperties)
16.         {
17.             //throw new NotImplementedException();
18.         }
20.         public async Task ParserBuffer(IByteBuffer buffer)
21.         {
22.             List<string> result = new List<string>();
23.             while (buffer.ReadableBytes > 0)
24.             {
25.                 result.Add(buffer.ReadString(this.Parser.GetNextFixedRecordLength(),
26.                     Encoding.GetEncoding("gb2312")));
27.             }
29.             // var str= buffer.ReadString(buffer.ReadableBytes, Encoding.UTF8);
31.             var byteBuffer = Unpooled.Buffer();
32.             byteBuffer.WriteString("\r\n", Encoding.UTF8);
33.             byteBuffer.WriteString("处理完成", Encoding.GetEncoding("gb2312"));
34.             await Sender.SendAndFlushAsync(byteBuffer);
35.             //  await Sender.SendAndFlushAsync("消息已接收",Encoding.GetEncoding("gb2312"));
36.             this.Parser.Close();
37.         }
39.         public Task<bool> ChangeDeviceStage(string deviceId)
40.         {
41.             throw new NotImplementedException();
42.         }
43.     }

复制代码

用测试Tcp调试工具结果如下

 
3.添加/编辑UDP服务组件

当添加/编辑UDP组件时， 设置Host:127.0.0.1 ,port:267 并且可以是否开启组播

 而基于udp服务代码如下，需要继承于UdpBehavior

1.   internal class UdpDeviceDataService : UdpBehavior, IUdpDeviceDataService
2.   {
3.       public Task<bool> ChangeDeviceStage(string deviceId)
4.       {
5.           throw new NotImplementedException();
6.       }
8.       public override async Task Dispatch(IEnumerable<byte> bytes)
9.       {
10.           await Sender.SendAndFlushAsync("\r\n", Encoding.UTF8);
11.           await Sender.SendAndFlushAsync("处理完成", Encoding.GetEncoding("gb2312"));
12.       }
13.   }

复制代码

测试结果如下：

4.添加/编辑WebSocket服务组件

当添加/编辑WebSocket组件时， 设置Host:127.0.0.1 ,port:55

  而基于websocket服务代码如下，需要继承于WSBehavior

1. internal class WSDeviceDataService : WSBehavior, IWSDeviceDataService
2. {
3.     protected override void OnMessage(MessageEventArgs e)
4.     {
5.         this.Client.Value.SendTo($"send:{e.Data},\r\n reply:hello,welcome to you!",ID);
6.     }
8.     protected override void OnOpen()
9.     {
10.      
11.     }
12. }

复制代码

测试结果如下：

5.添加/编辑UDP服务组件

当添加/编辑WebSocket组件时， 设置Host:127.0.0.1 ,port:345

 添加greet.proto文件，脚本如下：

1. syntax = "proto3";
3. package Greet;
5. service Greeter {
6.   // Sends a greeting
7.   rpc ChangeDeviceStage (DeviceRequest) returns (DeviceReply) {}
8. }
10. message DeviceRequest {
11.   string deviceId = 1;
12. }
14. message  DeviceReply {
15. bool message = 1;
16. }

复制代码

然后再创建GreeterBehavior，继承Greeter.GreeterBase, IServiceBehavior，代码如下

1. public partial class GreeterBehavior : Greeter.GreeterBase, IServiceBehavior
2. {
3.     private ServerReceivedDelegate received;
4.     public event ServerReceivedDelegate Received
5.     {
6.         add
7.         {
8.             if (value == null)
9.             {
10.                 received += value;
11.             }
12.         }
13.         remove
14.         {
15.             received -= value;
16.         }
17.     }
19.     public string MessageId { get; } = Guid.NewGuid().ToString("N");
20.     public async Task Write(object result, int statusCode = 200, string exceptionMessage = "")
21.     {
22.         if (received == null)
23.             return;
24.         var message = new TransportMessage(MessageId, new ReactiveResultMessage
25.         {
26.             ExceptionMessage = exceptionMessage,
27.             StatusCode = statusCode,
28.             Result = result
30.         });
31.         await received(message);
32.     }
34.     public T CreateProxy<T>(string key) where T : class
35.     {
36.         return ServiceLocator.GetService<IServiceProxyFactory>().CreateProxy<T>(key);
37.     }
39.     public object CreateProxy(Type type)
40.     {
41.         return ServiceLocator.GetService<IServiceProxyFactory>().CreateProxy(type);
42.     }
44.     public object CreateProxy(string key, Type type)
45.     {
46.         return ServiceLocator.GetService<IServiceProxyFactory>().CreateProxy(key, type);
47.     }
49.     public T CreateProxy<T>() where T : class
50.     {
51.         return ServiceLocator.GetService<IServiceProxyFactory>().CreateProxy<T>();
52.     }
54.     public T GetService<T>(string key) where T : class
55.     {
56.         if (ServiceLocator.Current.IsRegisteredWithKey<T>(key))
57.             return ServiceLocator.GetService<T>(key);
58.         else
59.             return ServiceLocator.GetService<IServiceProxyFactory>().CreateProxy<T>(key);
60.     }
62.     public T GetService<T>() where T : class
63.     {
64.         if (ServiceLocator.Current.IsRegistered<T>())
65.             return ServiceLocator.GetService<T>();
66.         else
67.             return ServiceLocator.GetService<IServiceProxyFactory>().CreateProxy<T>();
69.     }
71.     public object GetService(Type type)
72.     {
73.         if (ServiceLocator.Current.IsRegistered(type))
74.             return ServiceLocator.GetService(type);
75.         else
76.             return ServiceLocator.GetService<IServiceProxyFactory>().CreateProxy(type);
77.     }
79.     public object GetService(string key, Type type)
80.     {
81.         if (ServiceLocator.Current.IsRegisteredWithKey(key, type))
82.             return ServiceLocator.GetService(key, type);
83.         else
84.             return ServiceLocator.GetService<IServiceProxyFactory>().CreateProxy(key, type);
86.     }
87.     public void Publish(IntegrationEvent @event)
88.     {
89.         GetService<IEventBus>().Publish(@event);
90.     }
92. }

复制代码

 而基于grpc服务代码如下，需要继承于刚刚创建的GreeterBehavior

1.     public class GrpcDeviceDataService : GreeterBehavior, IGrpcDeviceDataService
2.     {
3.         public override Task<DeviceReply> ChangeDeviceStage(DeviceRequest request, ServerCallContext context)
4.         {
5.             return Task.FromResult(new DeviceReply
6.             {
7.                 Message = true
8.             }) ;
9.         }
10.     }

复制代码

以下是测试结果：

 6.添加/编辑MQTT服务组件

当添加/编辑MQTT组件时， 设置Host:127.0.0.1 ,port:425

   而基于mqtt服务代码如下，需要继承于MqttBehavior

1. public class MQTTDeviceDataService : MqttBehavior, IMQTTDeviceDataService
2. {
3.      public override async Task<bool> Authorized(string username, string password)
4.      {
5.          bool result = false;
6.          if (username == "admin" && password == "123456")
7.              result = true;
8.          return await Task.FromResult(result);
9.      }
11.      public async Task<bool> IsOnline(string deviceId)
12.      {
13.          return await base.GetDeviceIsOnine(deviceId);
14.      }
16.      public async Task Publish(string deviceId, WillMessage message)
17.      {
18.          var willMessage = new MqttWillMessage
19.          {
20.              WillMessage = message.Message,
21.              Qos = message.Qos,
22.              Topic = message.Topic,
23.              WillRetain = message.WillRetain
24.          };
25.          await Publish(deviceId, willMessage);
26.          await RemotePublish(deviceId, willMessage);
27.      }
28. }

复制代码

以下是测试结果：

三、总结

   木舟IOT平台会在github开源社区版本，可以自由更改代码，用于商业项目，但不能自营平台，如低代码平台，IOT平台等，如有违反，后果自负，还有最好不要更改命名空间，然后跟公司说是自己研发的，如果知道后，我在博客全网通报此人，以前surging相关的事件就算了，就当没发生过。，如果碰到困难，比较紧急的话，可以联系作者，加群：744677125

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