這邊我們要把前面的functional pattern跟Task組合成reactive programming。

import Base: get, map, filter, reduce

abstract Functor

function get{T <: Functor}(f::T)
end

function _map{T <: Functor}(x::T, f::Function)
    return T(f(get(x)))
end

function _filter{T <: Functor}(x::T, f::Function)
    if f(get(x))
        return x
    else
        return T(nothing)
    end
end

這邊基本上跟前面一樣，只不過我多宣告了_map跟_filter這兩個function。

immutable Disposible <: Functor
    value
end

function get(f::Disposible)
    return f.value
end

這邊是創建Functor的實作。

function init_task(f::Function, args::Vararg)
    x = @task f(args...)
    consume(x)
    return x
end

function map(f::Function)
    x = produce()
    while true
        y = _map(x, f)
        println(x, " -> ", y)
        x = produce(y)
    end
end

function filter(f::Function)
    x = produce()
    while true
        y = _filter(x, f)
        println(x, " -> ", y)
        x = produce(y)
    end
end

function bridge(f::Task, g::Task)
    x = produce()
    while true
        y = consume(f, x)
        z = consume(g, y)
        x = produce(z)
    end
end

這邊是前一篇的程式碼，map用到了functor的_map，filter用到了functor的_filter。
我們來測試看看！

t1 = init_task(map, x-> x+5)
t2 = init_task(map, x-> x*2)
t3 = init_task(bridge, t1, t2)
consume(t3, Disposible(5))

回傳

Disposible(5) -> Disposible(10)
Disposible(10) -> Disposible(20)

還可以

t4 = init_task(filter, x-> x>10)
t5 = init_task(bridge, t3, t4)
consume(t5, Disposible(5))

回傳

Disposible(5) -> Disposible(10)
Disposible(10) -> Disposible(20)
Disposible(20) -> Disposible(20)

這樣一來算是完成Reactive programming的部份功能了！