// Licensed to the .NET Foundation under one or more agreements. // The .NET Foundation licenses this file to you under the Apache 2.0 License. // See the LICENSE file in the project root for more information. using System.Reactive.Concurrency; using System.Reactive.Subjects; namespace System.Reactive.Linq { // REVIEW: Consider if these are worth retaining in the async space. partial class AsyncObservable { public static Func> ToAsync(Func function) { if (function == null) throw new ArgumentNullException(nameof(function)); return ToAsync(function, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Func function, IAsyncScheduler scheduler) { if (function == null) throw new ArgumentNullException(nameof(function)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return () => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { TResult res; try { res = function(); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(res).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Func function) { if (function == null) throw new ArgumentNullException(nameof(function)); return ToAsync(function, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Func function, IAsyncScheduler scheduler) { if (function == null) throw new ArgumentNullException(nameof(function)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { TResult res; try { res = function(arg1); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(res).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Func function) { if (function == null) throw new ArgumentNullException(nameof(function)); return ToAsync(function, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Func function, IAsyncScheduler scheduler) { if (function == null) throw new ArgumentNullException(nameof(function)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { TResult res; try { res = function(arg1, arg2); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(res).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Func function) { if (function == null) throw new ArgumentNullException(nameof(function)); return ToAsync(function, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Func function, IAsyncScheduler scheduler) { if (function == null) throw new ArgumentNullException(nameof(function)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { TResult res; try { res = function(arg1, arg2, arg3); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(res).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Func function) { if (function == null) throw new ArgumentNullException(nameof(function)); return ToAsync(function, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Func function, IAsyncScheduler scheduler) { if (function == null) throw new ArgumentNullException(nameof(function)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { TResult res; try { res = function(arg1, arg2, arg3, arg4); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(res).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Func function) { if (function == null) throw new ArgumentNullException(nameof(function)); return ToAsync(function, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Func function, IAsyncScheduler scheduler) { if (function == null) throw new ArgumentNullException(nameof(function)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4, arg5) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { TResult res; try { res = function(arg1, arg2, arg3, arg4, arg5); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(res).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Func function) { if (function == null) throw new ArgumentNullException(nameof(function)); return ToAsync(function, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Func function, IAsyncScheduler scheduler) { if (function == null) throw new ArgumentNullException(nameof(function)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4, arg5, arg6) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { TResult res; try { res = function(arg1, arg2, arg3, arg4, arg5, arg6); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(res).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Func function) { if (function == null) throw new ArgumentNullException(nameof(function)); return ToAsync(function, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Func function, IAsyncScheduler scheduler) { if (function == null) throw new ArgumentNullException(nameof(function)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4, arg5, arg6, arg7) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { TResult res; try { res = function(arg1, arg2, arg3, arg4, arg5, arg6, arg7); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(res).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Func function) { if (function == null) throw new ArgumentNullException(nameof(function)); return ToAsync(function, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Func function, IAsyncScheduler scheduler) { if (function == null) throw new ArgumentNullException(nameof(function)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { TResult res; try { res = function(arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(res).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Func function) { if (function == null) throw new ArgumentNullException(nameof(function)); return ToAsync(function, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Func function, IAsyncScheduler scheduler) { if (function == null) throw new ArgumentNullException(nameof(function)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { TResult res; try { res = function(arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(res).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Func function) { if (function == null) throw new ArgumentNullException(nameof(function)); return ToAsync(function, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Func function, IAsyncScheduler scheduler) { if (function == null) throw new ArgumentNullException(nameof(function)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { TResult res; try { res = function(arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(res).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Func function) { if (function == null) throw new ArgumentNullException(nameof(function)); return ToAsync(function, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Func function, IAsyncScheduler scheduler) { if (function == null) throw new ArgumentNullException(nameof(function)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { TResult res; try { res = function(arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(res).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Func function) { if (function == null) throw new ArgumentNullException(nameof(function)); return ToAsync(function, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Func function, IAsyncScheduler scheduler) { if (function == null) throw new ArgumentNullException(nameof(function)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11, arg12) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { TResult res; try { res = function(arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11, arg12); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(res).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Func function) { if (function == null) throw new ArgumentNullException(nameof(function)); return ToAsync(function, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Func function, IAsyncScheduler scheduler) { if (function == null) throw new ArgumentNullException(nameof(function)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11, arg12, arg13) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { TResult res; try { res = function(arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11, arg12, arg13); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(res).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Func function) { if (function == null) throw new ArgumentNullException(nameof(function)); return ToAsync(function, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Func function, IAsyncScheduler scheduler) { if (function == null) throw new ArgumentNullException(nameof(function)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11, arg12, arg13, arg14) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { TResult res; try { res = function(arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11, arg12, arg13, arg14); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(res).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Func function) { if (function == null) throw new ArgumentNullException(nameof(function)); return ToAsync(function, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Func function, IAsyncScheduler scheduler) { if (function == null) throw new ArgumentNullException(nameof(function)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11, arg12, arg13, arg14, arg15) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { TResult res; try { res = function(arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11, arg12, arg13, arg14, arg15); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(res).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Func function) { if (function == null) throw new ArgumentNullException(nameof(function)); return ToAsync(function, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Func function, IAsyncScheduler scheduler) { if (function == null) throw new ArgumentNullException(nameof(function)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11, arg12, arg13, arg14, arg15, arg16) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { TResult res; try { res = function(arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11, arg12, arg13, arg14, arg15, arg16); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(res).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Action action) { if (action == null) throw new ArgumentNullException(nameof(action)); return ToAsync(action, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Action action, IAsyncScheduler scheduler) { if (action == null) throw new ArgumentNullException(nameof(action)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return () => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { try { action(); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(Unit.Default).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Action action) { if (action == null) throw new ArgumentNullException(nameof(action)); return ToAsync(action, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Action action, IAsyncScheduler scheduler) { if (action == null) throw new ArgumentNullException(nameof(action)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { try { action(arg1); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(Unit.Default).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Action action) { if (action == null) throw new ArgumentNullException(nameof(action)); return ToAsync(action, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Action action, IAsyncScheduler scheduler) { if (action == null) throw new ArgumentNullException(nameof(action)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { try { action(arg1, arg2); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(Unit.Default).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Action action) { if (action == null) throw new ArgumentNullException(nameof(action)); return ToAsync(action, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Action action, IAsyncScheduler scheduler) { if (action == null) throw new ArgumentNullException(nameof(action)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { try { action(arg1, arg2, arg3); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(Unit.Default).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Action action) { if (action == null) throw new ArgumentNullException(nameof(action)); return ToAsync(action, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Action action, IAsyncScheduler scheduler) { if (action == null) throw new ArgumentNullException(nameof(action)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { try { action(arg1, arg2, arg3, arg4); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(Unit.Default).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Action action) { if (action == null) throw new ArgumentNullException(nameof(action)); return ToAsync(action, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Action action, IAsyncScheduler scheduler) { if (action == null) throw new ArgumentNullException(nameof(action)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4, arg5) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { try { action(arg1, arg2, arg3, arg4, arg5); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(Unit.Default).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Action action) { if (action == null) throw new ArgumentNullException(nameof(action)); return ToAsync(action, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Action action, IAsyncScheduler scheduler) { if (action == null) throw new ArgumentNullException(nameof(action)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4, arg5, arg6) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { try { action(arg1, arg2, arg3, arg4, arg5, arg6); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(Unit.Default).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Action action) { if (action == null) throw new ArgumentNullException(nameof(action)); return ToAsync(action, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Action action, IAsyncScheduler scheduler) { if (action == null) throw new ArgumentNullException(nameof(action)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4, arg5, arg6, arg7) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { try { action(arg1, arg2, arg3, arg4, arg5, arg6, arg7); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(Unit.Default).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Action action) { if (action == null) throw new ArgumentNullException(nameof(action)); return ToAsync(action, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Action action, IAsyncScheduler scheduler) { if (action == null) throw new ArgumentNullException(nameof(action)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { try { action(arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(Unit.Default).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Action action) { if (action == null) throw new ArgumentNullException(nameof(action)); return ToAsync(action, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Action action, IAsyncScheduler scheduler) { if (action == null) throw new ArgumentNullException(nameof(action)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { try { action(arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(Unit.Default).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Action action) { if (action == null) throw new ArgumentNullException(nameof(action)); return ToAsync(action, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Action action, IAsyncScheduler scheduler) { if (action == null) throw new ArgumentNullException(nameof(action)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { try { action(arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(Unit.Default).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Action action) { if (action == null) throw new ArgumentNullException(nameof(action)); return ToAsync(action, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Action action, IAsyncScheduler scheduler) { if (action == null) throw new ArgumentNullException(nameof(action)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { try { action(arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(Unit.Default).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Action action) { if (action == null) throw new ArgumentNullException(nameof(action)); return ToAsync(action, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Action action, IAsyncScheduler scheduler) { if (action == null) throw new ArgumentNullException(nameof(action)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11, arg12) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { try { action(arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11, arg12); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(Unit.Default).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Action action) { if (action == null) throw new ArgumentNullException(nameof(action)); return ToAsync(action, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Action action, IAsyncScheduler scheduler) { if (action == null) throw new ArgumentNullException(nameof(action)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11, arg12, arg13) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { try { action(arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11, arg12, arg13); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(Unit.Default).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Action action) { if (action == null) throw new ArgumentNullException(nameof(action)); return ToAsync(action, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Action action, IAsyncScheduler scheduler) { if (action == null) throw new ArgumentNullException(nameof(action)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11, arg12, arg13, arg14) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { try { action(arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11, arg12, arg13, arg14); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(Unit.Default).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Action action) { if (action == null) throw new ArgumentNullException(nameof(action)); return ToAsync(action, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Action action, IAsyncScheduler scheduler) { if (action == null) throw new ArgumentNullException(nameof(action)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11, arg12, arg13, arg14, arg15) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { try { action(arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11, arg12, arg13, arg14, arg15); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(Unit.Default).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } public static Func> ToAsync(Action action) { if (action == null) throw new ArgumentNullException(nameof(action)); return ToAsync(action, TaskPoolAsyncScheduler.Default); } public static Func> ToAsync(Action action, IAsyncScheduler scheduler) { if (action == null) throw new ArgumentNullException(nameof(action)); if (scheduler == null) throw new ArgumentNullException(nameof(scheduler)); return (arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11, arg12, arg13, arg14, arg15, arg16) => { var subject = new SequentialAsyncAsyncSubject(); // NB: We don't do anything with the result of scheduling the action; it can't be cancelled. scheduler.ScheduleAsync(async ct => { try { action(arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11, arg12, arg13, arg14, arg15, arg16); } catch (Exception ex) { await subject.OnErrorAsync(ex).RendezVous(scheduler); return; } await subject.OnNextAsync(Unit.Default).RendezVous(scheduler); await subject.OnCompletedAsync().RendezVous(scheduler); }); return subject.AsAsyncObservable(); }; } } }