# # Google Search 3.0

using Julio
using Julio: Events
using ContextManagers: @with, SharedResource

# This is an example taken from Rob Pike's 2012 talk *Go Concurrency Patterns*.

# Avoid timeout (search engine replicas)
# <https://talks.golang.org/2012/concurrency.slide#48>

# `replicas` is a tuple of callables with the signature `query -> promise`;
# e.g., `replicas = webs = (web1, web2)`.

first_response!(send_endpoint, query::AbstractString, replicas::Tuple) =
    put!(send_endpoint, Julio.select(Events.fetch.(query .|> replicas)...))

# Google Search 3.0
# <https://talks.golang.org/2012/concurrency.slide#50>

function replicated_search(webs, images, videos)
    query = "Julio"
    results = String[]
    send_endpoint, receive_endpoint = Julio.channel()
    try
        Julio.withtaskgroup() do tg
            @with(handle = SharedResource(send_endpoint)) do
                Julio.spawn!(first_response!, tg, handle, query, webs)
                Julio.spawn!(first_response!, tg, handle, query, images)
                Julio.spawn!(first_response!, tg, handle, query, videos)
            end
            Julio.spawn!(tg) do
                Julio.sleep(0.08)
                Julio.cancel!(tg)
            end
            append!(results, receive_endpoint)
            Julio.cancel!(tg)
        end
    finally
        close(receive_endpoint)
    end
    return results
end

# Setup code:

replicated_search_demo() = with_search_engines(replicated_search)

function with_close_hook(f)
    resources = []
    function closing(r)
        push!(resources, r)
        return r
    end
    try
        f(closing)
    finally
        foreach(close, resources)
    end
end

function spawn_fakesearch!(tg, label, closing)
    send_endpoint, receive_endpoint = Julio.channel()
    closing(send_endpoint)
    Julio.spawn!(tg) do
        try
            for (query, reply) in receive_endpoint
                sleep(rand(0.01:0.01:0.1))
                reply[] = "$label result for $query"
            end
        finally
            close(receive_endpoint)
        end
    end
    function request(query)
        reply = Julio.Promise()
        put!(send_endpoint, query => reply)
        return reply
    end
    return request
end

function with_search_engines(f)
    Julio.withtaskgroup() do tg
        with_close_hook() do closing
            web1 = spawn_fakesearch!(tg, "web1", closing)
            web2 = spawn_fakesearch!(tg, "web2", closing)
            image1 = spawn_fakesearch!(tg, "image1", closing)
            image2 = spawn_fakesearch!(tg, "image2", closing)
            video1 = spawn_fakesearch!(tg, "video1", closing)
            video2 = spawn_fakesearch!(tg, "video2", closing)
            f((web1, web2), (image1, image2), (video1, video2))
        end
    end
end

using Test

function test_replicated_search_demo()
    nok = 0
    for _ in 1:10
        results = replicated_search_demo()
        nok += length(results) == 3
    end
    @test nok > 0
end
