Added Join and OfType filters.

Author: Oren (electricessence)

Open.ChannelExtensions/Extensions.Filter.cs

@@ -8,39 +8,36 @@ namespace Open.ChannelExtensions
 {
 	public static partial class Extensions
 	{
-		class TransformingChannelReader<TIn, TOut> : ChannelReader<TOut>
+		class FilteringChannelReader<T> : ChannelReader<T>
 		{
-			public TransformingChannelReader(ChannelReader<TIn> source, Func<TIn, TOut> transform)
+			public FilteringChannelReader(ChannelReader<T> source, Func<T, bool> predicate)
 			{
 				_source = source ?? throw new ArgumentNullException(nameof(source));
-				_transform = transform ?? throw new ArgumentNullException(nameof(transform));
+				_predicate = predicate ?? throw new ArgumentNullException(nameof(predicate));
 				Contract.EndContractBlock();
 			}
 
-			private readonly ChannelReader<TIn> _source;
-			private readonly Func<TIn, TOut> _transform;
+			private readonly ChannelReader<T> _source;
+			private readonly Func<T, bool> _predicate;
 			public override Task Completion => _source.Completion;
 
-			public override bool TryRead(out TOut item)
+			public override bool TryRead(out T item)
 			{
-				if (_source.TryRead(out var e))
+				while (_source.TryRead(out item))
 				{
-					item = _transform(e);
-					return true;
+					if (_predicate(item))
+						return true;
 				}
 
 				item = default;
 				return false;
 			}
 
-			public override async ValueTask<TOut> ReadAsync(CancellationToken cancellationToken = default)
-				=> _transform(await _source.ReadAsync(cancellationToken));
-
 			public override ValueTask<bool> WaitToReadAsync(CancellationToken cancellationToken = default)
 				=> _source.WaitToReadAsync(cancellationToken);
 		}
 
-		public static ChannelReader<TOut> Transform<TIn, TOut>(this ChannelReader<TIn> source, Func<TIn, TOut> transform)
-			=> new TransformingChannelReader<TIn, TOut>(source, transform);
+		public static ChannelReader<T> Filter<T>(this ChannelReader<T> source, Func<T, bool> predicate)
+			=> new FilteringChannelReader<T>(source, predicate);
 	}
 }

Open.ChannelExtensions/Extensions.Join.cs

@@ -0,0 +1,102 @@
+using System;
+using System.Collections.Generic;
+using System.Diagnostics.Contracts;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+
+namespace Open.ChannelExtensions
+{
+	public static partial class Extensions
+	{
+		class JoiningChannelReader<TList, T> : ChannelReader<T>
+			where TList : IEnumerable<T>
+		{
+			readonly Channel<T> _buffer = Channel.CreateUnbounded<T>(new UnboundedChannelOptions { AllowSynchronousContinuations = true });
+			public JoiningChannelReader(ChannelReader<TList> source)
+			{
+				_source = source ?? throw new ArgumentNullException(nameof(source));
+				Contract.EndContractBlock();
+			}
+
+			private readonly ChannelReader<TList> _source;
+			public override Task Completion
+				=> _source.Completion.ContinueWith(t =>
+				{
+					// Need to be sure writing is done before we continue...
+					lock (_buffer)
+					{
+						_buffer.Writer.Complete(t.Exception);
+					}
+					return _buffer.Reader.Completion;
+				})
+				.Unwrap();
+
+			bool TryPipeItems()
+			{
+				lock (_buffer)
+				{
+					if (!_source.TryRead(out TList batch))
+						return false;
+
+					foreach (var i in batch)
+					{
+						// Assume this will always be true for our internal unbound channel.
+						_buffer.Writer.TryWrite(i);
+					}
+
+					return true;
+				}
+			}
+
+			public override bool TryRead(out T item)
+			{
+				do
+				{
+					if (_buffer.Reader.TryRead(out item))
+						return true;
+				}
+				while (TryPipeItems());
+
+				item = default;
+				return false;
+			}
+
+			public override ValueTask<bool> WaitToReadAsync(CancellationToken cancellationToken = default)
+			{
+				if (cancellationToken.IsCancellationRequested)
+					return new ValueTask<bool>(Task.FromCanceled<bool>(cancellationToken));
+
+				var b = _buffer.Reader.WaitToReadAsync(cancellationToken);
+				if (b.IsCompletedSuccessfully || _source.Completion.IsCompleted)
+					return b;
+
+				var s = _source.WaitToReadAsync(cancellationToken);
+				if (s.IsCompletedSuccessfully && !s.Result)
+					return b;
+
+				return WaitCore();
+
+				async ValueTask<bool> WaitCore()
+				{
+					cancellationToken.ThrowIfCancellationRequested();
+
+					// Not sure if there's a better way to 'WhenAny' with a ValueTask yet.
+					var bt = b.AsTask();
+					var st = s.AsTask();
+					var first = await Task.WhenAny(bt, st);
+					// Either one? Ok go.
+					if (first.Result) return true;
+					// Buffer returned false? We're done.
+					if (first == bt) return false;
+					// Second return false? Wait for buffer.
+					return await bt;
+				}
+			}
+		}
+
+		public static ChannelReader<T> Join<TList, T>(this ChannelReader<TList> source)
+			where TList : IEnumerable<T>
+			=> new JoiningChannelReader<TList, T>(source);
+	}
+}

