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(poper); } + + + /// + /// 数据流的提供者 + /// + public static IDataFlowPublisherProvider PublisherProvider { get; set; } + + /// + /// 创建单向数据发布者 + /// + /// + /// 流key + /// + /// 返回当前流的发布接口实现 + public static IDataPublisher CreatePublisher( string flowKey = "task_flow", string sourceName = "default") + { + var pusher = PublisherProvider?.CreatePublisher< TData>( flowKey, sourceName); + if (pusher==null) + { + throw new NotImplementedException($"无法找到对应的{flowKey}值的 IDataFlowPublisherProvider 的具体实现"); + } + + return pusher; + } + + + /// + /// 数据流订阅者的接收器 + /// + public static IDataFlowSubscriberReceiver SubscriberReceiver { get; set; } + + /// + /// 接收数据订阅者 + /// + /// + /// + /// 流key + /// + /// 是否接收成功 + public static bool ReceiveSubscriber(IDataSubscriber 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 { /// - /// 数据流提供者 + /// 数据流提供器 /// public interface IDataFlowProvider { @@ -16,4 +16,36 @@ namespace OSS.Tools.DataFlow /// 返回当前流的发布接口实现 IDataPublisher CreateFlow(IDataSubscriber subscriber, string flowKey, string sourceName = "default"); } + + + /// + /// 数据流发布者的提供器 + /// + public interface IDataFlowPublisherProvider + { + /// + /// 创建单向数据发布者 + /// + /// + /// 流key + /// + /// 返回当前流的发布接口实现 + IDataPublisher CreatePublisher(string flowKey, string sourceName = "default"); + } + + /// + /// 数据流订阅者的接收器 + /// + public interface IDataFlowSubscriberReceiver + { + /// + /// 接收数据订阅者 + /// + /// + /// + /// 流key + /// + /// 是否接收成功 + bool Receive(IDataSubscriber 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 /// 数据的订阅者 /// /// - public interface IDataSubscriber + public interface IDataSubscriber { /// /// 弹出数据 @@ -23,11 +23,7 @@ namespace OSS.Tools.DataFlow internal InterDataSubscriber(Func> subscribeFunc) { - if (subscribeFunc == null) - { - throw new ArgumentNullException("Func> poper 方法不能为空!"); - } - _subscriber = subscribeFunc; + _subscriber = subscribeFunc ?? throw new ArgumentNullException(nameof(subscribeFunc),"订阅者方法不能为空!"); } public Task 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 @@ netstandard2.0 - 0.1 + 0.2 数据异步流处理 - 数据的异步流缓冲处理 + 扩充单向发布者和订阅者相关接口 logo.png