dotnetty 新的篇章- 开源

一、前言

      因为微服务引擎依赖于dotnetty组件，很多协议都是针对于dotnetty 进行扩展，然后对于老版本https://github.com/azure/dotnetty 停止更新后，本人下载源码进行修改更新，并且大家要求独立仓库进行开源，所以今天整理了代码开源至https://github.com/microsurging/DotNetty,  也希望大家一起贡献代码，让dotnetty 生态更强大。
HttpFlv:http://demo.kayakiot.cn:281/httpflv.html  （黑衣人）
 HttpFlv:http://demo.kayakiot.cn:281/httpflv1.html  （大红包）
HttpFlv:http://demo.kayakiot.cn:281/httpflv2.html  （鹿鼎记）
rtmp:rtmp://demo.kayakiot.cn:76/live1/livestream2   （黑衣人）
rtmp:rtmp://demo.kayakiot.cn:76/live1/livestream3   （大红包）
rtmp:rtmp://demo.kayakiot.cn:76/live1/livestream4（鹿鼎记）
注：测试服务器带宽只有8MB, httpflv  缓冲做的没有rtmp好，然后httpflv卡就多刷新几次
  凯亚 (Kayak) 是什么?
       凯亚(Kayak)是基于.NET8.0软件环境下的surging微服务引擎进行开发的, 平台包含了微服务和物联网平台。支持异步和响应式编程开发，功能包含了物模型,设备,产品,网络组件的统一管理和微服务平台下的注册中心，服务路由，模块，中间服务等管理。还有多协议适配(TCP,MQTT,UDP,CoAP,HTTP,Grpc,websocket,rtmp,httpflv,webservice,等),通过灵活多样的配置适配能够接入不同厂家不同协议等设备。并且通过设备告警,消息通知,数据可视化等功能。能够让你能快速建立起微服务物联网平台系统。
     凯亚物联网平台：http://demo.kayakiot.cn:3100（用户名：fanly  密码：123456）
    链路跟踪Skywalking V8:http://117.72.121.2:8080/
      surging 微服务引擎开源地址：https://github.com/fanliang11/surging（后面surging 会移动到microsurging进行维护）
二、ValueTask扩展支持

IValueTaskPromise：

1.     public interface IValueTaskPromise: IPromise
2.     {   
3.             bool IsVoid { get; }
5.             bool IsCompleted { get; }
7.             bool IsSuccess { get; }
9.             bool IsFaulted { get; }
11.             bool IsCanceled { get; }
13.             bool TryComplete();
15.             void Complete();
17.             bool TrySetException(Exception exception);
19.             bool TrySetException(IEnumerable<Exception> exceptions);
21.             void SetException(Exception exception);
23.             void SetException(IEnumerable<Exception> exceptions);
25.             bool TrySetCanceled();
27.             void SetCanceled();
29.             bool SetUncancellable();
31.              IPromise Unvoid();
32.         }

复制代码

DefaultValueTaskPromise：