Open.ChannelExtensions/Extensions.Transform.cs

@@ -8,36 +8,39 @@ namespace Open.ChannelExtensions
 {
 	public static partial class Extensions
 	{
-		class FilteringChannelReader<T> : ChannelReader<T>
+		class TransformingChannelReader<TIn, TOut> : ChannelReader<TOut>
 		{
-			public FilteringChannelReader(ChannelReader<T> source, Func<T, bool> predicate)
+			public TransformingChannelReader(ChannelReader<TIn> source, Func<TIn, TOut> transform)
 			{
 				_source = source ?? throw new ArgumentNullException(nameof(source));
-				_predicate = predicate ?? throw new ArgumentNullException(nameof(predicate));
+				_transform = transform ?? throw new ArgumentNullException(nameof(transform));
 				Contract.EndContractBlock();
 			}
 
-			private readonly ChannelReader<T> _source;
-			private readonly Func<T, bool> _predicate;
+			private readonly ChannelReader<TIn> _source;
+			private readonly Func<TIn, TOut> _transform;
 			public override Task Completion => _source.Completion;
 
-			public override bool TryRead(out T item)
+			public override bool TryRead(out TOut item)
 			{
-				while (_source.TryRead(out item))
+				if (_source.TryRead(out var e))
 				{
-					if (_predicate(item))
-						return true;
+					item = _transform(e);
+					return true;
 				}
 
 				item = default;
 				return false;
 			}
 
+			public override async ValueTask<TOut> ReadAsync(CancellationToken cancellationToken = default)
+				=> _transform(await _source.ReadAsync(cancellationToken));
+
 			public override ValueTask<bool> WaitToReadAsync(CancellationToken cancellationToken = default)
 				=> _source.WaitToReadAsync(cancellationToken);
 		}
 
-		public static ChannelReader<T> Filter<T>(this ChannelReader<T> source, Func<T, bool> predicate)
-			=> new FilteringChannelReader<T>(source, predicate);
+		public static ChannelReader<TOut> Transform<TIn, TOut>(this ChannelReader<TIn> source, Func<TIn, TOut> transform)
+			=> new TransformingChannelReader<TIn, TOut>(source, transform);
 	}
 }

Open.ChannelExtensions/Extensions.TypeFilter.cs

@@ -0,0 +1,44 @@
+using System;
+using System.Diagnostics.Contracts;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+
+namespace Open.ChannelExtensions
+{
+	public static partial class Extensions
+	{
+		class TypeFilteringChannelReader<TSource, T> : ChannelReader<T>
+		{
+			public TypeFilteringChannelReader(ChannelReader<TSource> source)
+			{
+				_source = source ?? throw new ArgumentNullException(nameof(source));
+				Contract.EndContractBlock();
+			}
+
+			private readonly ChannelReader<TSource> _source;
+			public override Task Completion => _source.Completion;
+
+			public override bool TryRead(out T item)
+			{
+				while (_source.TryRead(out TSource s))
+				{
+					if(s is T i)
+					{
+						item = i;
+						return true;
+					}
+				}
+
+				item = default;
+				return false;
+			}
+
+			public override ValueTask<bool> WaitToReadAsync(CancellationToken cancellationToken = default)
+				=> _source.WaitToReadAsync(cancellationToken);
+		}
+
+		public static ChannelReader<T> OfType<TSource, T>(this ChannelReader<TSource> source)
+			=> new TypeFilteringChannelReader<TSource, T>(source);
+	}
+}