Async to IObservable

 1: 
 2: 
 3: 
 4: 
 5: 
 6: 
 7: 
 8: 
 9: 
10: 
11: 
12: 
13: 
14: 
15: 
16: 
17: 
18: 
19: 
20: 
21: 
22: 

open System

type Async<'a> with
  member this.ToObservable() =
    { new IObservable<_> with
        member x.Subscribe(o) =
          if o = null then nullArg "observer"
          let cts = new System.Threading.CancellationTokenSource()
          let invoked = ref 0
          let cancelOrDispose cancel =
            if System.Threading.Interlocked.CompareExchange(invoked, 1, 0) = 0 then
              if cancel then cts.Cancel() else cts.Dispose()
          let wrapper = async {
            try
              try
                let! r = this
                o.OnNext(r)
                o.OnCompleted()
              with e -> o.OnError(e)
            finally cancelOrDispose false }
          Async.StartImmediate(wrapper, cts.Token)
          { new IDisposable with member x.Dispose() = cancelOrDispose true } }

namespace System

Multiple items
type Async
static member AsBeginEnd : computation:('Arg -> Async<'T>) -> ('Arg * AsyncCallback * obj -> IAsyncResult) * (IAsyncResult -> 'T) * (IAsyncResult -> unit)
static member AwaitEvent : event:IEvent<'Del,'T> * ?cancelAction:(unit -> unit) -> Async<'T> (requires delegate and 'Del :> Delegate)
static member AwaitIAsyncResult : iar:IAsyncResult * ?millisecondsTimeout:int -> Async<bool>
static member AwaitTask : task:Task<'T> -> Async<'T>
static member AwaitWaitHandle : waitHandle:WaitHandle * ?millisecondsTimeout:int -> Async<bool>
static member CancelDefaultToken : unit -> unit
static member Catch : computation:Async<'T> -> Async<Choice<'T,exn>>
static member FromBeginEnd : beginAction:(AsyncCallback * obj -> IAsyncResult) * endAction:(IAsyncResult -> 'T) * ?cancelAction:(unit -> unit) -> Async<'T>
static member FromBeginEnd : arg:'Arg1 * beginAction:('Arg1 * AsyncCallback * obj -> IAsyncResult) * endAction:(IAsyncResult -> 'T) * ?cancelAction:(unit -> unit) -> Async<'T>
static member FromBeginEnd : arg1:'Arg1 * arg2:'Arg2 * beginAction:('Arg1 * 'Arg2 * AsyncCallback * obj -> IAsyncResult) * endAction:(IAsyncResult -> 'T) * ?cancelAction:(unit -> unit) -> Async<'T>
static member FromBeginEnd : arg1:'Arg1 * arg2:'Arg2 * arg3:'Arg3 * beginAction:('Arg1 * 'Arg2 * 'Arg3 * AsyncCallback * obj -> IAsyncResult) * endAction:(IAsyncResult -> 'T) * ?cancelAction:(unit -> unit) -> Async<'T>
static member FromContinuations : callback:(('T -> unit) * (exn -> unit) * (OperationCanceledException -> unit) -> unit) -> Async<'T>
static member Ignore : computation:Async<'T> -> Async<unit>
static member OnCancel : interruption:(unit -> unit) -> Async<IDisposable>
static member Parallel : computations:seq<Async<'T>> -> Async<'T []>
static member RunSynchronously : computation:Async<'T> * ?timeout:int * ?cancellationToken:CancellationToken -> 'T
static member Sleep : millisecondsDueTime:int -> Async<unit>
static member Start : computation:Async<unit> * ?cancellationToken:CancellationToken -> unit
static member StartAsTask : computation:Async<'T> * ?taskCreationOptions:TaskCreationOptions * ?cancellationToken:CancellationToken -> Task<'T>
static member StartChild : computation:Async<'T> * ?millisecondsTimeout:int -> Async<Async<'T>>
static member StartChildAsTask : computation:Async<'T> * ?taskCreationOptions:TaskCreationOptions -> Async<Task<'T>>
static member StartImmediate : computation:Async<unit> * ?cancellationToken:CancellationToken -> unit
static member StartWithContinuations : computation:Async<'T> * continuation:('T -> unit) * exceptionContinuation:(exn -> unit) * cancellationContinuation:(OperationCanceledException -> unit) * ?cancellationToken:CancellationToken -> unit
static member SwitchToContext : syncContext:SynchronizationContext -> Async<unit>
static member SwitchToNewThread : unit -> Async<unit>
static member SwitchToThreadPool : unit -> Async<unit>
static member TryCancelled : computation:Async<'T> * compensation:(OperationCanceledException -> unit) -> Async<'T>
static member CancellationToken : Async<CancellationToken>
static member DefaultCancellationToken : CancellationToken

Full name: Microsoft.FSharp.Control.Async

--------------------
type Async<'T>

Full name: Microsoft.FSharp.Control.Async<_>

val this : Async<'T>

member Async.ToObservable : unit -> IObservable<'T>

Full name: Script.ToObservable

type IObservable<'T> =
  member Subscribe : observer:IObserver<'T> -> IDisposable

Full name: System.IObservable<_>

val x : IObservable<'T>

member IObservable.Subscribe : callback:('T -> unit) -> IDisposable
IObservable.Subscribe(observer: IObserver<'T>) : IDisposable

val o : IObserver<'T>

val nullArg : argumentName:string -> 'T

Full name: Microsoft.FSharp.Core.Operators.nullArg

val cts : Threading.CancellationTokenSource

namespace System.Threading

Multiple items
type CancellationTokenSource =
  new : unit -> CancellationTokenSource
  member Cancel : unit -> unit + 1 overload
  member Dispose : unit -> unit
  member IsCancellationRequested : bool
  member Token : CancellationToken
  static member CreateLinkedTokenSource : [<ParamArray>] tokens:CancellationToken[] -> CancellationTokenSource + 1 overload

Full name: System.Threading.CancellationTokenSource

--------------------
Threading.CancellationTokenSource() : unit

val invoked : int ref

Multiple items
val ref : value:'T -> 'T ref

Full name: Microsoft.FSharp.Core.Operators.ref

--------------------
type 'T ref = Ref<'T>

Full name: Microsoft.FSharp.Core.ref<_>

val cancelOrDispose : (bool -> unit)

val cancel : bool

type Interlocked =
  static member Add : location1:int * value:int -> int + 1 overload
  static member CompareExchange : location1:int * value:int * comparand:int -> int + 6 overloads
  static member Decrement : location:int -> int + 1 overload
  static member Exchange : location1:int * value:int -> int + 6 overloads
  static member Increment : location:int -> int + 1 overload
  static member Read : location:int64 -> int64

Full name: System.Threading.Interlocked

Threading.Interlocked.CompareExchange<'T (requires reference type)>(location1: byref<'T>, value: 'T, comparand: 'T) : 'T
Threading.Interlocked.CompareExchange(location1: byref<nativeint>, value: nativeint, comparand: nativeint) : nativeint
Threading.Interlocked.CompareExchange(location1: byref<obj>, value: obj, comparand: obj) : obj
Threading.Interlocked.CompareExchange(location1: byref<float>, value: float, comparand: float) : float
Threading.Interlocked.CompareExchange(location1: byref<float32>, value: float32, comparand: float32) : float32
Threading.Interlocked.CompareExchange(location1: byref<int64>, value: int64, comparand: int64) : int64
Threading.Interlocked.CompareExchange(location1: byref<int>, value: int, comparand: int) : int

Threading.CancellationTokenSource.Cancel() : unit
Threading.CancellationTokenSource.Cancel(throwOnFirstException: bool) : unit

Threading.CancellationTokenSource.Dispose() : unit

val wrapper : Async<unit>

val async : AsyncBuilder

Full name: Microsoft.FSharp.Core.ExtraTopLevelOperators.async

val r : 'T

IObserver.OnNext(value: 'T) : unit

IObserver.OnCompleted() : unit

val e : exn

IObserver.OnError(error: exn) : unit

static member Async.StartImmediate : computation:Async<unit> * ?cancellationToken:Threading.CancellationToken -> unit

property Threading.CancellationTokenSource.Token: Threading.CancellationToken

type IDisposable =
  member Dispose : unit -> unit

Full name: System.IDisposable

val x : IDisposable

IDisposable.Dispose() : unit