// ZeroMQ's three patterns inside one program, each a socket that binds and one that connects:
// PUB/SUB price ticks filtered by topic prefix, PUSH/PULL jobs shared out to a worker, and a
// REQ/REP stock service whose REP side is answered by a tick.
//
//   quillo run zeromq.qwe      then open http://localhost:8080/pubsub, /work and /ask
//
// No broker is needed: ZeroMQ is peer to peer. The binds take free ports on 127.0.0.1; to reach
// another program, bind a known port (tcp://0.0.0.0:5556) there and connect to it here.

use "zeromq.qwer"

let ready = false

fn sockets() {
    if !ready {
        zmq_socket("ticker", "pub", {})?
        let ticks = zmq_bind("ticker", "tcp://127.0.0.1:0")?
        zmq_socket("watch", "sub", {})?
        zmq_connect("watch", ticks)?
        zmq_subscribe("watch", "price.")?

        zmq_socket("jobs", "push", {})?
        let jobs = zmq_bind("jobs", "tcp://127.0.0.1:0")?
        zmq_socket("worker", "pull", {})?
        zmq_connect("worker", jobs)?

        zmq_socket("stock", "rep", { reply_timeout_ms: 5000 })?
        let svc = zmq_bind("stock", "tcp://127.0.0.1:0")?
        zmq_socket("client", "req", { timeout_ms: 3000 })?
        zmq_connect("client", svc)?

        // A subscription reaches the publisher a moment after it is made (ZeroMQ's "slow joiner").
        sleep(300)
        ready = true
    }
    true
}

// The stock service: each request read by the REP socket is answered, in turn.
let stock = { "sku-7": 12, "sku-9": 0 }
fn serve() {
    if !ready { return 0 }
    let asked = zmq_receive("stock", 10, 0) ?? []
    for q in asked {
        zmq_reply("stock", q.id, [q.payload, `${stock[q.payload] ?? -1}`])
    }
    len(asked)
}
every(100, serve, { name: "zmq-stock" })

// Three ticks, two of them `price.`: the SUB socket takes only those, as [topic, payload] frames.
route GET /pubsub {
    sockets()?
    zmq_send("ticker", ["price.EURUSD", "1.0842"])?
    zmq_send("ticker", ["news.fx", "ECB meets today"])?
    zmq_send("ticker", ["price.GBPUSD", "1.2611"])?
    let got = zmq_receive("watch", 10, 1000)?
    sleep(100)
    let more = zmq_receive("watch", 10, 0)?
    { ticks: (got + more) | map((m) => { topic: m.topic, price: m.payload }) }
}

// Jobs pushed, pulled by the worker in the order they went.
route GET /work {
    sockets()?
    for n in [1, 2, 3] {
        zmq_send("jobs", { job: n, task: "resize", file: `photo-${n}.jpg` })?
    }
    let got = zmq_receive("worker", 10, 1000)?
    sleep(100)
    let jobs = got + zmq_receive("worker", 10, 0)?
    { pulled: jobs | map((m) => json.parse(m.payload)?.file) }
}

// A request that waits for the REP side's answer, given by the tick above.
route GET /ask {
    sockets()?
    let answer = zmq_request("client", "sku-7", 3000)?
    { sku: answer.frames[0], in_stock: answer.frames[1], status: zmq_status("watch")? }
}