Add PipesExamples project which is part of a potential refactor of Pipes

LanguageExt.Core/Traits/Foldable/Foldable.Extensions.cs

@@ -347,7 +347,7 @@ public static K<M, S> FoldM<T, A, M, S>(
     /// <remarks>
     /// Note that to produce the outermost application of the operator the
     /// entire input list must be traversed.  Like all left-associative folds,
-    /// `FoldBack' will diverge if given an infinite list.
+    /// `FoldBack` will diverge if given an infinite list.
     /// </remarks>
     public static S FoldBack<T, A, S>(this K<T, A> ta, S initialState, Func<S, Func<A, S>> f) 
         where T : Foldable<T> =>
@@ -366,7 +366,7 @@ public static S FoldBack<T, A, S>(this K<T, A> ta, S initialState, Func<S, Func<
     /// <remarks>
     /// Note that to produce the outermost application of the operator the
     /// entire input list must be traversed.  Like all left-associative folds,
-    /// `FoldBack' will diverge if given an infinite list.
+    /// `FoldBack` will diverge if given an infinite list.
     /// </remarks>
     public static S FoldBack<T, A, S>(this K<T, A> ta, S initialState, Func<S, A, S> f) 
         where T : Foldable<T> =>
@@ -385,7 +385,7 @@ public static S FoldBack<T, A, S>(this K<T, A> ta, S initialState, Func<S, A, S>
     /// <remarks>
     /// Note that to produce the outermost application of the operator the
     /// entire input list must be traversed.  Like all left-associative folds,
-    /// `FoldBack' will diverge if given an infinite list.
+    /// `FoldBack` will diverge if given an infinite list.
     /// </remarks>
     public static K<M, S> FoldBackM<T, A, M, S>(
         this K<T, A> ta, 
@@ -408,7 +408,7 @@ public static K<M, S> FoldBackM<T, A, M, S>(
     /// <remarks>
     /// Note that to produce the outermost application of the operator the
     /// entire input list must be traversed.  Like all left-associative folds,
-    /// `FoldBack' will diverge if given an infinite list.
+    /// `FoldBack` will diverge if given an infinite list.
     /// </remarks>
     public static K<M, S> FoldBackM<T, A, M, S>(
         this K<T, A> ta, 
@@ -422,7 +422,7 @@ public static K<M, S> FoldBackM<T, A, M, S>(
     /// Given a structure with elements whose type is a `Monoid`, combine them
     /// via the monoid's `Append` operator.  This fold is right-associative and
     /// lazy in the accumulator.  When you need a strict left-associative fold,
-    /// use 'foldMap'' instead, with 'id' as the map.
+    /// use `FoldMap` instead, with `identity` as the map.
     /// </summary>
     public static A Fold<T, A>(this K<T, A> tm) 
         where T : Foldable<T>
@@ -433,7 +433,7 @@ public static A Fold<T, A>(this K<T, A> tm)
     /// Given a structure with elements whose type is a `Monoid`, combine them
     /// via the monoid's `Append` operator.  This fold is right-associative and
     /// lazy in the accumulator.  When you need a strict left-associative fold,
-    /// use 'foldMap'' instead, with 'id' as the map.
+    /// use `FoldMap` instead, with `identity` as the map.
     /// </summary>
     public static A FoldWhile<T, A>(this K<T, A> tm, Func<(A State, A Value), bool> predicate) 
         where T : Foldable<T>
@@ -444,7 +444,7 @@ public static A FoldWhile<T, A>(this K<T, A> tm, Func<(A State, A Value), bool>
     /// Given a structure with elements whose type is a `Monoid`, combine them
     /// via the monoid's `Append` operator.  This fold is right-associative and
     /// lazy in the accumulator.  When you need a strict left-associative fold,
-    /// use 'foldMap'' instead, with 'id' as the map.
+    /// use `FoldMap` instead, with `identity` as the map.
     /// </summary>
     public static A FoldUntil<T, A>(this K<T, A> tm, Func<(A State, A Value), bool> predicate) 
         where T : Foldable<T>

LanguageExt.Pipes/Effect/Effect.cs

@@ -3,6 +3,7 @@
 using System.Linq;
 using static LanguageExt.Pipes.Proxy;
 using System.Runtime.CompilerServices;
+using LanguageExt.Async.Linq;
 using LanguageExt.Traits;
 
 namespace LanguageExt.Pipes;
@@ -38,19 +39,28 @@ K<M, R> Go(Proxy<Void, Unit, Unit, Void, M, R> p) =>
                 ProxyM<Void, Unit, Unit, Void, M, R> (var mx)    => M.Bind(mx, Go),
                 Pure<Void, Unit, Unit, Void, M, R> (var r)       => M.Pure(r),
                 Iterator<Void, Unit, Unit, Void, M, R> iter      => runIterator(iter, Go),
+                IteratorAsync<Void, Unit, Unit, Void, M, R> iter => runAsyncIterator(iter, Go),
                 Request<Void, Unit, Unit, Void, M, R> (var v, _) => closed<K<M, R>>(v),
                 Respond<Void, Unit, Unit, Void, M, R> (var v, _) => closed<K<M, R>>(v),
                 _                                                => throw new NotSupportedException()
             };
     }
 
-    internal static K<M, R> runIterator<M, R>(
+    static K<M, R> runIterator<M, R>(
         Iterator<Void, Unit, Unit, Void, M, R> iter,
         Func<Proxy<Void, Unit, Unit, Void, M, R>, K<M, R>> go)
         where M : Monad<M> =>
         from _ in iter.Run().Select(e => e.RunEffect()).Actions()
         from r in go(iter.Next())
         select r;
+
+    static K<M, R> runAsyncIterator<M, R>(
+        IteratorAsync<Void, Unit, Unit, Void, M, R> iter,
+        Func<Proxy<Void, Unit, Unit, Void, M, R>, K<M, R>> go)
+        where M : Monad<M> =>
+        from _ in iter.Run().Select(e => e.RunEffect()).Actions()
+        from r in go(iter.Next())
+        select r;
 
     [Pure]
     static Proxy<UOut, UIn, DIn, DOut, N, R> HoistX<UOut, UIn, DIn, DOut, M, N, R>(

LanguageExt.Pipes/Iterator/CoreTypes.cs

@@ -16,6 +16,14 @@ public abstract record Iterator<UOut, UIn, DIn, DOut, M, A>(
     public abstract IEnumerable<Proxy<UOut, UIn, DIn, DOut, M, Unit>> Run();
 }
 
+public abstract record IteratorAsync<UOut, UIn, DIn, DOut, M, A>(
+    Func<Proxy<UOut, UIn, DIn, DOut, M, A>> Next) 
+    : Proxy<UOut, UIn, DIn, DOut, M, A>
+    where M : Monad<M>
+{
+    public abstract IAsyncEnumerable<Proxy<UOut, UIn, DIn, DOut, M, Unit>> Run();
+}
+
 public record IteratorFoldable<UOut, UIn, DIn, DOut, F, X, M, A>(
     K<F, X> Items,
     Func<X, Proxy<UOut, UIn, DIn, DOut, M, Unit>> Yield,
@@ -122,16 +130,16 @@ public record IteratorAsyncEnumerable<UOut, UIn, DIn, DOut, F, X, M, A>(
     IAsyncEnumerable<X> Items,
     Func<X, Proxy<UOut, UIn, DIn, DOut, M, Unit>> Yield,
     Func<Proxy<UOut, UIn, DIn, DOut, M, A>> Next) 
-    : Iterator<UOut, UIn, DIn, DOut, M, A>(Next)
+    : IteratorAsync<UOut, UIn, DIn, DOut, M, A>(Next)
     where M : Monad<M>
     where F : Foldable<F>
 {
     public override Proxy<UOut, UIn, DIn, DOut, M, A> ToProxy() =>
         this;
 
-    public override IEnumerable<Proxy<UOut, UIn, DIn, DOut, M, Unit>> Run()
+    public override async IAsyncEnumerable<Proxy<UOut, UIn, DIn, DOut, M, Unit>> Run()
     {
-        foreach (var item in Items.ToBlockingEnumerable())
+        await foreach (var item in Items)
         {
             yield return Yield(item);
         }

LanguageExt.Pipes/Producer/Producer.cs

@@ -294,12 +294,13 @@ Unit countDown()
             return unit;
         }
 
-        K<M, Unit> runEffect(Producer<OUT, M, Unit> p) =>
+        K<M, ForkIO<Unit>> runEffect(Producer<OUT, M, Unit> p) =>
             (from _1 in p | receive()
              let _2 = countDown()
              select unit)
             .ToEffect()
-            .RunEffect();    
+            .RunEffect()
+            .ForkIO();    
     }
 
     /// <summary>

LanguageExt.Pipes/Proxy.Prelude.cs

@@ -30,12 +30,11 @@ public static Consumer<A, A> awaiting<A>() =>
     public static Producer<A, Unit> yield<A>(A value) =>
         PureProxy.ProducerYield(value);
 
-    // TODO: Decide whether I want to put these back or not
     /// <summary>
     /// Create a queue
     /// </summary>
     /// <remarks>A `Queue` is a `Producer` with an `Enqueue`, and a `Done` to cancel the operation</remarks>
-    // [Pure, MethodImpl(mops)]
+    [Pure, MethodImpl(mops)]
     public static Queue<A, M, Unit> Queue<M, A>() 
         where M : Monad<M>
     {
@@ -75,7 +74,9 @@ public static Producer<X, Unit> yieldAll<X>(IAsyncEnumerable<X> xs) =>
     /// <returns>`Producer`</returns>
     [Pure, MethodImpl(mops)]
     public static Producer<X, Unit> yieldAll<X>(IObservable<X> xs) =>
-        yieldAll(xs.ToAsyncEnumerable(new CancellationToken()));
+        from t in PureProxy.ProducerLiftIO<X, CancellationToken>(IO.token)
+        from r in yieldAll(xs.ToAsyncEnumerable(t))
+        select r;
 
     // TODO: IMPLEMENT TAIL CALLS
     [Pure, MethodImpl(mops)]