Merge pull request #28 from biaslab/develop-1.5.0

update version to 1.5.0

.github/workflows/ci.yml

@@ -22,6 +22,7 @@ jobs:
           - '1.5'
           - '1.6'
           - '1.7'
+          - '1.8'
           - 'nightly'
         os:
           - ubuntu-latest
@@ -57,7 +58,7 @@ jobs:
       - uses: actions/checkout@v2
       - uses: julia-actions/setup-julia@v1
         with:
-          version: '1.7'
+          version: '1.8'
       - run: |
           julia --project=docs -e '
             using Pkg

Project.toml

@@ -1,7 +1,7 @@
 name = "Rocket"
 uuid = "df971d30-c9d6-4b37-b8ff-e965b2cb3a40"
 authors = ["Dmitri Bagaev <[email protected]>"]
-version = "1.4.0"
+version = "1.5.0"
 
 [deps]
 DataStructures = "864edb3b-99cc-5e75-8d2d-829cb0a9cfe8"

docs/make.jl

@@ -39,12 +39,14 @@ makedocs(
                 "Network"           => "observables/types/network.md",
                 "Defer"             => "observables/types/defer.md",
                 "Zipped"            => "observables/types/zipped.md",
+                "Labeled"           => "observables/types/labeled.md",
             ],
             "Actors" => [
                 "Lambda"        => "actors/types/lambda.md",
                 "Logger"        => "actors/types/logger.md",
                 "Sync"          => "actors/types/sync.md",
                 "Keep"          => "actors/types/keep.md",
+                "CircularKeep"  => "actors/types/circularkeep.md",
                 "Buffer"        => "actors/types/buffer.md",
                 "Void"          => "actors/types/void.md",
                 "Function"      => "actors/types/function.md",

docs/src/actors/types/circularkeep.md

@@ -0,0 +1,9 @@
+# [CircularKeep actor](@id actor_circularkeep)
+
+```@docs
+circularkeep
+```
+
+```@docs
+CircularKeepActor
+```

docs/src/observables/types/labeled.md

@@ -0,0 +1,9 @@
+# [Labeled Observable](@id observable_labeled)
+
+```@docs
+labeled
+```
+
+```@docs
+LabeledObservable
+```

src/Rocket.jl

@@ -28,6 +28,7 @@ include("actor/logger.jl")
 include("actor/void.jl")
 include("actor/sync.jl")
 include("actor/keep.jl")
+include("actor/circularkeep.jl")
 include("actor/buffer.jl")
 include("actor/server.jl")
 include("actor/storage.jl")
@@ -46,6 +47,7 @@ include("subjects/recent.jl")
 @generate_subscribe! RecentSubjectInstance AbstractSubject
 
 include("observable/generate.jl")
+include("observable/labeled.jl")
 include("observable/single.jl")
 include("observable/array.jl")
 include("observable/iterable.jl")

src/actor/circularkeep.jl

@@ -0,0 +1,83 @@
+export CircularKeepActor, circularkeep, getvalues
+
+import DataStructures: CircularBuffer
+
+"""
+    CirucalKeepActor{D}() where D
+
+Circual keep actor is similar to keep actor, but uses `CircularBuffer` as a storage. 
+It saves all incoming successful `next` events in a `values` circular buffer, throws an ErrorException on `error!` event and does nothing on completion event.
+
+# Examples
+```jldoctest
+using Rocket
+
+source = from(1:5)
+actor  = circularkeep(Int, 3)
+
+subscribe!(source, actor)
+show(getvalues(actor))
+
+# output
+[3, 4, 5]
+```
+
+See also: [`Actor`](@ref), [`keep`](@ref), [`circularkeep`](@ref)
+"""
+struct CircularKeepActor{T} <: Actor{T}
+    values :: CircularBuffer{T}
+
+    CircularKeepActor{T}(capacity::Int) where T = new{T}(CircularBuffer{T}(capacity))
+end
+
+getvalues(actor::CircularKeepActor) = actor.values
+
+on_next!(actor::CircularKeepActor{T}, data::T) where T = push!(actor.values, data)
+on_error!(actor::CircularKeepActor, err)               = error(err)
+on_complete!(actor::CircularKeepActor)                 = begin end
+
+"""
+    circularkeep(::Type{T}, capacity::Int) where T
+
+# Arguments
+- `::Type{T}`: Type of keep data
+- `capacity::Int`: circular buffer capacity
+
+Creation operator for the `CircularKeepActor` actor.
+
+# Examples
+
+```jldoctest
+using Rocket
+
+actor = circularkeep(Int, 3)
+actor isa CircularKeepActor{Int}
+
+# output
+true
+```
+
+See also: [`CircularKeepActor`](@ref), [`AbstractActor`](@ref)
+"""
+circularkeep(::Type{T}, capacity::Int) where T = CircularKeepActor{T}(capacity)
+
+# Julia iterable interface
+
+Base.IteratorSize(::Type{ <: CircularKeepActor })   = Base.HasLength()
+Base.IteratorEltype(::Type{ <: CircularKeepActor }) = Base.HasEltype()
+
+Base.IndexStyle(::Type{ <: CircularKeepActor }) = Base.IndexLinear()
+
+Base.eltype(::Type{ <: CircularKeepActor{T} }) where T = T
+
+Base.iterate(actor::CircularKeepActor)        = iterate(actor.values)
+Base.iterate(actor::CircularKeepActor, state) = iterate(actor.values, state)
+
+Base.size(actor::CircularKeepActor)        = (length(actor.values), )
+Base.length(actor::CircularKeepActor)      = length(actor.values)
+Base.getindex(actor::CircularKeepActor, I) = Base.getindex(actor.values, I)
+
+Base.getindex(actor::CircularKeepActor, ::Unrolled.FixedRange{A, B}) where { A, B } = getindex(actor, A:B)
+
+Base.firstindex(actor::CircularKeepActor) = firstindex(actor.values)
+Base.lastindex(actor::CircularKeepActor)  = lastindex(actor.values)

src/actor/test.jl

@@ -419,7 +419,13 @@ macro ts(expr)
         elseif arg.head === :vect
             push!(values, collect(arg.args))
         elseif arg.head === :tuple
-            push!(values, (arg.args..., ))
+            if all(expr -> expr isa Expr && expr.head == :(=), arg.args) # NamedTuple case 
+                names = map(d -> d.args[1], arg.args)
+                items = map(d -> d.args[2], arg.args)
+                push!(values, NamedTuple{tuple(names...)}(items))
+            else # regular tuple case
+                push!(values, (arg.args..., ))
+            end
         elseif arg.head === :call
             if arg.args[1] === :e
                 push!(tests, TestActorErrorTest(length(arg.args) === 2 ? arg.args[2] : nothing))

src/observable/labeled.jl

@@ -0,0 +1,59 @@
+export LabeledObservable, labeled
+
+import Base: show
+
+"""
+    labeled(names::Val, stream)
+
+Creation operator for the `LabeledObservable` that wraps given `stream`, that produces `Tuple` values into a `NamedTuple` with given `names`.
+
+# Arguments
+- `names`: a `Val` object that contains a tuple of symbols
+- `stream`: an observable that emits a `Tuple`, length of the `Tuple` events must be equal to the length of the `names` argument
+
+# Examples
+
+```jldoctest
+using Rocket
+
+source = labeled(Val((:x, :y)), from([ (1, 2), (2, 3), (3, 4) ]))
+subscribe!(source, logger())
+;
+
+# output
+
+[LogActor] Data: (x = 1, y = 2)
+[LogActor] Data: (x = 2, y = 3)
+[LogActor] Data: (x = 3, y = 4)
+[LogActor] Completed
+```
+
+See also: [`ScheduledSubscribable`](@ref), [`subscribe!`](@ref), [`from`](@ref)
+"""
+labeled(::Val{Names}, stream::S) where { Names, S }               = labeled(eltype(S), Val(Names), stream)
+labeled(::Type{D}, ::Val{Names}, stream::S) where { D, Names, S } = LabeledObservable{NamedTuple{Names, D}, S}(stream)
+
+"""
+    LabeledObservable{D, S}()
+
+An Observable that emits `NamesTuple` items from a source `Observable` that emits `Tuple` items.
+
+See also: [`Subscribable`](@ref), [`labeled`](@ref)
+"""
+@subscribable struct LabeledObservable{D, S} <: Subscribable{D}
+    stream :: S
+end
+
+struct LabeledActor{T, N, A} <: Actor{T} 
+    actor :: A
+end
+
+@inline on_next!(actor::LabeledActor{T, N}, data::T) where { T, N } = next!(actor.actor, NamedTuple{N, T}(data))
+@inline on_error!(actor::LabeledActor, err)                         = error!(actor.actor, err)
+@inline on_complete!(actor::LabeledActor)                           = complete!(actor.actor)
+
+function on_subscribe!(observable::LabeledObservable{R}, actor::A) where { N, T, R <: NamedTuple{N, T}, A }
+    return subscribe!(observable.stream, LabeledActor{T, N, A}(actor))
+end
+
+Base.show(io::IO, ::LabeledObservable{D, S}) where { D, S } = print(io, "LabeledObservable($D, $S)")

test/actor/test_circularkeep_actor.jl

@@ -0,0 +1,81 @@
+module RocketKeepActorTest
+
+using Test
+using Rocket
+using DataStructures
+
+@testset "CircularKeepActor" begin
+
+    println("Testing: actor CircularKeepActor")
+
+    @testset begin
+        source = from([ 1, 2, 3 ])
+        actor  = CircularKeepActor{Int}(2)
+
+        subscribe!(source, actor)
+
+        @test actor.values == [ 2, 3 ]
+        @test getvalues(actor) == [ 2, 3 ]
+    end
+
+    @testset begin
+        source = from(1:100)
+        actor  = CircularKeepActor{Int}(2)
+
+        subscribe!(source, actor)
+
+        @test actor.values == [ 99, 100 ]
+        @test getvalues(actor) == [ 99, 100 ]
+        @test length(getvalues(actor)) == 2
+    end
+
+    @testset begin
+        source = from([ 1, 2, 3 ])
+        actor  = CircularKeepActor{Int}(2)
+
+        subscribe!(source, actor)
+
+        @test actor[1] === 2
+        @test actor[2] === 3
+
+        @test collect(actor) == [ 2, 3 ]
+    end
+
+    @testset begin
+        source = from([ 1, 2, 3 ])
+        actor  = CircularKeepActor{Int}(2)
+
+        subscribe!(source, actor)
+
+        i = 2
+        for item in actor
+            @test item === i
+            i += 1
+        end
+    end
+
+    @testset begin
+        source = from([ 1, 2, 3 ])
+        actor  = CircularKeepActor{Int}(2)
+
+        subscribe!(source, actor)
+
+        @test actor[1:end] == [ 2, 3 ]
+    end
+
+    @testset begin
+        source = faulted(Int, "Error")
+        actor  = CircularKeepActor{Int}(2)
+
+        @test_throws ErrorException subscribe!(source, actor)
+        @test actor.values == []
+        @test getvalues(actor) == []
+    end
+
+    @testset begin
+        @test circularkeep(Int, 3) isa CircularKeepActor{Int}
+        @test capacity(getvalues(circularkeep(Int, 3))) === 3
+    end
+end
+
+end

test/observable/test_observable_labeled.jl

@@ -0,0 +1,42 @@
+module RocketLabeledObservableTest
+
+using Test
+using Rocket
+
+include("../test_helpers.jl")
+
+@testset "LabeledObservable" begin
+
+    @testset begin
+        @test labeled(Val((:x, )), from([ (1, ), (2, ), (3, ) ])) isa LabeledObservable{NamedTuple{(:x, ), Tuple{Int}}}
+        @test labeled(Val((:x, :y)), from([ (1, 1.0), (2, 2.0), (3, 3.0) ])) isa LabeledObservable{NamedTuple{(:x, :y), Tuple{Int, Float64}}}
+    end
+
+    @testset begin
+        source = labeled(Val((:x, )), from([ (1, ), (2, ), (3, ) ]))
+        io = IOBuffer()
+
+        show(io, source)
+
+        printed = String(take!(io))
+
+        @test occursin("LabeledObservable", printed)
+        @test occursin(string(eltype(source)), printed)
+    end
+
+    run_testset([
+        (
+            source = labeled(Val((:x, )), from([ (1, ), (2, ), (3, ) ])),
+            values = @ts([ (x = 1, ), (x = 2, ), (x = 3, ), c ]),
+            source_type = NamedTuple{(:x, ), Tuple{Int}}
+        ),
+        (
+            source = labeled(Val((:x, :y)), from([ (1, 2.0), (2, 3.0), (3, 4.0) ])),
+            values = @ts([ (x = 1, y = 2.0), (x = 2, y = 3.0), (x = 3, y = 4.0), c ]),
+            source_type = NamedTuple{(:x, :y), Tuple{Int, Float64}}
+        ),
+    ])
+
+end
+
+end

test/runtests.jl

@@ -18,13 +18,15 @@ doctest(Rocket)
     include("./actor/test_lambda_actor.jl")
     include("./actor/test_logger_actor.jl")
     include("./actor/test_keep_actor.jl")
+    include("./actor/test_circularkeep_actor.jl")
     include("./actor/test_buffer_actor.jl")
     include("./actor/test_sync_actor.jl")
     include("./actor/test_function_actor.jl")
     include("./actor/test_storage_actor.jl")
 
     include("./test_subscribable.jl")
     include("./observable/test_observable_function.jl")
+    include("./observable/test_observable_labeled.jl")
     include("./observable/test_observable_single.jl")
     include("./observable/test_observable_array.jl")
     include("./observable/test_observable_iterable.jl")