Introduce self-deregistering closures for on / onany

src/Observables.jl

@@ -84,24 +84,73 @@ end
 
 Base.show(io::IO, ::MIME"application/prs.juno.inline", x::Observable) = x
 
+
+"""
+    mutable struct ObserverFunction <: Function
+
+Fields:
+
+    f::Function
+    observable::AbstractObservable
+    weak::Bool
+
+`ObserverFunction` is intended as the return value for `on` because
+we can remove the created closure from `obsfunc.observable`'s listener vectors when
+ObserverFunction goes out of scope - as long as the `weak` flag is set.
+If the `weak` flag is not set, nothing happens
+when the ObserverFunction goes out of scope and it can be safely ignored.
+It can still be useful because it is easier to call `off(obsfunc)` instead of `off(observable, f)`
+to release the connection later.
+"""
+mutable struct ObserverFunction <: Function
+    f
+    observable::AbstractObservable
+    weak::Bool
+
+    function ObserverFunction(f, observable::AbstractObservable, weak)
+        obsfunc = new(f, observable, weak)
+
+        # If the weak flag is set, deregister the function f from the observable
+        # storing it in its listeners once the ObserverFunction is garbage collected.
+        # This should free all resources associated with f unless there
+        # is another reference to it somewhere else.
+        if obsfunc.weak
+            finalizer(obsfunc) do obsfunc
+                off(obsfunc)
+            end
+        end
+
+        obsfunc
+    end
+end
+
+
 """
     on(f, observable::AbstractObservable)
 
 Adds function `f` as listener to `observable`. Whenever `observable`'s value
 is set via `observable[] = val` `f` is called with `val`.
+
+Returns `true` if `f` could be removed, otherwise `false`.
 """
-function on(f, observable::AbstractObservable)
+function on(f, observable::AbstractObservable; weak = false)
     push!(listeners(observable), f)
     for g in addhandler_callbacks
         g(f, observable)
     end
-    return f
+    # Return a ObserverFunction so that the caller is responsible
+    # to keep a reference to it around as long as they want the connection to
+    # persist. If the ObserverFunction is garbage collected, f will be released from
+    # observable's listeners as well.
+    return ObserverFunction(f, observable, weak)
 end
 
 """
     off(observable::AbstractObservable, f)
 
 Removes `f` from listeners of `observable`.
+
+Returns `true` if `f` could be removed, otherwise `false`.
 """
 function off(observable::AbstractObservable, f)
     callbacks = listeners(observable)
@@ -111,10 +160,29 @@ function off(observable::AbstractObservable, f)
             for g in removehandler_callbacks
                 g(observable, f)
             end
-            return
+            return true
         end
     end
-    throw(KeyError(f))
+    return false
+end
+
+
+function off(observable::AbstractObservable, obsfunc::ObserverFunction)
+    f = obsfunc.f
+    # remove the function inside obsfunc as usual
+    off(observable, f)
+end
+
+"""
+    off(obsfunc::ObserverFunction)
+
+Remove the listener function `obsfunc.f` from the listeners of `obsfunc.observable`.
+Once `obsfunc` goes out of scope, this should allow `obsfunc.f` and all the values
+it might have closed over to be garbage collected (unless there
+are other references to it).
+"""
+function off(obsfunc::ObserverFunction)
+    off(obsfunc.observable, obsfunc)
 end
 
 """
@@ -220,16 +288,26 @@ end
 """
     onany(f, args...)
 
-Calls `f` on updates to any oservable refs in `args`.
+Calls `f` on updates to any observable refs in `args`.
 `args` may contain any number of `Observable` objects.
 `f` will be passed the values contained in the refs as the respective argument.
 All other objects in `args` are passed as-is.
 """
-function onany(f, args...)
+function onany(f, args...; weak = false)
     callback = OnUpdate(f, args)
+
+    # store all returned ObserverFunctions
+    obsfuncs = ObserverFunction[]
     for observable in args
-        (observable isa AbstractObservable) && on(callback, observable)
+        if observable isa AbstractObservable
+            obsfunc = on(callback, observable, weak = weak)
+            push!(obsfuncs, obsfunc)
+        end
     end
+
+    # same principle as with `on`, this collection needs to be
+    # stored by the caller or the connections made will be cut
+    obsfuncs
 end
 
 struct MapUpdater{F, T} <: InternalFunction

src/observablepair.jl

@@ -6,14 +6,20 @@ struct ObservablePair{S, T} <: AbstractObservable{T}
     excluded::Vector{Function}
     function ObservablePair(first::AbstractObservable{S}, second::AbstractObservable{T}; f = identity, g = identity) where {S, T}
         excluded = Function[]
-        first2second = on(first) do val
+
+        # the two observables should trigger each other, but only in one direction
+        # as otherwise there will be an infinite loop of updates
+
+        first2second_observerfunc = on(first) do val
             setindex!(second, f(val), notify = !in(excluded))
         end
-        push!(excluded, first2second)
-        second2first = on(second) do val
+        push!(excluded, first2second_observerfunc.f) # in notify, the wrapped function is compared
+
+        second2first_observerfunc = on(second) do val
             setindex!(first, g(val), notify = !in(excluded))
         end
-        push!(excluded, second2first)
+        push!(excluded, second2first_observerfunc.f)
+
         new{S, T}(first, second, f, g, excluded)
     end
 end

test/runtests.jl

@@ -13,9 +13,9 @@ using Test
     end
     r[] = 2
 
-    off(r, f)
+    @test off(r, f)
     @test !(f in r.listeners)
-    @test_throws KeyError off(r, f)
+    @test off(r, f) == false
     r[] = 3 # shouldn't call test
 end
 
@@ -44,6 +44,72 @@ end
     @test r2[] == 4
 end
 
+@testset "disconnect observerfuncs" begin
+
+    x = Observable(1)
+    y = Observable(2)
+    z = Observable(3)
+
+    of1 = on(x) do x
+        println(x)
+    end
+
+    of2_3 = onany(y, z) do y, z
+        println(y, z)
+    end
+
+    off(of1)
+    off.(of2_3)
+    for obs in (x, y, z)
+        @test isempty(obs.listeners)
+    end
+end
+
+# this struct is just supposed to show that a value in memory was released
+mutable struct ToFinalize
+    val
+
+    function ToFinalize(val, finalized_flag::Ref)
+        tf = new(val)
+        finalizer(tf) do tf
+            finalized_flag[] = true
+        end
+    end
+end
+
+@testset "weak connections" begin
+    a = Observable(1)
+
+    finalized_flag = Ref(false)
+
+    function create_a_dangling_listener()
+        t = ToFinalize(1, finalized_flag)
+
+        obsfunc = on(a; weak = true) do a
+            t.val += 1
+        end
+
+        a[] = 2
+        @test t.val == 2
+
+        # obsfunc falls out of scope here and should deregister the closure
+        # when it gets garbage collected, which should in turn free t
+        nothing
+    end
+
+    GC.enable(false)
+    create_a_dangling_listener()
+    @test length(Observables.listeners(a)) == 1
+    @test finalized_flag[] == false
+
+    GC.enable(true)
+    GC.gc()
+    # somehow this needs a double sweep, maybe first obsfunc, then ToFinalize?
+    GC.gc()
+    @test isempty(Observables.listeners(a))
+    @test finalized_flag[] == true
+end
+
 @testset "macros" begin
     a = Observable(2)
     b = Observable(3)