扩充单向数据发布者和订阅者相关接口

Data/OSS.Tools.DataFlow/DataFlowFactory.cs

@@ -42,8 +42,52 @@ namespace OSS.Tools.DataFlow
             var pusher = FlowProvider?.CreateFlow(poper, flowKey, sourceName);
             return pusher ?? new DefaultDataFlow<TData>(poper);
         }
+
+
+        /// <summary>
+        /// 数据流的提供者
+        /// </summary>
+        public static IDataFlowPublisherProvider PublisherProvider { get; set; }
+
+        /// <summary>
+        /// 创建单向数据发布者
+        /// </summary>
+        /// <typeparam name="TData"></typeparam>
+        /// <param name="flowKey"> 流key  </param>
+        /// <param name="sourceName"></param>
+        /// <returns> 返回当前流的发布接口实现 </returns>
+        public static IDataPublisher<TData> CreatePublisher<TData>( string flowKey = "task_flow", string sourceName = "default")
+        {
+            var pusher = PublisherProvider?.CreatePublisher< TData>( flowKey, sourceName);
+            if (pusher==null)
+            {
+                throw new NotImplementedException($"无法找到对应的{flowKey}值的 IDataFlowPublisherProvider 的具体实现");
+            }
+
+            return pusher;
+        }
+
+
+        /// <summary>
+        /// 数据流订阅者的接收器
+        /// </summary>
+        public static IDataFlowSubscriberReceiver SubscriberReceiver { get; set; }
+
+        /// <summary>
+        /// 接收数据订阅者
+        /// </summary>
+        /// <typeparam name="TData"></typeparam>
+        /// <param name="subscriber"></param>
+        /// <param name="flowKey"> 流key  </param>
+        /// <param name="sourceName"></param>
+        /// <returns> 是否接收成功 </returns>
+        public static bool ReceiveSubscriber<TData>(IDataSubscriber<TData> subscriber, string flowKey,
+            string sourceName = "default")
+        {
+            return SubscriberReceiver?.Receive(subscriber, flowKey, sourceName) ?? false;
+        }
     }
 
+    
 
- 
 }

Data/OSS.Tools.DataFlow/IDataFlowProvider.cs

@@ -2,7 +2,7 @@
 namespace OSS.Tools.DataFlow
 {
     /// <summary>
-    ///  数据流提供者
+    ///  数据流提供器
     /// </summary>
     public interface IDataFlowProvider
     {
@@ -16,4 +16,36 @@ namespace OSS.Tools.DataFlow
         /// <returns> 返回当前流的发布接口实现 </returns>
         IDataPublisher<TData> CreateFlow<TData>(IDataSubscriber<TData> subscriber, string flowKey, string sourceName = "default");
     }
+
+
+    /// <summary>
+    ///  数据流发布者的提供器
+    /// </summary>
+    public interface IDataFlowPublisherProvider
+    {
+        /// <summary>
+        /// 创建单向数据发布者
+        /// </summary>
+        /// <typeparam name="TData"></typeparam>
+        /// <param name="flowKey"> 流key  </param>
+        /// <param name="sourceName"></param>
+        /// <returns> 返回当前流的发布接口实现 </returns>
+        IDataPublisher<TData> CreatePublisher<TData>(string flowKey, string sourceName = "default");
+    }
+    
+    /// <summary>
+    ///  数据流订阅者的接收器
+    /// </summary>
+    public interface IDataFlowSubscriberReceiver
+    {
+        /// <summary>
+        /// 接收数据订阅者
+        /// </summary>
+        /// <typeparam name="TData"></typeparam>
+        /// <param name="subscriber"></param>
+        /// <param name="flowKey"> 流key  </param>
+        /// <param name="sourceName"></param>
+        /// <returns> 是否接收成功 </returns>
+        bool Receive<TData>(IDataSubscriber<TData> subscriber, string flowKey, string sourceName = "default");
+    }
 }

Data/OSS.Tools.DataFlow/IDataSubscriber.cs

@@ -7,7 +7,7 @@ namespace OSS.Tools.DataFlow
     ///  数据的订阅者
     /// </summary>
     /// <typeparam name="TData"></typeparam>
-    public interface IDataSubscriber<TData>
+    public interface IDataSubscriber<in TData>
     {
         /// <summary>
         /// 弹出数据
@@ -23,11 +23,7 @@ namespace OSS.Tools.DataFlow
 
         internal InterDataSubscriber(Func<TData, Task<bool>> subscribeFunc)
         {
-            if (subscribeFunc == null)
-            {
-                throw new ArgumentNullException("Func<TData, Task<bool>> poper 方法不能为空！");
-            }
-            _subscriber = subscribeFunc;
+            _subscriber = subscribeFunc ?? throw new ArgumentNullException(nameof(subscribeFunc),"订阅者方法不能为空！");
         }
 
         public Task<bool> Subscribe(TData data)

Data/OSS.Tools.DataFlow/OSS.Tools.DataFlow.csproj

@@ -2,9 +2,9 @@
 
   <PropertyGroup>
     <TargetFramework>netstandard2.0</TargetFramework>
-    <Version>0.1</Version>
+    <Version>0.2</Version>
     <Description>数据异步流处理</Description>
-    <PackageReleaseNotes>数据的异步流缓冲处理</PackageReleaseNotes>
+    <PackageReleaseNotes>扩充单向发布者和订阅者相关接口</PackageReleaseNotes>
     <PackageIcon>logo.png</PackageIcon>
   </PropertyGroup>