Continuations over REST

irpc.js

// set up a mock HTTP client/server API
// should be pretty easy to replace this with
// the real ones

var endpoints = {}
async function fetch(path, opts){
    let data = await endpoints[path](JSON.parse(opts.body))
    return { async json(){ return data } }
}
async function serve(path, fn){
    endpoints[path] = fn;
}



// a cute little helper data structure
// call .pop() to get a promise for the next element added to queue
// call .push() to add something to a queue, fulfilling a promise if promised
function makeAwaitableQueue(){
    let queue = [],
        resolvers = [];
    return {
        pop(){
            if(queue.length > 0) return Promise.resolve(queue.shift());
            return new Promise((resolve, reject) => resolvers.push(resolve) )
        },
        push(payload){
            if(resolvers.length > 0) resolvers.shift()(payload)
            else queue.push(payload);
        }
    }
}


// clientside RPC call function supporting function parameter arguments
async function call(path, ...args){
    let message = {
        type: 'call',
        args: args.map((k, i) => 
            typeof k === 'function' ? '__func' : k)
    }
    while(true) {
        let result = await fetch(path, {
            method: 'POST',
            body: JSON.stringify(message),
        })
        let data = await result.json()
        if(data.type == 'callback'){
            let value = await args[data.index](...data.args)
            message = { type: 'response', id: data.id, value: value }
        }else if(data.type == 'return'){
            return data.value
        }
    }
}

// serverside RPC register function supporting function parameter arguments
function register(path, fn){
    let callbacks = []
    serve(path, async function(body){
        if(body.type == 'call'){
            let queue = makeAwaitableQueue();
            Promise.resolve(fn(...body.args.map((k, i) => 
                (k === '__func') ? (...args) => new Promise((resolve, reject) => {
                    let id = callbacks.length;
                    callbacks[id] = function(value){
                        resolve(value)
                        return queue.pop()
                    }
                    queue.push({ type: 'callback', id: id, index: i, args: args })
                }) : k))).then(value => 
                    queue.push({ type: 'return', value: value }))
            return queue.pop()
        }else if(body.type == 'response'){
            return await callbacks[body.id](body.value)
        }
    })
}


// register an example reduce endpoint
register('/reduce', async function(list, fn){
    if(list.length === 0) return null;
    if(list.length === 1) return list[0];
    let init = await fn(list[0], list[1])
    for(let i = 2; i < list.length; i++){
        init = await fn(init, list[i])
    }
    return init
})

// example using the reduce endpoint
;(async function(){
    let result = await call('/reduce', [1, 2, 3, 4], (a, b) => {
        return a + b
    })
    console.log('final result: ', result)
})()