Add Observables.jl / Makie compatibility extension

Project.toml

@@ -8,15 +8,23 @@ DataStructures = "864edb3b-99cc-5e75-8d2d-829cb0a9cfe8"
 Sockets = "6462fe0b-24de-5631-8697-dd941f90decc"
 Unrolled = "9602ed7d-8fef-5bc8-8597-8f21381861e8"
 
+[weakdeps]
+Observables = "510215fc-4207-5dde-b226-833fc4488ee2"
+
+[extensions]
+RocketObservablesExt = "Observables"
+
 [compat]
 DataStructures = "0.17, 0.18, 0.19"
+Observables = "0.4, 0.5"
 Unrolled = "0.1.3"
 julia = "1"
 
 [extras]
 Coverage = "a2441757-f6aa-5fb2-8edb-039e3f45d037"
 Documenter = "e30172f5-a6a5-5a46-863b-614d45cd2de4"
+Observables = "510215fc-4207-5dde-b226-833fc4488ee2"
 Test = "8dfed614-e22c-5e08-85e1-65c5234f0b40"
 
 [targets]
-test = ["Test", "Coverage", "Documenter"]
+test = ["Test", "Coverage", "Documenter", "Observables"]

README.md

@@ -83,6 +83,10 @@ $ julia make.jl
 
 in the `docs/` directory to build a local version of the documentation.
 
