Introduction
In F#, IObservable<'T> and Async<'T> solve different problems. An observable represents a stream of events over time, while an async workflow represents one asynchronous computation that eventually produces a single result. Mixing them works well once you keep that distinction clear and choose the right conversion point.
Understand the Difference in Shape
Async<'T> is pull-like and single-result: you start a computation and eventually receive one value or an exception. IObservable<'T> is push-like and multi-result: values arrive whenever the producer emits them.
That means you usually combine them in one of two directions:
The shape mismatch is why naive code often feels awkward. If you try to "await an observable" without deciding which event you want, the code has no clear meaning.
Process Observable Events With Async Work
A common pattern is a stream of events where each event needs an asynchronous operation such as HTTP I/O or database access. The safe approach is to subscribe to the observable and start an async workflow for each event.
1open System
2
3let source = new Event<int>()
4
5source.Publish.Add(fun value ->
6 async {
7 do! Async.Sleep 100
8 printfn "Processed %d" value
9 }
10|> Async.Start) source.Trigger 1 source.Trigger 2 System.Threading.Thread.Sleep 300 ``` This is the simplest bridge. Each pushed value starts one async workflow. The tradeoff is that `Async.Start` makes fire-and-forget work easy. That is fine for quick examples, but in production you need a policy for errors, cancellation, and concurrency limits. ## Convert One Observable Event Into Async Sometimes you want to wait asynchronously for the next observable value. In that case, create an async workflow that completes when the next event arrives. ```fsharp open System let awaitNextEvent (obs: IObservable<'T>) = Async.FromContinuations(fun (cont, econt, ccont) -> let subscription = ref None let observer = { new IObserver<'T> with member _.OnNext value = match !subscription with | Some d -> d.Dispose() | None -> () cont value member _.OnError ex = econt ex member _.OnCompleted() = ccont (OperationCanceledException("Observable completed before value arrived")) } subscription := Some (obs.Subscribe observer)) let eventSource = new Event<string>() async { let! value = awaitNextEvent eventSource.Publish printfn "Received: %s" value } |> Async.Start eventSource.Trigger "ready" System.Threading.Thread.Sleep 100 ``` This makes the conversion explicit. The async workflow waits for one item, not the whole stream. ## Keep Backpressure and Ordering in Mind Observable streams can emit values faster than async work can complete. If you start one async workflow per event with no limits, you can easily overwhelm downstream services. A simple coordination approach is to serialize work through a mailbox. ```fsharp open System let processor = MailboxProcessor.Start(fun inbox -> let rec loop () = async { let! value = inbox.Receive() do! Async.Sleep 100 printfn "Handled %d" value return! loop () } loop ()) let events = new Event<int>() events.Publish.Add(fun value -> processor.Post value) events.Trigger 10 events.Trigger 20 events.Trigger 30 System.Threading.Thread.Sleep 400 ``` This does not turn observables into async automatically. It gives you a deliberate concurrency boundary, which is usually what you want. ## Use the Right Abstraction for Long Streams If most of your logic is stream-based, plain `IObservable<'T>` plus Reactive Extensions operators may be a better fit than jumping back and forth to async workflows. If most of your logic is workflow-based, an async sequence abstraction can feel more natural. The key is to avoid unnecessary conversions. Convert once at a boundary where the program logic clearly changes shape. ## Common Pitfalls - Treating `IObservable<'T>` as if it were just an async value that happens later. - Starting one async workflow per event without any error handling or concurrency control. - Forgetting to dispose subscriptions when converting from observables to single-result async workflows. - Mixing stream operators and async code so often that control flow becomes hard to reason about. - Ignoring completion semantics when waiting for the next event. ## Summary - '`Async<'T>` models one asynchronous result, while `IObservable<'T>` models a stream of results.' - Subscribe and start async work when each pushed event needs asynchronous processing. - Convert an observable to async only when you can define which event should complete the workflow. - Add concurrency control if events can arrive faster than async work finishes. - Choose one dominant abstraction and convert only at clear boundaries.