1.     public class DefaultValueTaskPromise: IValueTaskPromise
2.     {
3.         private readonly CancellationToken _token;
4. #if NET
5.         private readonly TaskCompletionSource _tcs;
6. #else
7.         private readonly ManualResetValueTaskSource<object> _tcs;
8. #endif
10.         private int v_uncancellable = SharedConstants.False;
12.         public DefaultValueTaskPromise()
13.         {
14.             _token = CancellationToken.None;
15. #if NET
16.             _tcs = new TaskCompletionSource();
17. #else
18.             _tcs = new ManualResetValueTaskSource<object>();
19. #endif
20.         }
22.         public DefaultValueTaskPromise(object state)
23.         {
24. #if NET
25.             _tcs = new TaskCompletionSource(state);
26. #else
27.             _tcs = new ManualResetValueTaskSource<object>(state);
28. #endif
29.         }
31.         public DefaultValueTaskPromise(CancellationToken cancellationToken)
32.         {
33.             _token= cancellationToken;
34.         }
38.         public ValueTask ValueTask
39.         {
40.             [MethodImpl(InlineMethod.AggressiveOptimization)]
41.             get => _tcs.AwaitVoid(_token);
42.         }
44.         public bool IsVoid => false;
46.         public bool IsSuccess => ValueTask.IsCompletedSuccessfully;
48.         public bool IsCompleted => ValueTask.IsCompleted;
50.         public bool IsFaulted => ValueTask.IsFaulted;
52.         public bool IsCanceled => ValueTask.IsCanceled;
54.        public  Task  Task => ValueTask.AsTask();
56.         public virtual bool TryComplete()
57.         {
58. #if NET
59.             return _tcs.TrySetResult();
60. #else
61.             return _tcs.SetResult(0);
62. #endif
63.         }
65.         public virtual void Complete()
66.         {
67. #if NET
68.             _tcs.SetResult();
69. #else
70.             _tcs.SetResult(0);
71. #endif
72.         }
73.         public virtual void SetCanceled()
74.         {
75.             if (SharedConstants.False < (uint)Volatile.Read(ref v_uncancellable)) { return; }
76.             _tcs.SetCanceled();
77.         }
79.         public virtual void SetException(Exception exception)
80.         {
81.             if (exception is AggregateException aggregateException)
82.             {
83.                 SetException(aggregateException.InnerExceptions);
84.                 return;
85.             }
86.             _tcs.SetException(exception);
87.         }
89.         public virtual void SetException(IEnumerable<Exception> exceptions)
90.         {
91.             _tcs.SetException(exceptions.FirstOrDefault());
92.         }
94.         public virtual bool TrySetCanceled()
95.         {
96.             if (SharedConstants.False < (uint)Volatile.Read(ref v_uncancellable)) { return false; }
97.               _tcs.SetCanceled();
98.             return true;
99.         }
101.         public virtual bool TrySetException(Exception exception)
102.         {
103.             if (exception is AggregateException aggregateException)
104.             {
105.                 return TrySetException(aggregateException.InnerExceptions);
106.             }
107.               _tcs.SetException(exception);
108.             return true;
109.         }
111.         public virtual bool TrySetException(IEnumerable<Exception> exceptions)
112.         {
113.               _tcs.SetException(exceptions.FirstOrDefault());
114.             return true;
115.         }
117.         public bool SetUncancellable()
118.         {
119.             if (SharedConstants.False >= (uint)Interlocked.CompareExchange(ref v_uncancellable, SharedConstants.True, SharedConstants.False))
120.             {
121.                 return true;
122.             }
123.             return !IsCompleted;
124.         }
126.         public override string ToString() => "TaskCompletionSource[status: " + ValueTask.AsTask().Status.ToString() + "]";
128.         public IPromise Unvoid() => this;
129.          
130.     }

复制代码

ManualResetValueTaskSource：

