Signal / slot pattern on top of Reactive.jl

[femtotrader]

Hello,

I'm looking for a signal / slot library in Julia.
https://en.wikipedia.org/wiki/Signals_and_slots

see https://groups.google.com/forum/#!topic/julia-users/T1LUOtpEXXo

According doc
"Reactive.jl is a Julia package for Reactive Programming.
It makes writing event-driven programs simple."

but it's quite different from signal / slot pattern !
There's probably more features, but signal / slot is (in my mind) simpler.

I wonder if a signal / slot pattern can be build on top of Reactive.jl

Here is an example API

Define signal

stop_modified = Signal()

Emit signal

emit(stop_modified, args)

Define slot

function slot_on_stop_modified(args)
    println("stop was modified o $args")
end

Connect slot to signal

connect(stop_modified, slot_on_stop_modified)

A signal can be emit without being consumed
Several slots can be connected to a Signal.

A simple Python library for this pattern can be found here
http://signalslot.readthedocs.io/en/latest/usage.html

Kind regards

[shashi]

from your description, it seems to me that emit is exactly the same as push! and connect the same as map (with argument order exchanged). You will need to however call Reactive.preserve on the map signal to stop it from being garbage collected.

[femtotrader]

Here is a first implementation on top of Reactive.jl

using Reactive

Signal() = Reactive.Signal(Bool, false)
typealias Slot Function

function emit(signal::Reactive.Signal, args...; kwargs...)
    push!(signal, true)
#    push!(signal, false)
end

function myslot01(value::Bool, args...; kwargs...)
    println("myslot01 with $args and $kwargs")
end

function myslot02(value::Bool, args...; kwargs...)
    println("myslot02 with $args and $kwargs")
end

function myslot03(value::Bool, a, b; x=-1, y=-1)
    println("myslot03 with a=$a b=$b x=$x y=$y")
end

function connect(signal::Reactive.Signal, slot::Slot)
    preserve(map(slot, signal))
end

function is_connected(signal::Reactive.Signal, slot::Slot)
    # ToDo
end

#function disconnect(signal::Signal, slot::Slot)
    #ERROR: LoadError: TypeError: Tuple: in parameter, expected Type{T}, got Function
#end


stop_modified = Signal()
connect(stop_modified, myslot01)
connect(stop_modified, myslot02)
#connect(stop_modified, myslot03)
emit(stop_modified, 1, 2, x=3, y=4)

But I'm facing some problems...

emit is not able to pass arguments (positionnal arguments and keyword arguments)

I don't know how to implement is_connected

Removing comments for disconnect raises an error.

Pinging @barche who was involved in https://groups.google.com/forum/#!topic/julia-users/T1LUOtpEXXo discussion

[femtotrader]

An other issue with my current implementation

If there isn't enough delay between 2 emit calls... slots are not executed twice !!

using Reactive

Signal() = Reactive.Signal(Bool, false)
typealias Slot Function

function emit(signal::Reactive.Signal, args...; kwargs...)
    push!(signal, !signal.value)
end

function myslot01(value::Bool, args...; kwargs...)
    println("myslot01 with $args and $kwargs")
end

function myslot02(value::Bool, args...; kwargs...)
    println("myslot02 with $args and $kwargs")
end

function myslot03(value::Bool, a, b; x=-1, y=-1)
    println("myslot03 with a=$a b=$b x=$x y=$y")
end

function connect(signal::Reactive.Signal, slot::Slot)
    preserve(map(slot, signal))
end

function is_connected(signal::Reactive.Signal, slot::Slot)
    # ToDo
end

#function disconnect(signal::Signal, slot::Slot)
    #ERROR: LoadError: TypeError: Tuple: in parameter, expected Type{T}, got Function
#end


stop_modified = Signal()
connect(stop_modified, myslot01)
connect(stop_modified, myslot02)
#connect(stop_modified, myslot03)
emit(stop_modified, 1, 2, x=3, y=4)

#sleep(0.01)

emit(stop_modified, 10, 20, x=30, y=40)

[penntaylor]

@femtotrader, the reason the slot isn't called the second time unless you put in a delay is that Reactive's event processing loop runs in a separate thread. If your program ends before the scheduler yields to the event loop, the event queue is never emptied and your slot isn't called.

Run the code below to see behavior similar to what you're describing. It should print out the numbers 0 through 5 followed by "...done". Toggle the Bool under the comment about forcing flushing, and it should print out 0 through 10 followed by "...done".

using Reactive

x = Signal(0)
y = map(a->println("$a"), x)

for v = 1:5
  push!(x, v)
end

# Give Reactive's event loop a chance to process the previous pushes
yield()

for v = 6:10
  push!(x, v)
end

# Set false to force flushing
if true
  # A simple yield doesn't work here. I don't understand why.
  yield()
else
  # But, if we yield in a loop to flush out all remaining messages, things work as
  # expected
  while(isready(Reactive._messages))
    yield()
  end
end

println("...done")

@shashi: what do you think about exporting a flush() function from Reactive that wraps up that while(isready(... loop? Emptying the event queue looks to me to be the issue with #82. (Possibly #83 as well, but I'm not at all sure about that one.)

[timholy]

There's Reactive.run_till_now() which is the same as your flush. flush is a nicer name, though.

However, it's not quite so simple. flush is an exported name from Base, but it would have to be called Reactive.flush() (not exported) because Base.flush takes an argument and the argument-less version for Reactive really doesn't fit that pattern.

[penntaylor]

When I replace my flush loop with a call to Reactive.run_till_now(), some events arrive out-of-order, and the program hangs and has to be terminated with an Interrupt signal. I didn't really dig into it, but from the looks of the event loop and how run_till_now() operates, I suspect run is getting called twice in quick succession -- once from the main thread, and once from the thread where the usual Reactive event loop runs -- and this creates a deadlock-ish situation. Explicitly yielding to the event loop from the main thread avoids the problem. (I keep saying "thread", but what I really mean is "the thing that async and Tasks use to handle concurrency".)

Anyway, good point about Base.flush.