Observables API

Any observable-like should implement a valid subscribable logic.

Traits

Rocket.as_subscribableFunction
as_subscribable(any)

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

See also: SubscribableTrait

source

Types

Rocket.SubscribableType

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}()

See also: SubscribableTrait, SimpleSubscribableTrait

source
Rocket.ScheduledSubscribableType

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}()

See also: SubscribableTrait, ScheduledSubscribableTrait

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.
[...]

See also: on_subscribe!, as_subscribable

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

See also: Subscribable, ScheduledSubscribable, SimpleSubscribableTrait, ScheduledSubscribableTrait, Teardown, logger

source

Errors