Observables API

Any observable-like should implement a valid subscribable logic.

Traits

Rocket.as_subscribable — Function

as_subscribable(any)

This function checks subscribable trait behavior specification. Can be used explicitly to specify subscribable trait behavior for any object.

Types

Rocket.AbstractSubscribable — Type

Supertype type for Subscribable and ScheduledSubscribable types.

source

Rocket.Subscribable — Type

Super type for any simple subscribable object. Automatically specifies a SimpleSubscribableTrait trait behavior. Objects with specified SimpleSubscribableTrait subscribable trait must implement: on_subscribe!(subscribable, actor) method. Subscribable is a subtype of AbstractSubscribable type.

Examples

using Rocket

struct MySubscribable <: Subscribable{String} end

Rocket.as_subscribable(MySubscribable)

# output

SimpleSubscribableTrait{String}()

source

Rocket.ScheduledSubscribable — Type

Super type for any scheduled subscribable object. Automatically specifies a ScheduledSubscribableTrait trait behavior. Objects with specified ScheduledSubscribableTrait subscribable trait must implement: on_subscribe!(subscribable, actor, scheduler) method. ScheduledSubscribable is a subtype of AbstractSubscribable type.

Examples

using Rocket

struct MyScheduledSubscribable <: ScheduledSubscribable{String} end

Rocket.as_subscribable(MyScheduledSubscribable)

# output

ScheduledSubscribableTrait{String}()

source

Rocket.subscribe! — Function

subscribe!(subscribable::T, actor::S)   where { T, S }
subscribe!(subscribable::T, factory::F) where { T, F <: AbstractActorFactory }
subscribe!(subscriptions::Tuple)
subscribe!(subscriptions::AbstractVector)

subscribe! function is used to attach an actor to subscribable. It also checks types of subscribable and actors to be a valid Subscribable and Actor objects respectively. Passing not valid subscribable or/and actor object will throw an error. If the input argument to the subscribe! function is either a tuple or a vector, it will first check that all of the arguments are valid source and actor objects and if its true will subscribe for each of them individually.

Arguments

subscribable: valid subscribable object
actor: valid actor object

Examples

using Rocket

source = from((1, 2, 3))
subscribe!(source, logger())
;

# output

[LogActor] Data: 1
[LogActor] Data: 2
[LogActor] Data: 3
[LogActor] Completed

using Rocket

source = from((1, 2, 3))
subscribe!(source, 1)
;

# output

ERROR: Type Int64 is not a valid actor type.
[...]

using Rocket

source = from((1, 2, 3))
subscribe!(1, logger())
;

# output

ERROR: Type Int64 is not a valid subscribable type.
[...]

source

Rocket.on_subscribe! — Function

on_subscribe!(subscribable, actor)
on_subscribe!(subscribable, actor, scheduler)

Every valid subscribable object have to define its own method for on_subscribe! function which specifies subscription logic and has return a valid Teardown object.

Objects with specified SimpleSubscribableTrait subscribable trait must implement: on_subscribe!(subscribable, actor) method. Objects with specified ScheduledSubscribableTrait subscribable trait must implement: on_subscribe!(subscribable, actor, scheduler) method.

Arguments

subscribable: Subscribable object
actor: Actor object
scheduler: Scheduler object (only for scheduled subscribables)

Examples

using Rocket

struct MySubscribable <: Subscribable{Int} end

function Rocket.on_subscribe!(subscribable::MySubscribable, actor)
    next!(actor, 0)
    complete!(actor)
    return voidTeardown
end

subscribe!(MySubscribable(), logger())
;

# output

[LogActor] Data: 0
[LogActor] Completed

using Rocket

struct MyScheduledSubscribable <: ScheduledSubscribable{Int} end

Rocket.getscheduler(::MyScheduledSubscribable) = AsapScheduler()

function Rocket.on_subscribe!(subscribable::MyScheduledSubscribable, actor, scheduler)
    next!(actor, 0, scheduler)
    complete!(actor, scheduler)
    return voidTeardown
end

subscribe!(MyScheduledSubscribable(), logger())
;

# output

[LogActor] Data: 0
[LogActor] Completed

source

Errors

Rocket.InvalidSubscribableTraitUsageError — Type

This error will be thrown if subscribe! function is called with invalid subscribable object