Merge pull request #54 from ReactiveBayes/fix-43

Fix async pairwise

src/operators/pairwise.jl

@@ -66,10 +66,11 @@ mutable struct PairwiseActor{L, A} <: Actor{L}
 end
 
 function on_next!(actor::PairwiseActor{L}, data::L) where L 
-    if actor.previous !== nothing
-        next!(actor.actor, (actor.previous, data))
-    end
+    previous = actor.previous
     actor.previous = data
+    if !isnothing(previous)
+        next!(actor.actor, (previous, data))
+    end
 end
 
 on_error!(actor::PairwiseActor, err) = error!(actor.actor, err)

test/operators/test_operator_pairwise.jl

@@ -17,6 +17,11 @@ include("../test_helpers.jl")
             values      = @ts([ (1, 2), (2, 3), (3, 4), (4, 5), c ]),
             source_type = Tuple{Int, Int}
         ),
+        (
+            source      = from(1:5) |> async() |> pairwise(),
+            values      = @ts([ (1, 2), (2, 3), (3, 4), (4, 5), c ]),
+            source_type = Tuple{Int, Int}
+        ),
         (
             source      = of(1) |> pairwise(),
             values      = @ts(c),
@@ -70,4 +75,15 @@ include("../test_helpers.jl")
 
 end
 
+@testset "Issue #43" begin 
+    subject = Subject(Int)
+    paired = subject |> pairwise()
+    values = []
+    subscribe!(paired, (v) -> push!(values, v))
+    @sync for i = 1:10
+        @async next!(subject, i)
+    end
+    @test values == [(1, 2), (2, 3), (3, 4), (4, 5), (5, 6), (6, 7), (7, 8), (8, 9), (9, 10)]
+end
+
 end