+## Integration with Makie & Observables.jl
+
+Rocket.jl ships a package extension that, when [Observables.jl](https://github.com/JuliaGizmos/Observables.jl) is loaded, provides a bidirectional compatibility layer with the reactive primitive behind the [Makie](https://docs.makie.org/) ecosystem. A Rocket source (subject or operator pipeline) can be converted into an `Observable` that drives a Makie plot, and any Makie `Observable` can be consumed directly by Rocket's `subscribe!` and operators — letting you build rich, RxJS-inspired reactive logic on top of Makie widgets. See the [Makie & Observables.jl guide](https://reactivebayes.github.io/Rocket.jl/stable/integrations/observables/) for runnable examples.
+
 ## First example
 
 Normally you use arrays to process data.

docs/Project.toml

@@ -1,7 +1,11 @@
 [deps]
+CairoMakie = "13f3f980-e62b-5c42-98c6-ff1f3baf88f0"
 Documenter = "e30172f5-a6a5-5a46-863b-614d45cd2de4"
 LiveServer = "16fef848-5104-11e9-1b77-fb7a48bbb589"
+Observables = "510215fc-4207-5dde-b226-833fc4488ee2"
 Rocket = "df971d30-c9d6-4b37-b8ff-e965b2cb3a40"
 
 [compat]
+CairoMakie = "0.12, 0.13, 0.14, 0.15"
 Documenter = "1.0.0"
+Observables = "0.5"

docs/make.jl

@@ -8,6 +8,7 @@ makedocs(
     pages = [
         "Home" => "index.md",
         "Getting started" => "getting-started.md",
+        "Makie & Observables.jl" => "integrations/observables.md",
         "Manual" => [
             "Observable" => "observables/about.md",
             "Actor" => "actors/about.md",

docs/src/integrations/observables.md

@@ -0,0 +1,117 @@
+# [Makie & Observables.jl](@id section_makie_observables)
+
+[`Observables.jl`](https://github.com/JuliaGizmos/Observables.jl) is the reactive primitive that
+powers the [Makie](https://docs.makie.org/) plotting ecosystem. An `Observables.Observable` is a
+single container that holds a value and notifies a set of listeners whenever that value changes.
+
+Rocket.jl is a full [ReactiveX](https://reactivex.io/) / [RxJS](https://rxjs.dev/)-style reactive
+programming library: it provides cold and hot observables, [Subjects](@ref section_subjects),
+[schedulers](@ref schedulers_api), and a large catalogue of composable
+[operators](@ref operators_api). With this in mind, Rocket.jl can be used to express rich reactive
+logic that then drives a Makie visualisation.
+
+When the `Observables` package is loaded alongside Rocket.jl, a package extension
+(`RocketObservablesExt`) is loaded automatically and provides a **bidirectional compatibility
+layer** between the two libraries. This requires Julia `1.9` or higher.
+
+!!! note
+    The extension only adds methods to functions that already exist in either package (this is a
+    hard requirement of Julia's package-extension mechanism, which cannot export new names).
+    Concretely, it overloads the `Observables.Observable` constructor and teaches Rocket's
+    [`subscribe!`](@ref) to accept an `Observables.Observable`.
+
+## Comparison with Observables.jl
+
+| Concept | Observables.jl | Rocket.jl |
+| :--- | :--- | :--- |
+| Reactive container | `Observable(value)` | [`Subject`](@ref), [`BehaviorSubject`](@ref), [`RecentSubject`](@ref), [`ReplaySubject`](@ref) |
+| Current value | `observable[]` / `to_value(x)` | `getcurrent(behaviorsubject)` / `getrecent(recentsubject)` |
+| Push a new value | `observable[] = x` | [`next!(subject, x)`](@ref next!) |
+| React to changes | `on(f, observable)` | [`subscribe!(source, actor)`](@ref subscribe!) |
+| Stop reacting | `off(observable, f)` | [`unsubscribe!(subscription)`](@ref unsubscribe!) |
+| Derive a value | `map(f, observable)` | [`map`](@ref operator_map) + dozens of other [operators](@ref operators_api) |
+| Error / completion | not modelled | [`error!`](@ref) / [`complete!`](@ref) events |
+
+The key conceptual difference: an `Observable` is a *value that changes*, whereas a Rocket
+observable is a *stream of values over time*. A [`BehaviorSubject`](@ref) (which always carries a
+current value and replays it to new subscribers) is the Rocket type that most closely matches the
+semantics of an `Observable`.
+
+## Using Rocket.jl with Makie
+
+Because Makie decides whether something is reactive by checking whether it is an
+`Observables.AbstractObservable`, a Rocket observable cannot be handed to Makie *directly*. Instead,
+you convert a Rocket source into a genuine `Observable` with a single call — `Observable(source)` —
+and pass that to Makie. From then on, every value emitted by the Rocket source is pushed into the
+`Observable`, and Makie updates automatically.
+
+In the example below we build the plotting data with a Rocket pipeline: a [`BehaviorSubject`](@ref)
+holds an amplitude, and a [`map`](@ref operator_map) operator turns it into a curve. Converting the
+pipeline to an `Observable` lets Makie plot it.
+
+```@example makie
+using Rocket, Observables, CairoMakie
+CairoMakie.activate!() # hide
+
+x = range(0, 4π; length = 300)
+
+# A Rocket pipeline: an amplitude drives the y-data of a sine curve.
+amplitude = BehaviorSubject(1.0)
+curve     = amplitude |> map(Vector{Float64}, a -> a .* sin.(x))
+
+# Convert the Rocket pipeline into an Observable that Makie understands.
+ydata = Observable(sin.(x), curve)
+
+fig = Figure()
+ax  = Axis(fig[1, 1], title = "Driven by a Rocket BehaviorSubject")
+ylims!(ax, -3, 3)
+lines!(ax, x, ydata)
+fig
+```
+
+Now we push a new amplitude through the Rocket subject. The emission flows through the `map`
+operator, into the `Observable`, and Makie redraws the figure — we never touched the `Observable`
+or the plot directly:
+
+```@example makie
+next!(amplitude, 2.5)
+fig
+```
+
+## Using Makie observables with Rocket.jl
+
+The bridge also works the other way around: any `Observables.Observable` (for instance the value
+observable of a Makie `Slider` or `Menu`) is a valid Rocket source. You can [`subscribe!`](@ref) to
+it and run it through the full Rocket operator pipeline.
+
+```@example reverse
+using Rocket, Observables
+
+# Pretend this is the value observable of a Makie widget.
+slider = Observable(0)
+
+# Compose Rocket operators on top of it.
+squares_of_evens = keep(Int)
+subscription = subscribe!(slider |> filter(iseven) |> map(Int, x -> x ^ 2), squares_of_evens)
+
+for value in 1:6
+    slider[] = value
+end
+
+getvalues(squares_of_evens)
+```
+
+```@example reverse
+unsubscribe!(subscription) # detach the listener from the Observable
+```
+
+The current value of the `Observable` is emitted immediately upon subscription (mirroring
+[`BehaviorSubject`](@ref) semantics), and every subsequent `slider[] = value` is forwarded into the
+Rocket stream.
+
+## Summary
+
+* `Observable(source)` / `Observable(initial, source)` — convert a Rocket source (any subscribable,
+  subject, or operator pipeline) into an `Observables.Observable` for Makie.
+* [`subscribe!`](@ref) and every Rocket [operator](@ref operators_api) accept an
+  `Observables.Observable` directly — letting you build reactive pipelines on top of Makie widgets.

ext/RocketObservablesExt.jl

@@ -0,0 +1,100 @@
+module RocketObservablesExt
+
+# Compatibility layer between Rocket.jl and Observables.jl (the reactive primitive used
+# throughout the Makie ecosystem). This extension is loaded automatically by Pkg whenever
+# both `Rocket` and `Observables` are available (Julia >= 1.9).
+#
+# Per the package-extension rules, an extension may only add methods to functions that
+# already exist (in the parent package or the weak dependency) - it cannot export new names.
+# We therefore expose the bridge purely through existing entry points:
+#
+#   * Rocket source       -> `Observables.Observable(source)` / `Observable(initial, source)`
+#   * `Observables.Observable` -> a valid Rocket subscribable usable with `subscribe!` and operators
+
+using Rocket
+using Observables
+
+# ---------------------------------------------------------------------------------------- #
+# Direction A: a Rocket source -> an `Observables.Observable` (so Makie can consume it).     #
+# ---------------------------------------------------------------------------------------- #
+
+# A minimal `next`-only actor that pushes every Rocket emission into the target Observable.
+# Writing `observable[] = value` notifies all of the Observable's listeners (e.g. Makie plots).
+struct ObservableUpdateActor{O} <: Rocket.NextActor{Any}
+    observable::O
+end
+
+Rocket.on_next!(actor::ObservableUpdateActor, data) = setindex!(actor.observable, data)
+
+# Any object that Rocket recognises as a subscribable (plain observables and subjects alike).
+const RocketSource = Union{Rocket.AbstractSubscribable, Rocket.AbstractSubject}
+
+# Subscribe the bridge actor and tie the subscription's lifetime to the Observable. The actor
+# references the Observable and the Rocket source references the actor, so the Observable keeps
+# updating for as long as the source is alive. When the Observable (and source) become
+# unreachable, the finalizer disposes of the Rocket subscription.
+function bridge_rocket_to_observable!(observable, source)
+    subscription = subscribe!(source, ObservableUpdateActor(observable))
+    finalizer(observable) do _
+        unsubscribe!(subscription)
+    end
+    return observable
+end
+
+# Generic entry point: the caller supplies an explicit initial value. Works for any cold
+# observable, subject, or operator pipeline.
+function Observables.Observable(initial, source::RocketSource)
+    T = Rocket.subscribable_extract_type(source)
+    observable = Observables.Observable{T}(initial)
+    return bridge_rocket_to_observable!(observable, source)
+end
+
+# `BehaviorSubject` always carries a current value, so no initial value is required.
+function Observables.Observable(source::Rocket.BehaviorSubjectInstance{D}) where {D}
+    observable = Observables.Observable{D}(Rocket.getcurrent(source))
+    return bridge_rocket_to_observable!(observable, source)
+end
+
+# `RecentSubject` carries its most recently emitted value (or `nothing` if it never emitted).
+function Observables.Observable(source::Rocket.RecentSubjectInstance{D}) where {D}
+    recent = Rocket.getrecent(source)
+    recent === nothing && throw(ArgumentError(
+        "cannot create an `Observable` from a `RecentSubject` that has not emitted yet; " *
+        "use `Observable(initial, source)` to provide an initial value.",
+    ))
+    observable = Observables.Observable{D}(recent)
+    return bridge_rocket_to_observable!(observable, source)
+end
+
+# ---------------------------------------------------------------------------------------- #
+# Direction B: an `Observables.Observable` -> a Rocket subscribable (so Rocket operators and  #
+# `subscribe!` can consume Makie observables).                                               #
+# ---------------------------------------------------------------------------------------- #
+
+# Teardown that detaches the listener registered with `Observables.on`.
+struct ObservableSubscription{F} <: Rocket.Teardown
+    observerfunc::F
+end
+
+Rocket.as_teardown(::Type{<:ObservableSubscription}) = Rocket.UnsubscribableTeardownLogic()
+
+function Rocket.on_unsubscribe!(subscription::ObservableSubscription)
+    Observables.off(subscription.observerfunc)
+    return nothing
+end
+
+# Treat every `AbstractObservable{T}` as a simple Rocket subscribable producing values of type `T`.
+Rocket.as_subscribable(::Type{<:Observables.AbstractObservable{T}}) where {T} =
+    Rocket.SimpleSubscribableTrait{T}()
+
+function Rocket.on_subscribe!(observable::Observables.AbstractObservable, actor)
+    # Emit the current value immediately (BehaviorSubject-like semantics) ...
+    next!(actor, observable[])
+    # ... then forward every subsequent update.
+    observerfunc = Observables.on(observable) do value
+        next!(actor, value)
+    end
+    return ObservableSubscription(observerfunc)
+end
+
+end

test/ext/observables.jl

@@ -0,0 +1,91 @@
+module RocketObservablesExtTest
+
+using Test, Rocket
+using Observables
+
+@testset "Observables.jl compatibility extension" begin
+
+    @testset "extension is loaded" begin
+        @test Base.get_extension(Rocket, :RocketObservablesExt) !== nothing
+    end
+
+    @testset "Rocket -> Observable: BehaviorSubject" begin
+        subject = BehaviorSubject(0)
+        observable = Observable(subject)
+
+        @test observable isa Observable{Int}
+        @test observable[] == 0
+
+        updates = Int[]
+        on(value -> push!(updates, value), observable)
+
+        next!(subject, 42)
+        @test observable[] == 42
+
+        next!(subject, 100)
+        @test observable[] == 100
+
+        @test updates == [42, 100]
+    end
+
+    @testset "Rocket -> Observable: generic source with explicit initial value" begin
+        subject = Subject(Int)
+        observable = Observable(-1, subject)
+
+        @test observable isa Observable{Int}
+        @test observable[] == -1
+
+        next!(subject, 7)
+        @test observable[] == 7
+    end
+
+    @testset "Rocket -> Observable: RecentSubject" begin
+        empty_subject = RecentSubject(Int)
+        @test_throws ArgumentError Observable(empty_subject)
+
+        subject = RecentSubject(Int)
+        next!(subject, 5)
+        observable = Observable(subject)
+        @test observable[] == 5
+
+        next!(subject, 9)
+        @test observable[] == 9
+    end
+
+    @testset "Observable -> Rocket: subscribe! and teardown" begin
+        observable = Observable(10)
+
+        collected = keep(Int)
+        subscription = subscribe!(observable, collected)
+
+        # current value is emitted immediately on subscription
+        @test getvalues(collected) == [10]
+
+        observable[] = 20
+        observable[] = 30
+        @test getvalues(collected) == [10, 20, 30]
+
+        unsubscribe!(subscription)
+
+        observable[] = 40
+        @test getvalues(collected) == [10, 20, 30]
+    end
+
+    @testset "Observable -> Rocket: operator pipeline" begin
+        observable = Observable(1)
+
+        collected = keep(Int)
+        subscription = subscribe!(observable |> map(Int, x -> x * 2) |> filter(x -> x > 2), collected)
+
+        observable[] = 2
+        observable[] = 3
+
+        # initial 1 -> 2 is filtered out; 2 -> 4 and 3 -> 6 pass
+        @test getvalues(collected) == [4, 6]
+
+        unsubscribe!(subscription)
+    end
+
+end
+
+end

test/runtests.jl

@@ -113,6 +113,12 @@ doctest(Rocket)
     @testset "Detect ambiguities" begin
         @test length(Test.detect_ambiguities(Rocket)) == 0
     end
+
+    # Package extensions require Julia >= 1.9. Loading `Observables` here activates
+    # `RocketObservablesExt`, so this include must run after the ambiguity check above.
+    if VERSION >= v"1.9"
+        include("./ext/observables.jl")
+    end
 end
 
 end