{"cells":[{"cell_type":"markdown","metadata":{},"source":["# AsyncSeq (Polyglot)"]},{"cell_type":"code","execution_count":1,"metadata":{"duration_ms":9258,"language":"fsharp"},"outputs":[],"source":["#!import ../../lib/fsharp/Notebooks.livemd"]},{"cell_type":"code","execution_count":2,"metadata":{"duration_ms":192,"language":"fsharp"},"outputs":[],"source":["#!import ../../lib/fsharp/Testing.livemd"]},{"cell_type":"code","execution_count":3,"metadata":{"duration_ms":679,"language":"fsharp"},"outputs":[],"source":["#r \"nuget: fsharp.control.asyncseq, 3.2.1\"\n","#r \"nuget: system.reactive, 6.0.1-preview.1\"\n","#r \"nuget: system.reactive.linq, 6.0.1-preview.1\""]},{"cell_type":"code","execution_count":4,"metadata":{"duration_ms":38,"language":"fsharp"},"outputs":[],"source":["#!import ../../lib/fsharp/Common.fs"]},{"cell_type":"code","execution_count":5,"metadata":{"duration_ms":219,"language":"fsharp"},"outputs":[],"source":["#!import ../../lib/fsharp/Async.fs"]},{"cell_type":"code","execution_count":6,"metadata":{"duration_ms":10,"language":"fsharp"},"outputs":[],"source":["#if !INTERACTIVE\n","open Lib\n","#endif"]},{"cell_type":"code","execution_count":7,"metadata":{"duration_ms":13,"language":"fsharp"},"outputs":[],"source":["open Common"]},{"cell_type":"markdown","metadata":{},"source":["## subscribeEvent"]},{"cell_type":"code","execution_count":8,"metadata":{"duration_ms":39,"language":"fsharp"},"outputs":[],"source":["let inline subscribeEvent (event: IEvent<'H, 'A>) map =\n","    let observable = System.Reactive.Linq.Observable.FromEventPattern<'H, 'A>(event.AddHandler, event.RemoveHandler)\n","    System.Reactive.Linq.Observable.Select (observable, fun event -> map event.EventArgs)\n","    |> FSharp.Control.AsyncSeq.ofObservableBuffered"]},{"cell_type":"code","execution_count":9,"metadata":{"duration_ms":78,"language":"fsharp"},"outputs":[],"source":["///- --test\n","\n","type TestEvent () as self =\n","    member val Calls = [] with get, set\n","    member val Event = Event<ErrorEventHandler, ErrorEventArgs> () with get\n","\n","    member _.AddCall text =\n","        self.Calls <- self.Calls @ [ text ]\n","\n","    member _.EventInterface =\n","        { new IEvent<ErrorEventHandler, ErrorEventArgs> with\n","            member _.AddHandler handler =\n","                self.AddCall \"AddHandler\"\n","                self.Event.Publish.AddHandler handler\n","\n","            member _.RemoveHandler handler =\n","                self.AddCall \"RemoveHandler\"\n","                self.Event.Publish.RemoveHandler handler\n","\n","            member _.Subscribe observer =\n","                self.AddCall \"IObservable.Subscribe\"\n","                let disposable = self.Event.Publish.Subscribe observer\n","                new_disposable (fun () ->\n","                    self.AddCall \"IObservable.Dispose\"\n","                    disposable.Dispose ()\n","                )\n","        }\n","\n","    member _.Subscribe () =\n","        subscribeEvent\n","            self.EventInterface\n","            (fun args ->\n","                let result = args.GetException () |> SpiralSm.format_exception\n","                self.AddCall $\"TestEvent.Subscribe({result})\"\n","                result\n","            )\n","\n","    member _.Iter subscription =\n","        subscription\n","        |> FSharp.Control.AsyncSeq.iteriAsync (fun i error -> async {\n","            self.AddCall $\"TestEvent.Iter({i}: {error})\"\n","        })\n","\n","    member _.WaitCall text = async {\n","        while self.Calls |> List.last <> text do\n","            do! Async.SwitchToThreadPool ()\n","    }"]},{"cell_type":"code","execution_count":10,"metadata":{"duration_ms":407,"language":"fsharp"},"outputs":[{"name":"stdout","output_type":"stream","text":["11:53:49 \u001B[94md\u001B[0m #1 Async.runWithTimeoutAsync / timeout: 300\n","<null>\n","\n","[\"1\"; \"AddHandler\"; \"2\"; \"RemoveHandler\"]"]}],"source":["///- --test\n","\n","let testEvent = TestEvent ()\n","\n","async {\n","    testEvent.AddCall \"1\"\n","    let! child = testEvent.Subscribe () |> testEvent.Iter |> Async.StartChild\n","    do! testEvent.WaitCall \"AddHandler\"\n","    testEvent.AddCall \"2\"\n","    do! child\n","    testEvent.AddCall \"3\"\n","}\n","|> Async.runWithTimeout 300\n","|> _assertEqual None\n","\n","testEvent.Calls\n","|> Seq.toList\n","|> _assertEqual [ \"1\"; \"AddHandler\"; \"2\"; \"RemoveHandler\" ]"]},{"cell_type":"code","execution_count":11,"metadata":{"duration_ms":395,"language":"fsharp"},"outputs":[{"name":"stdout","output_type":"stream","text":["11:53:49 \u001B[94md\u001B[0m #2 Async.runWithTimeoutAsync / timeout: 300\n","<null>\n","\n","[\"1\"; \"AddHandler\"; \"2\"; \"IObservable.Subscribe\"; \"3\"; \"RemoveHandler\"; \"IObservable.Dispose\"]"]}],"source":["///- --test\n","\n","let testEvent = TestEvent ()\n","\n","async {\n","    testEvent.AddCall \"1\"\n","    let! child = testEvent.Subscribe () |> testEvent.Iter |> Async.StartChild\n","    do! testEvent.WaitCall \"AddHandler\"\n","    testEvent.AddCall \"2\"\n","    use _ = testEvent.EventInterface.Subscribe (fun args ->\n","        testEvent.AddCall $\"testEvent.EventInterface.Subscribe({args})\"\n","    )\n","    testEvent.AddCall \"3\"\n","    do! child\n","    testEvent.AddCall \"4\"\n","}\n","|> Async.runWithTimeout 300\n","|> _assertEqual None\n","\n","testEvent.Calls\n","|> _assertEqual [ \"1\"; \"AddHandler\"; \"2\"; \"IObservable.Subscribe\"; \"3\"; \"RemoveHandler\"; \"IObservable.Dispose\" ]"]},{"cell_type":"code","execution_count":12,"metadata":{"duration_ms":387,"language":"fsharp"},"outputs":[{"name":"stdout","output_type":"stream","text":["11:53:50 \u001B[94md\u001B[0m #3 Async.runWithTimeoutAsync / timeout: 300\n","<null>\n","\n","[\"1\"; \"AddHandler\"; \"2\"; \"IObservable.Subscribe\"; \"3\"; \"TestEvent.Subscribe(System.Exception: error)\";\n"," \"TestEvent.Iter(0: System.Exception: error)\"; \"testEvent.EventInterface.Subscribe(System.Exception: error)\"; \"4\";\n"," \"RemoveHandler\"; \"IObservable.Dispose\"]"]}],"source":["///- --test\n","\n","let testEvent = TestEvent ()\n","\n","async {\n","    testEvent.AddCall \"1\"\n","    let! child = testEvent.Subscribe () |> testEvent.Iter |> Async.StartChild\n","    do! testEvent.WaitCall \"AddHandler\"\n","    testEvent.AddCall \"2\"\n","    use _ = testEvent.EventInterface.Subscribe (fun args ->\n","        async {\n","            do! testEvent.WaitCall \"TestEvent.Iter(0: System.Exception: error)\"\n","            testEvent.AddCall $\"testEvent.EventInterface.Subscribe({args.GetException () |> SpiralSm.format_exception})\"\n","        }\n","        |> Async.RunSynchronously\n","    )\n","    testEvent.AddCall \"3\"\n","    testEvent.Event.Trigger (null, ErrorEventArgs (Exception \"error\"))\n","    testEvent.AddCall \"4\"\n","    do! child\n","    testEvent.AddCall \"5\"\n","}\n","|> Async.runWithTimeout 300\n","|> _assertEqual None\n","\n","testEvent.Calls\n","|> _assertEqual [\n","    \"1\"\n","    \"AddHandler\"\n","    \"2\"\n","    \"IObservable.Subscribe\"\n","    \"3\"\n","    \"TestEvent.Subscribe(System.Exception: error)\"\n","    \"TestEvent.Iter(0: System.Exception: error)\"\n","    \"testEvent.EventInterface.Subscribe(System.Exception: error)\"\n","    \"4\"\n","    \"RemoveHandler\"\n","    \"IObservable.Dispose\"\n","]"]},{"cell_type":"markdown","metadata":{},"source":["## subscribeToken"]},{"cell_type":"code","execution_count":13,"metadata":{"duration_ms":41,"language":"fsharp"},"outputs":[],"source":["let subscribeToken (token : System.Threading.CancellationToken) =\n","    let tcs = new System.Threading.Tasks.TaskCompletionSource ()\n","    System.Action tcs.SetResult |> token.Register |> ignore\n","    let start = System.DateTime.Now.Ticks\n","    FSharp.Control.AsyncSeq.unfoldAsync\n","        (fun () -> async {\n","            do! tcs.Task |> Async.AwaitTask\n","            return Some (System.DateTime.Now.Ticks - start, ())\n","        })\n","        ()"]},{"cell_type":"code","execution_count":14,"metadata":{"duration_ms":152,"language":"fsharp"},"outputs":[{"name":"stdout","output_type":"stream","text":["true"]}],"source":["///- --test\n","\n","let cts = new System.Threading.CancellationTokenSource ()\n","\n","async {\n","    let! child =\n","        cts.Token\n","        |> subscribeToken\n","        |> FSharp.Control.AsyncSeq.tryFirst\n","        |> Async.StartChild\n","\n","    do! Async.Sleep 100\n","    cts.Cancel ()\n","    return! child\n","}\n","|> Async.RunSynchronously\n","|> Option.get\n","|> fun x -> x > 900000\n","|> _assertEqual true"]}],"metadata":{"kernelspec":{"display_name":"Spiral","language":"spiral","name":"spiral"}},"nbformat":4,"nbformat_minor":5}