Flter cleanup.

Author: Oren (electricessence)

Filters/AcceptOrPass.cs

@@ -1,37 +1,36 @@
-using Open.Threading;
-using System.Threading.Tasks.Dataflow;
+using System.Threading.Tasks.Dataflow;
 
 namespace Open.Threading.Dataflow
 {
 
-	internal class AutoCompleteFilter<T> : TargetBlockFilter<T>
+    internal class AutoCompleteFilter<T> : TargetBlockFilterBase<T>
 	{
 		public AutoCompleteFilter(int limit, ITargetBlock<T> target) : base(target)
 		{
 			Limit = limit;
 		}
 
 		public int Limit { get; }
-		public int AllowedCount { get; private set; }
+        public int AllowedCount { get; private set; }
 
 
-		// The key here is to reject the message ahead of time.
-		public override DataflowMessageStatus OfferMessage(DataflowMessageHeader messageHeader, T messageValue, ISourceBlock<T> source, bool consumeToAccept)
+        // The key here is to reject the message ahead of time.
+        public override DataflowMessageStatus OfferMessage(DataflowMessageHeader messageHeader, T messageValue, ISourceBlock<T> source, bool consumeToAccept)
 		{
 			var result = DataflowMessageStatus.DecliningPermanently;
 			var completed = false;
-			// There are multiple operations happening here that require synchronization to get right.
-			ThreadSafety.LockConditional(_target,
+            // There are multiple operations happening here that require synchronization to get right.
+            ThreadSafety.LockConditional(Target,
 				() => AllowedCount < Limit,
 				() =>
 				{
-					AllowedCount++;
-					result = _target.OfferMessage(messageHeader, messageValue, source, consumeToAccept);
+                    AllowedCount++;
+					result = Target.OfferMessage(messageHeader, messageValue, source, consumeToAccept);
 					completed = AllowedCount == Limit;
 				}
 			);
 
-			if (completed) _target.Complete();
+			if (completed) Target.Complete();
 
 			return result;
 		}

Filters/AutoComplete.cs

@@ -1,34 +1,27 @@
-using Open.Threading;
-using System.Threading.Tasks.Dataflow;
+using System.Threading.Tasks.Dataflow;
 
 namespace Open.Threading.Dataflow
 {
-	internal class ChangedFilter<T> : TargetBlockFilter<T>
+    internal class ChangedFilter<T> : TargetBlockFilter<T>
 	{
-		readonly DataflowMessageStatus _defaultResponseForDuplicate;
-
-		T _last;
-
-		public ChangedFilter(DataflowMessageStatus defaultResponseForDuplicate, ITargetBlock<T> target) : base(target)
+		public ChangedFilter(ITargetBlock<T> target, DataflowMessageStatus defaultResponseForDuplicate)
+            : base(target, defaultResponseForDuplicate, null)
 		{
-			_defaultResponseForDuplicate = defaultResponseForDuplicate;
 		}
 
-		// The key here is to reject the message ahead of time.
-		public override DataflowMessageStatus OfferMessage(DataflowMessageHeader messageHeader, T messageValue, ISourceBlock<T> source, bool consumeToAccept)
-		{
-			return ThreadSafety.LockConditional(
-					_target,
-					() => !messageValue.Equals(_last),
-					() => _last = messageValue)
-				? _target.OfferMessage(messageHeader, messageValue, source, consumeToAccept)
-				: _defaultResponseForDuplicate;
-		}
+        readonly object SyncLock = new object();
+        T _last;
+
+        protected override bool Accept(T messageValue)
+            => ThreadSafety.LockConditional(
+                SyncLock,
+                () => !messageValue.Equals(_last),
+                () => _last = messageValue);
 	}
 
     public static partial class DataFlowExtensions
     {
         public static ITargetBlock<T> OnlyIfChanged<T>(this ITargetBlock<T> target, DataflowMessageStatus defaultResponseForDuplicate)
-            => new ChangedFilter<T>(defaultResponseForDuplicate, target);
+            => new ChangedFilter<T>(target, defaultResponseForDuplicate);
     }
 }

Filters/Changed.cs

@@ -5,32 +5,26 @@ namespace Open.Threading.Dataflow
 {
 	internal class DistinctFilter<T> : TargetBlockFilter<T>
 	{
-		private readonly DataflowMessageStatus _defaultResponseForDuplicate;
-
-		private readonly HashSet<T> _set = new HashSet<T>();
-
-		public DistinctFilter(DataflowMessageStatus defaultResponseForDuplicate, ITargetBlock<T> target) : base(target)
+		public DistinctFilter(ITargetBlock<T> target, DataflowMessageStatus defaultResponseForDuplicate)
+            : base(target, defaultResponseForDuplicate, null)
 		{
-			_defaultResponseForDuplicate = defaultResponseForDuplicate;
 		}
 
-		// The key here is to reject the message ahead of time.
-		public override DataflowMessageStatus OfferMessage(DataflowMessageHeader messageHeader, T messageValue, ISourceBlock<T> source, bool consumeToAccept)
+        private readonly HashSet<T> _set = new HashSet<T>();
+
+        protected override bool Accept(T messageValue)
 		{
 			bool didntHave;
-			lock (_target) // Assure order of acceptance.
+			lock (Target) // Assure order of acceptance.
 				didntHave = _set.Add(messageValue);
-			if (didntHave)
-				return _target.OfferMessage(messageHeader, messageValue, source, consumeToAccept);
-
-			return _defaultResponseForDuplicate;
+            return didntHave;
 		}
 	}
 
     public static partial class DataFlowExtensions
     {
         public static ITargetBlock<T> Distinct<T>(this ITargetBlock<T> target, DataflowMessageStatus defaultResponseForDuplicate)
-            => new DistinctFilter<T>(defaultResponseForDuplicate, target);
+            => new DistinctFilter<T>(target, defaultResponseForDuplicate);
     }
 
 }

Filters/Distinct.cs

@@ -0,0 +1,71 @@
+using System;
+using System.Threading.Tasks.Dataflow;
+
+namespace Open.Threading.Dataflow
+{
+    internal class TargetBlockFilter<T> : TargetBlockFilterBase<T>
+	{
+        private readonly DataflowMessageStatus _filterDeclineStatus;
+        private readonly Func<T, bool> _filter;
+
+        public TargetBlockFilter(
+            ITargetBlock<T> target,
+            DataflowMessageStatus filterDeclineStatus,
+            Func<T, bool> filter) :base(target)
+		{
+            _filter = filter;
+            _filterDeclineStatus = filterDeclineStatus;
+        }
+
+        protected virtual bool Accept(T messageValue)
+            => _filter(messageValue);
+
+        public override DataflowMessageStatus OfferMessage(
+            DataflowMessageHeader messageHeader, T messageValue, ISourceBlock<T> source, bool consumeToAccept)
+            => Accepting && !Accept(messageValue)
+                ? _filterDeclineStatus
+                : base.OfferMessage(messageHeader, messageValue, source, consumeToAccept);
+    }
+
+    public static partial class DataFlowExtensions
+    {
+        /// <summary>
+        /// Synchronously filters out incomming messages before they arrive to the target.
+        /// Passes the message to the target if the filter function returns true.
+        /// If the filter function returns false, then the message will either be accepted and disappear (default), or will be declined if the decli
+        /// </summary>
+        /// <typeparam name="T">The message type.</typeparam>
+        /// <param name="target">The Target-Block to pass messages too.</param>
+        /// <param name="filter">Passes the message to the target if the filter function returns true.</param>
+        /// <param name="decline">If true, and the filter returns false, the message will be declined, allowing it to continue elsewhere.  If false (default) the message is accepted even though it is not being passed to the target.</param>
+        /// <returns>A filter block that preceeds the target..</returns>
+        public static ITargetBlock<T> Filter<T>(this ITargetBlock<T> target,
+            Func<T, bool> filter,
+            bool decline = false)
+            => new TargetBlockFilter<T>(
+                target,
+                decline ? DataflowMessageStatus.Declined : DataflowMessageStatus.Accepted,
+                filter);
+
+
+        /// <summary>
+        /// Processes an item through an acceptor function.
+        /// If the acceptor returns true, then it was accepted and subsequently received/taken from the source (no longer available to downstream targets).
+        /// </summary>
+        /// <typeparam name="T">The message type</typeparam>
+        /// <param name="source">The source block to receive from.</param>
+        /// <param name="acceptor">The function to process the item and decide if accepted.</param>
+        /// <returns>The original source block to allow for more acceptors or filters to be applied.</returns>
+        public static ISourceBlock<T> TakeOrContinue<T>(this ISourceBlock<T> source,
+            Func<T, bool> acceptor)
+        {
+            var receiver = DataflowBlock
+                .NullTarget<T>() // If the acceptor returns true, the message is dropped.
+                .Filter(acceptor, true); // Else it's declined and continues on to the next target.
+
+            source.LinkToWithCompletion(receiver);
+            return source;
+        }
+    }
+
+}

Filters/Filter.cs

@@ -15,10 +15,10 @@ Part of the "Open" set of libraries.</Description>
     <RepositoryUrl>https://github.com/electricessence/Open.Threading.Dataflow/</RepositoryUrl>
     <RepositoryType>git</RepositoryType>
     <PackageTags>dotnet, dotnet-core, dotnetcore, cs, dataflow, tpl, extensions</PackageTags>
-    <Version>2.2.0</Version>
-    <PackageReleaseNotes></PackageReleaseNotes>
-    <AssemblyVersion>2.2.0.0</AssemblyVersion>
-    <FileVersion>2.2.0.0</FileVersion>
+    <Version>2.2.1</Version>
+    <PackageReleaseNotes>Cleanup of filters with and added .Filter extension for ITargetBlock&lt;T&gt;.</PackageReleaseNotes>
+    <AssemblyVersion>2.2.1.0</AssemblyVersion>
+    <FileVersion>2.2.1.0</FileVersion>
     <PackageLicenseExpression>MIT</PackageLicenseExpression>
   </PropertyGroup>
 

Open.Threading.Dataflow.csproj

@@ -0,0 +1,50 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using System.Threading.Tasks.Dataflow;
+
+namespace Open.Threading.Dataflow
+{
+    internal abstract class TargetBlockFilterBase<T> : ITargetBlock<T>
+    {
+        protected readonly ITargetBlock<T> Target;
+
+        protected TargetBlockFilterBase(ITargetBlock<T> target)
+        {
+            Target = target ?? throw new ArgumentNullException(nameof(target));
+        }
+
+        protected int _state;
+        const int ACCEPTING = 0;
+        const int REJECTING = 1;
+
+        protected bool Accepting => _state == ACCEPTING && !Target.Completion.IsCompleted;
+
+        protected void CompleteInternal()
+        {
+            if(_state == ACCEPTING)
+                Interlocked.CompareExchange(ref _state, REJECTING, ACCEPTING);
+        }
+
+        public void Complete()
+        {
+            CompleteInternal();
+            Target.Complete();
+        }
+
+        public void Fault(Exception exception)
+        {
+            CompleteInternal();
+            Target.Fault(exception);
+        }
+
+        public Task Completion => Target.Completion;
+
+        // The key here is to reject the message ahead of time.
+        public virtual DataflowMessageStatus OfferMessage(
+            DataflowMessageHeader messageHeader, T messageValue, ISourceBlock<T> source, bool consumeToAccept)
+            => Accepting
+                ? Target.OfferMessage(messageHeader, messageValue, source, consumeToAccept)
+                : DataflowMessageStatus.DecliningPermanently;
+    }
+}