fix(): fix and tests for lazy observable

Project.toml

@@ -1,7 +1,7 @@
 name = "Rocket"
 uuid = "df971d30-c9d6-4b37-b8ff-e965b2cb3a40"
 authors = ["Dmitri Bagaev <bvdmitri@gmail.com>"]
-version = "1.3.18"
+version = "1.3.19"
 
 [deps]
 DataStructures = "864edb3b-99cc-5e75-8d2d-829cb0a9cfe8"

src/observable/lazy.jl

@@ -16,6 +16,7 @@ end
 end
 
 getpending(lazy::LazyObservable) = lazy.pending
+pushpending!(lazy::LazyObservable, subscription::LazySubscription, actor) = push!(lazy.pending, (subscription, actor))
 
 getstream(lazy::LazyObservable) = lazy.stream
 setstream!(lazy::LazyObservable, stream) = lazy.stream = stream
@@ -49,7 +50,9 @@ function on_subscribe!(observable::LazyObservable{D}, actor) where D
  if stream !== nothing
  return LazySubscription(subscribe!(stream, actor))
  else
- return LazySubscription(observable)
+ subscription = LazySubscription(observable)
+ pushpending!(observable, subscription, actor)
+ return subscription
  end
 end
 
@@ -62,7 +65,7 @@ function on_unsubscribe!(subscription::LazySubscription)
 end
 
 function unsubscribe_lazy!(subscription::LazySubscription, observable::LazyObservable)
- filter!(p -> first(p) === subscription, observable.pending)
+ filter!(p -> first(p) !== subscription, observable.pending)
  return nothing
 end
 

test/observable/test_observable_lazy.jl

@@ -0,0 +1,98 @@
+module RocketLazyObservableTest
+
+using Test
+using Rocket
+
+include("../test_helpers.jl")
+
+@testset "LazyObservable" begin
+
+ println("Testing: lazy")
+
+ @testset begin
+
+ values = []
+ stream = lazy(Int)
+ subscription = subscribe!(stream, (x) -> push!(values, x))
+ set!(stream, from(1:3))
+ @test values == [ 1, 2, 3 ]
+ subscription = subscribe!(stream, (x) -> push!(values, x))
+ @test values == [ 1, 2, 3, 1, 2, 3 ]
+
+ end
+
+ @testset begin
+
+ values = []
+ stream = lazy(Int)
+ subscription = subscribe!(stream, (x) -> push!(values, x))
+ unsubscribe!(subscription)
+ set!(stream, from(1:3))
+ @test values == [ ]
+ subscription = subscribe!(stream, (x) -> push!(values, x))
+ @test values == [ 1, 2, 3 ]
+
+ end
+
+ @testset begin
+
+ values1 = []
+ values2 = []
+ stream = lazy(Int)
+ subscription1 = subscribe!(stream, (x) -> push!(values1, x))
+ subscription2 = subscribe!(stream, (x) -> push!(values2, x))
+ set!(stream, from(1:3))
+ @test values1 == [ 1, 2, 3 ]
+ @test values2 == [ 1, 2, 3 ]
+
+ end
+
+ @testset begin
+
+ values1 = []
+ values2 = []
+ stream = lazy(Int)
+ subscription1 = subscribe!(stream, (x) -> push!(values1, x))
+ subscription2 = subscribe!(stream, (x) -> push!(values2, x))
+ unsubscribe!(subscription1)
+ set!(stream, from(1:3))
+ @test values1 == [ ]
+ @test values2 == [ 1, 2, 3 ]
+
+ end
+
+ @testset begin
+
+ values1 = []
+ values2 = []
+ stream = lazy(Int)
+ subscription1 = subscribe!(stream, (x) -> push!(values1, x))
+ subscription2 = subscribe!(stream, (x) -> push!(values2, x))
+ unsubscribe!(subscription2)
+ set!(stream, from(1:3))
+ @test values1 == [ 1, 2, 3]
+ @test values2 == [ ]
+
+ end
+
+ @testset begin
+
+ values = []
+ stream = lazy(Int)
+ subject = Subject(Int)
+ next!(subject, 1)
+ subscription = subscribe!(stream, (x) -> push!(values, x))
+ set!(stream, subject)
+ @test values == [ ]
+ next!(subject, 2)
+ @test values == [ 2 ]
+ unsubscribe!(subscription)
+ next!(subject, 3)
+ @test values == [ 2 ]
+
+ end
+
+
+end
+
+end

test/runtests.jl

@@ -33,6 +33,7 @@ doctest(Rocket)
  include("./observable/test_observable_timer.jl")
  include("./observable/test_observable_interval.jl")
  include("./observable/test_observable_proxy.jl")
+ include("./observable/test_observable_lazy.jl")
  include("./observable/test_observable_combine_latest.jl")
  include("./observable/test_observable_combine_updates.jl")
  include("./observable/test_observable_collect_latest.jl")