implement functional reactive programming in julia

reactive.jl

import Base: map, push!, reduce, mapreduce

type Stream{T}
    value::T
    update_rule::Function
    subscribers::Vector{Stream}
end

function Stream(startvalue, update_rule=self -> nothing, subscribers=[])
    T = typeof(startvalue)
    Stream{T}(startvalue, update_rule, subscribers)
end

function push!(s::Stream, t)
    s.value = t
    update!(s)
    s
end

function update!(s::Stream)
    s.update_rule(s)
    foreach(update!, s.subscribers)
end

function map(f, s::Stream)
    startvalue = f(s.value)
    update_rule(self) = (self.value = f(s.value))
    ret = Stream(startvalue, update_rule)
    push!(s.subscribers, ret)
    ret
end

function reduce(op, startvalue, stream::Stream)
    update_rule(self) = (self.value = op(self.value, stream.value))
    ret = Stream(startvalue, update_rule)
    push!(stream.subscribers, ret)
    ret
end

function mapreduce(f, op, startvalue, stream::Stream)
    reduce(op, startvalue, map(f, stream))
end

function heartbeat(dt, t_max=10)
    ret = Stream(nothing)
    t = 0.
    @async while t < t_max
        t += dt
        sleep(dt)
        update!(ret)
    end
    ret
end

history{T}(s::Stream{T}) = reduce(push!, T[], s)

function randstream(items; dt=1, t_max=10)
    s = heartbeat(dt)
    map(_ -> rand([items...]), s)
end

## example application

@enum MouseEvent LeftClick RightClick SomethingElse

mouseevents = randstream(instances(MouseEvent))
isclick(event) = event != SomethingElse

clickcount = mapreduce(isclick, +, 0, mouseevents)

map(display, mouseevents)
map(display ∘ isclick, mouseevents)
map(display, clickcount);