From bf9544353dbc07199d62df998de35e4616e7eee5 Mon Sep 17 00:00:00 2001
From: KK <1985088337@qq.com>
Date: Thu, 3 Jun 2021 00:07:16 +0800
Subject: [PATCH] =?UTF-8?q?=E6=89=A9=E5=85=85=E5=8D=95=E5=90=91=E6=95=B0?=
 =?UTF-8?q?=E6=8D=AE=E5=8F=91=E5=B8=83=E8=80=85=E5=92=8C=E8=AE=A2=E9=98=85?=
 =?UTF-8?q?=E8=80=85=E7=9B=B8=E5=85=B3=E6=8E=A5=E5=8F=A3?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 Data/OSS.Tools.DataFlow/DataFlowFactory.cs    | 46 ++++++++++++++++++-
 Data/OSS.Tools.DataFlow/IDataFlowProvider.cs  | 34 +++++++++++++-
 Data/OSS.Tools.DataFlow/IDataSubscriber.cs    |  8 +---
 .../OSS.Tools.DataFlow.csproj                 |  4 +-
 4 files changed, 82 insertions(+), 10 deletions(-)

diff --git a/Data/OSS.Tools.DataFlow/DataFlowFactory.cs b/Data/OSS.Tools.DataFlow/DataFlowFactory.cs
index 77c1419..a1ce252 100644
--- a/Data/OSS.Tools.DataFlow/DataFlowFactory.cs
+++ b/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;
+        }
     }
 
+    
 
- 
 }
diff --git a/Data/OSS.Tools.DataFlow/IDataFlowProvider.cs b/Data/OSS.Tools.DataFlow/IDataFlowProvider.cs
index a6fc479..563144b 100644
--- a/Data/OSS.Tools.DataFlow/IDataFlowProvider.cs
+++ b/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");
+    }
 }
diff --git a/Data/OSS.Tools.DataFlow/IDataSubscriber.cs b/Data/OSS.Tools.DataFlow/IDataSubscriber.cs
index 4024a03..a24bbc2 100644
--- a/Data/OSS.Tools.DataFlow/IDataSubscriber.cs
+++ b/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)
diff --git a/Data/OSS.Tools.DataFlow/OSS.Tools.DataFlow.csproj b/Data/OSS.Tools.DataFlow/OSS.Tools.DataFlow.csproj
index 6e0a20f..ea5934d 100644
--- a/Data/OSS.Tools.DataFlow/OSS.Tools.DataFlow.csproj
+++ b/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>