123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381 |
- // Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information.
- using System;
- using System.Collections.Generic;
- using System.Linq;
- using System.Threading;
- using System.Threading.Tasks;
- namespace System.Linq
- {
- public static partial class AsyncEnumerable
- {
- public static IAsyncEnumerable<TSource> Catch<TSource, TException>(this IAsyncEnumerable<TSource> source, Func<TException, IAsyncEnumerable<TSource>> handler)
- where TException : Exception
- {
- if (source == null)
- throw new ArgumentNullException("source");
- if (handler == null)
- throw new ArgumentNullException("handler");
- return Create(() =>
- {
- var e = source.GetEnumerator();
- var cts = new CancellationTokenDisposable();
- var a = new AssignableDisposable { Disposable = e };
- var d = new CompositeDisposable(cts, a);
- var done = false;
- var f = default(Action<TaskCompletionSource<bool>, CancellationToken>);
- f = (tcs, ct) =>
- {
- if (!done)
- {
- e.MoveNext(ct).ContinueWith(t =>
- {
- t.Handle(tcs,
- res =>
- {
- tcs.TrySetResult(res);
- },
- ex =>
- {
- var err = default(IAsyncEnumerator<TSource>);
- try
- {
- ex.Flatten().Handle(ex_ =>
- {
- var exx = ex_ as TException;
- if (exx != null)
- {
- err = handler(exx).GetEnumerator();
- return true;
- }
- return false;
- });
- }
- catch (Exception ex2)
- {
- tcs.TrySetException(ex2);
- return;
- }
- if (err != null)
- {
- e = err;
- a.Disposable = e;
- done = true;
- f(tcs, ct);
- }
- }
- );
- });
- }
- else
- {
- e.MoveNext(ct).ContinueWith(t =>
- {
- t.Handle(tcs, res =>
- {
- tcs.TrySetResult(res);
- });
- });
- }
- };
- return Create(
- (ct, tcs) =>
- {
- f(tcs, cts.Token);
- return tcs.Task.UsingEnumerator(a);
- },
- () => e.Current,
- d.Dispose
- );
- });
- }
- public static IAsyncEnumerable<TSource> Catch<TSource>(this IEnumerable<IAsyncEnumerable<TSource>> sources)
- {
- if (sources == null)
- throw new ArgumentNullException("sources");
- return sources.Catch_();
- }
- public static IAsyncEnumerable<TSource> Catch<TSource>(params IAsyncEnumerable<TSource>[] sources)
- {
- if (sources == null)
- throw new ArgumentNullException("sources");
- return sources.Catch_();
- }
- public static IAsyncEnumerable<TSource> Catch<TSource>(this IAsyncEnumerable<TSource> first, IAsyncEnumerable<TSource> second)
- {
- if (first == null)
- throw new ArgumentNullException("first");
- if (second == null)
- throw new ArgumentNullException("second");
- return new[] { first, second }.Catch_();
- }
- private static IAsyncEnumerable<TSource> Catch_<TSource>(this IEnumerable<IAsyncEnumerable<TSource>> sources)
- {
- return Create(() =>
- {
- var se = sources.GetEnumerator();
- var e = default(IAsyncEnumerator<TSource>);
- var cts = new CancellationTokenDisposable();
- var a = new AssignableDisposable();
- var d = new CompositeDisposable(cts, se, a);
- var error = default(Exception);
- var f = default(Action<TaskCompletionSource<bool>, CancellationToken>);
- f = (tcs, ct) =>
- {
- if (e == null)
- {
- var b = false;
- try
- {
- b = se.MoveNext();
- if (b)
- e = se.Current.GetEnumerator();
- }
- catch (Exception ex)
- {
- tcs.TrySetException(ex);
- return;
- }
- if (!b)
- {
- if (error != null)
- {
- tcs.TrySetException(error);
- return;
- }
- tcs.TrySetResult(false);
- return;
- }
- error = null;
- a.Disposable = e;
- }
- e.MoveNext(ct).ContinueWith(t =>
- {
- t.Handle(tcs,
- res =>
- {
- tcs.TrySetResult(res);
- },
- ex =>
- {
- e.Dispose();
- e = null;
- error = ex;
- f(tcs, ct);
- }
- );
- });
- };
- return Create(
- (ct, tcs) =>
- {
- f(tcs, cts.Token);
- return tcs.Task.UsingEnumerator(a);
- },
- () => e.Current,
- d.Dispose
- );
- });
- }
- public static IAsyncEnumerable<TSource> Finally<TSource>(this IAsyncEnumerable<TSource> source, Action finallyAction)
- {
- if (source == null)
- throw new ArgumentNullException("source");
- if (finallyAction == null)
- throw new ArgumentNullException("finallyAction");
- return Create(() =>
- {
- var e = source.GetEnumerator();
- var cts = new CancellationTokenDisposable();
- var r = new Disposable(finallyAction);
- var d = new CompositeDisposable(cts, e, r);
- var f = default(Action<TaskCompletionSource<bool>, CancellationToken>);
- f = (tcs, ct) =>
- {
- e.MoveNext(ct).ContinueWith(t =>
- {
- t.Handle(tcs, res =>
- {
- tcs.TrySetResult(res);
- });
- });
- };
- return Create(
- (ct, tcs) =>
- {
- f(tcs, cts.Token);
- return tcs.Task.UsingEnumeratorSync(r);
- },
- () => e.Current,
- d.Dispose
- );
- });
- }
- public static IAsyncEnumerable<TSource> OnErrorResumeNext<TSource>(this IAsyncEnumerable<TSource> first, IAsyncEnumerable<TSource> second)
- {
- if (first == null)
- throw new ArgumentNullException("first");
- if (second == null)
- throw new ArgumentNullException("second");
- return OnErrorResumeNext_(new[] { first, second });
- }
- public static IAsyncEnumerable<TSource> OnErrorResumeNext<TSource>(params IAsyncEnumerable<TSource>[] sources)
- {
- if (sources == null)
- throw new ArgumentNullException("sources");
- return OnErrorResumeNext_(sources);
- }
- public static IAsyncEnumerable<TSource> OnErrorResumeNext<TSource>(this IEnumerable<IAsyncEnumerable<TSource>> sources)
- {
- if (sources == null)
- throw new ArgumentNullException("sources");
- return OnErrorResumeNext_(sources);
- }
- private static IAsyncEnumerable<TSource> OnErrorResumeNext_<TSource>(IEnumerable<IAsyncEnumerable<TSource>> sources)
- {
- return Create(() =>
- {
- var se = sources.GetEnumerator();
- var e = default(IAsyncEnumerator<TSource>);
- var cts = new CancellationTokenDisposable();
- var a = new AssignableDisposable();
- var d = new CompositeDisposable(cts, se, a);
- var f = default(Action<TaskCompletionSource<bool>, CancellationToken>);
- f = (tcs, ct) =>
- {
- if (e == null)
- {
- var b = false;
- try
- {
- b = se.MoveNext();
- if (b)
- e = se.Current.GetEnumerator();
- }
- catch (Exception ex)
- {
- tcs.TrySetException(ex);
- return;
- }
- if (!b)
- {
- tcs.TrySetResult(false);
- return;
- }
- a.Disposable = e;
- }
- e.MoveNext(ct).ContinueWith(t =>
- {
- t.Handle(tcs,
- res =>
- {
- if (res)
- {
- tcs.TrySetResult(true);
- }
- else
- {
- e.Dispose();
- e = null;
- f(tcs, ct);
- }
- },
- ex =>
- {
- e.Dispose();
- e = null;
- f(tcs, ct);
- }
- );
- });
- };
- return Create(
- (ct, tcs) =>
- {
- f(tcs, cts.Token);
- return tcs.Task.UsingEnumerator(a);
- },
- () => e.Current,
- d.Dispose
- );
- });
- }
- public static IAsyncEnumerable<TSource> Retry<TSource>(this IAsyncEnumerable<TSource> source)
- {
- if (source == null)
- throw new ArgumentNullException("source");
- return new[] { source }.Repeat().Catch();
- }
- public static IAsyncEnumerable<TSource> Retry<TSource>(this IAsyncEnumerable<TSource> source, int retryCount)
- {
- if (source == null)
- throw new ArgumentNullException("source");
- if (retryCount < 0)
- throw new ArgumentOutOfRangeException("retryCount");
- return new[] { source }.Repeat(retryCount).Catch();
- }
- private static IEnumerable<TSource> Repeat<TSource>(this IEnumerable<TSource> source)
- {
- while (true)
- foreach (var item in source)
- yield return item;
- }
- private static IEnumerable<TSource> Repeat<TSource>(this IEnumerable<TSource> source, int count)
- {
- for (var i = 0; i < count; i++)
- foreach (var item in source)
- yield return item;
- }
- }
- }
|