using System;

using System.Linq;

using System.Reactive.Concurrency;

using System.Reactive.Disposables;

using System.Reactive.Linq;

using System.Reactive.Threading.Tasks;

using System.Threading;

using System.Threading.Tasks;

​

namespace ConsoleApplication10

{

    public class Program

    {

        public static void Main(string[] args)

        {

            Run().Wait();

        }

​

        async static Task Run()

        {

            var userId = await GetUserId("Peter");

            var invoices = ReadInvoicesFromDb(userId).Where(invoice => invoice.Total == 0) // filter by synchronous predicate

                                                     .WhereAsync(invoice => IsInvoiceProcessed(invoice.Id)); // filter by asynchronous predicate

​

            try

            {               

                await invoices.ForEachAsync2(async invoice => {

                    Console.WriteLine("Processing invoice: " + invoice.Id);

                    await Task.Delay(TimeSpan.FromMilliseconds(500));

                    if (invoice.Id == 7)

                        throw new Exception();

                    Console.WriteLine("Processed invoice: " + invoice.Id); 

                }); 

            } catch {

                Console.WriteLine("Exception");

            }

        }

​

        static IObservable<Invoice> ReadInvoicesFromDb(string userId)

        {

            return Observable.Create<Invoice>(async obs => {

                for (; ; ) {

                    var batch = await Task.FromResult(Enumerable.Range(0, 10).Select(i => new Invoice() { Id = i }).ToList());

​

                    if (batch.Count == 0)

                        break;

​

                    foreach (var invoice in batch) {

                        var inv = invoice;

                        Console.WriteLine("OnNext " + inv.Id); 

                        obs.OnNext(inv); // producer posts item and immediately continues

                                         // appropriate scheduler will take care of the rest

                        await Task.Delay(TimeSpan.FromMilliseconds(50)); //simulate loading

                    }

​

                    break; //we don't need more for now

                }

                //Subscriber will have enough time to run

                await Task.Delay(TimeSpan.FromMilliseconds(500));

​

                Console.WriteLine("OnCompleted");

                obs.OnCompleted();

​

                return Disposable.Empty;

            });

        }

​

        private static Task ProcessInvoice(Invoice invoice)

        {

            return Task.FromResult(true);

        }

​

        private static Task UpdateInvoiceTotal(int id, int total)

        {

            return Task.FromResult(true);

        }

​

        private static int CalculateTotal(Invoice invoice)

        {

            Thread.Sleep(50 * invoice.Id); //CPU intensive

            return 10;

        }

​

        private static async Task<bool> IsInvoiceProcessed(int id)

        {

            await Task.Delay(TimeSpan.FromMilliseconds(100));

            return id % 2 == 0;

        }

​

        private static Task<string> GetUserId(object name)

        {

            return Task.FromResult("Pan");

        }

    }

​

    internal class Invoice

    {

        public decimal Total { get; set; }

        public int Id { get; set; }

    }

​

    static class RxExtensions

    {

        public static IObservable<T> WhereAsync<T>(this IObservable<T> source, Func<T, Task<bool>> asyncPredicate)

        {

            return source.SelectMany(item => asyncPredicate(item)

                                                .ToObservable()

                                                .Where(res => !res)

                                                .Select(_ => item));

        }

        public static Task ForEachAsync2<T>(this IObservable<T> source, Func<T, Task> handler) 

        {

            return source.Select(item => Observable.FromAsync(() => handler(item)))

                         .Concat()

                         .ToTask();

        }

    }

}

​