1. internal interface IStrongBox<T>
2. {
3.      ref T Value { get; }
5.      bool RunContinuationsAsynchronously { get; set; }
6. }
8. public enum ContinuationOptions
9. {
10.      None,
12.      ForceDefaultTaskScheduler
13. }
15. public class ManualResetValueTaskSource<T> : IStrongBox<ManualResetValueTaskSourceLogic<T>>, IValueTaskSource<T>, IValueTaskSource
16. {
17.      private ManualResetValueTaskSourceLogic<T> _logic;
18.      private readonly Action _cancellationCallback;
20.      [MethodImpl(MethodImplOptions.AggressiveInlining)]
21.      public ManualResetValueTaskSource(ContinuationOptions options = ContinuationOptions.None)
22.      {
23.          _logic = new ManualResetValueTaskSourceLogic<T>(this, options,null);
24.          _cancellationCallback = SetCanceled;
25.      }
27.      public ManualResetValueTaskSource(object state, ContinuationOptions options = ContinuationOptions.None)
28.      {
29.          _logic = new ManualResetValueTaskSourceLogic<T>(this, options,state);
30.          _cancellationCallback = SetCanceled;
31.      }
33.      public short Version => _logic.Version;
36.      [MethodImpl(MethodImplOptions.AggressiveInlining)]
37.      public bool SetResult(T result)
38.      {
39.          lock (_cancellationCallback)
40.          {
41.              if (_logic.Completed)
42.              {
43.                  return false;
44.              }
46.              _logic.SetResult(result);
47.              return true;
48.          }
49.      }
51.      [MethodImpl(MethodImplOptions.AggressiveInlining)]
52.      public void SetException(Exception error)
53.      {
54.          if (Monitor.TryEnter(_cancellationCallback))
55.          {
56.              if (_logic.Completed)
57.              {
58.                  Monitor.Exit(_cancellationCallback);
59.                  return;
60.              }
62.              _logic.SetException(error);
63.              Monitor.Exit(_cancellationCallback);
64.          }
65.      }
67.      public void SetCanceled() => SetException(new TaskCanceledException());
69.      public T GetResult(short token) => _logic.GetResult(token);
71.      void IValueTaskSource.GetResult(short token) => _logic.GetResult(token);
73.      public ValueTaskSourceStatus GetStatus(short token) => _logic.GetStatus(token);
75.      public bool RunContinuationsAsynchronously { get; set; } = true;
77.      public void OnCompleted(Action<object> continuation, object state, short token, ValueTaskSourceOnCompletedFlags flags) => _logic.OnCompleted(continuation, state, token, flags);
79.      ref ManualResetValueTaskSourceLogic<T> IStrongBox<ManualResetValueTaskSourceLogic<T>>.Value => ref _logic;
81.      [MethodImpl(MethodImplOptions.AggressiveInlining)]
82.      public ValueTask<T> AwaitValue(CancellationToken cancellation)
83.      {
84.          CancellationTokenRegistration? registration = cancellation == CancellationToken.None
85.              ? (CancellationTokenRegistration?)null
86.              : cancellation.Register(_cancellationCallback);
87.          return _logic.AwaitValue(this, registration);
88.      }
90.      public ValueTask AwaitVoid(CancellationToken cancellation)
91.      {
92.          CancellationTokenRegistration? registration = cancellation == CancellationToken.None
93.              ? (CancellationTokenRegistration?)null
94.              : cancellation.Register(_cancellationCallback);
95.          return _logic.AwaitVoid(this, registration);
96.      }
98.      public void Reset() => _logic.Reset();
99. }
101. internal struct ManualResetValueTaskSourceLogic<TResult>
102. {
103.      private static readonly Action<object> s_sentinel = s => throw new InvalidOperationException();
105.      private readonly IStrongBox<ManualResetValueTaskSourceLogic<TResult>> _parent;
106.      private readonly ContinuationOptions _options;
107.      private Action<object> _continuation;
108.      private object _continuationState;
109.      private object _capturedContext;
110.      private ExecutionContext _executionContext;
111.      private bool _completed;
112.      private TResult _result;
113.      private ExceptionDispatchInfo _error;
114.      private CancellationTokenRegistration? _registration;
115.       
116.      public ManualResetValueTaskSourceLogic(IStrongBox<ManualResetValueTaskSourceLogic<TResult>> parent, ContinuationOptions options,object state)
117.      {
118.          _parent = parent ?? throw new ArgumentNullException(nameof(parent));
119.          _options = options;
120.          _continuation = null;
121.          _continuationState = null;
122.          _capturedContext = null;
123.          _executionContext = null;
124.          _completed = state != null;
125.          _result =state==null? default(TResult): (TResult)state;
126.          _error = null;
127.          Version = 0;
128.          _registration = null;
129.      }
131.      public short Version { get; private set; }
133.      public bool Completed => _completed;
135.      private void ValidateToken(short token)
136.      {
137.          if (token != Version)
138.          {
139.              throw new InvalidOperationException();
140.          }
141.      }
143.      public ValueTaskSourceStatus GetStatus(short token)
144.      {
145.         // ValidateToken(token);
147.          return
148.              !_completed ? ValueTaskSourceStatus.Pending :
149.              _error == null ? ValueTaskSourceStatus.Succeeded :
150.              _error.SourceException is OperationCanceledException ? ValueTaskSourceStatus.Canceled :
151.              ValueTaskSourceStatus.Faulted;
152.      }
154.      public TResult GetResult(short token)
155.      {
156.         // ValidateToken(token);
158.          if (!_completed)
159.          {
160.              return _result;
161.          }
163.          TResult result = _result;
164.          ExceptionDispatchInfo error = _error;
165.          Reset();
167.          error?.Throw();
168.          return result;
169.      }
171.      public void Reset()
172.      {
173.          Version++;
175.          _registration?.Dispose();
177.          _completed = false;
178.          _continuation = null;
179.          _continuationState = null;
180.          _result = default(TResult);
181.          _error = null;
182.          _executionContext = null;
183.          _capturedContext = null;
184.          _registration = null;
185.      }
187.      public void OnCompleted(Action<object> continuation, object state, short token, ValueTaskSourceOnCompletedFlags flags)
188.      {
189.          if (continuation == null)
190.          {
191.              throw new ArgumentNullException(nameof(continuation));
192.          }
194.          ValidateToken(token);
197.          if ((flags & ValueTaskSourceOnCompletedFlags.FlowExecutionContext) != 0)
198.          {
199.              _executionContext = ExecutionContext.Capture();
200.          }
202.          if ((flags & ValueTaskSourceOnCompletedFlags.UseSchedulingContext) != 0)
203.          {
204.              SynchronizationContext sc = SynchronizationContext.Current;
205.              if (sc != null && sc.GetType() != typeof(SynchronizationContext))
206.              {
207.                  _capturedContext = sc;
208.              }
209.              else
210.              {
211.                  TaskScheduler ts = TaskScheduler.Current;
212.                  if (ts != TaskScheduler.Default)
213.                  {
214.                      _capturedContext = ts;
215.                  }
216.              }
217.          }
219.          _continuationState = state;
220.          if (Interlocked.CompareExchange(ref _continuation, continuation, null) != null)
221.          {
222.              _executionContext = null;
224.              object cc = _capturedContext;
225.              _capturedContext = null;
227.              switch (cc)
228.              {
229.                  case null:
230.                      Task.Factory.StartNew(continuation, state, CancellationToken.None, TaskCreationOptions.DenyChildAttach, TaskScheduler.Default);
231.                      break;
233.                  case SynchronizationContext sc:
234.                      sc.Post(s =>
235.                      {
236.                          var tuple = (Tuple, object>)s;
237.                          tuple.Item1(tuple.Item2);
238.                      }, Tuple.Create(continuation, state));
239.                      break;
241.                  case TaskScheduler ts:
242.                      Task.Factory.StartNew(continuation, state, CancellationToken.None, TaskCreationOptions.DenyChildAttach, ts);
243.                      break;
244.              }
245.          }
246.      }
248.      public void SetResult(TResult result)
249.      {
250.          _result = result;
251.          SignalCompletion();
252.      }
254.      [MethodImpl(MethodImplOptions.AggressiveInlining)]
255.      public void SetException(Exception error)
256.      {
257.          _error = ExceptionDispatchInfo.Capture(error);
258.          SignalCompletion();
259.      }
261.      private void SignalCompletion()
262.      {
263.          if (_completed)
264.          {
265.              throw new InvalidOperationException("Double completion of completion source is prohibited");
266.          }
268.          _completed = true;
270.          if (Interlocked.CompareExchange(ref _continuation, s_sentinel, null) != null)
271.          {
272.              if (_executionContext != null)
273.              {
274.                  ExecutionContext.Run(
275.                      _executionContext,
276.                      s => ((IStrongBox<ManualResetValueTaskSourceLogic<TResult>>)s).Value.InvokeContinuation(),
277.                      _parent ?? throw new InvalidOperationException());
278.              }
279.              else
280.              {
281.                  InvokeContinuation();
282.              }
283.          }
284.      }
286.      private void InvokeContinuation()
287.      {
288.          object cc = _capturedContext;
289.          _capturedContext = null;
291.          if (_options == ContinuationOptions.ForceDefaultTaskScheduler)
292.          {
293.              cc = TaskScheduler.Default;
294.          }
296.          switch (cc)
297.          {
298.              case null:
299.                  if (_parent.RunContinuationsAsynchronously)
300.                  {
301.                      var c = _continuation;
302.                      if (_executionContext != null)
303.                      {
304.                          ThreadPool.QueueUserWorkItem(s => c(s), _continuationState);
305.                      }
306.                      else
307.                      {
308.                          ThreadPool.UnsafeQueueUserWorkItem(s => c(s),  _continuationState);
309.                      }
310.                  }
311.                  else
312.                  {
313.                      _continuation(_continuationState);
314.                  }
315.                  break;
317.              case SynchronizationContext sc:
318.                  sc.Post(s =>
319.                  {
320.                      ref ManualResetValueTaskSourceLogic<TResult> logicRef = ref ((IStrongBox<ManualResetValueTaskSourceLogic<TResult>>)s).Value;
321.                      logicRef._continuation(logicRef._continuationState);
322.                  }, _parent ?? throw new InvalidOperationException());
323.                  break;
325.              case TaskScheduler ts:
326.                  Task.Factory.StartNew(_continuation, _continuationState, CancellationToken.None, TaskCreationOptions.DenyChildAttach, ts);
327.                  break;
328.          }
329.      }
331.      public ValueTask<T> AwaitValue<T>(IValueTaskSource<T> source, CancellationTokenRegistration? registration)
332.      {
333.          _registration = registration;
334.          return new ValueTask<T>(source, Version);
335.      }
337.      public ValueTask AwaitVoid(IValueTaskSource source, CancellationTokenRegistration? registration)
338.      {
339.          _registration = registration;
340.          return new ValueTask(source, Version);
341.      }
342. }

复制代码

然后把DefaultPromise 替换成DefaultValueTaskPromise，如下图所示

三、扩展支持rtmp编解码

如下图所示 ：

四、demo展示

开启了三个通道进行推流，cpu,内存都比较稳定

 在凯亚物联网平台你也可以创建rtmp组件

 
视频中心

 

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！