Useful requestor factories

[jamesdiacono]

As far as I am aware, only a single non-parseq requestor factory has ever been published: requestorize, from How JavaScript Works. As a result, parseq users have been left to write their own.

This is a place where we can share and learn about requestor factories that have proven useful.

Please provide an example of your requestor factory in use. You can include a link to its implementation if it is non-trivial. For example:

thru(requestor)

Returns a requestor that produces the same value it receives. The requestor
parameter is optional.

parseq.sequence([
    a(),
    thru(),
    thru(b()),
    c()
])

In the preceeding example, requestors b() and c() receive the value produced by
a(). The value produced by b() is discarded.

[douglascrockford]

There is also the 'widget' from https://www.crockford.com/parseq_demo.html and 'delay' from https://www.crockford.com/pp/eventuality.pptx

[jamesdiacono]

My bad, I missed those two, and also the read_file factory from the book.

Closing a discussion removes it from the Discussions page, did you mean for that to happen? I was hoping others could contribute their own factories.

[bunglegrind]

For similar purposes, I "requestorize" the ramda utility function tap ( https://ramdajs.com/docs/#tap ), which enables me to perform side effects jobs within a sequence of requestors, and debug the program too:

requestorize(R.tap(console.log))

[bunglegrind]

Well, I wrote a bunch of helper functions which basically enrich the parseq library with the patterns I more frequenly employ. Let me provide couple of example:

wrap_reason: it's factory which takes a requestor and returns a never failing requestor. I find it useful when I want to start many downloads in parallel and I don't want to block the overall task if one or more are failing.

function wrap_reason(requestor) {
    return function (cb, value) {
        return requestor(function (value, reason) {
            return cb({value, reason});
        }, value);
    };
}

factory_maker (make_factory? factory_of_factories? Phil Karlton said that naming things is hard in computer science - I would add, expecially when english is not your first language).

Ok, this is a bit complicated to describe. When "requestorizing" an eventual task (fs.readFile for instance), you need to pass its parameters as the factory arguments (if they are known in advance), or if they are know in a later turn, as the requestor second argument.
In my experience I've found very often that in the same program the same eventual task is executed with a different set of parameters known in advance or known in a later turn.
So, in order to avoid to write a different factory for every combination of parameters known in advance/later (the code becomes a mess, and it's definitely not dry), I wrote this factory_maker helper function.
In short, I decided to pass all the task parameters as the requestor second argument and to adapt it to the different "knowledge" context by providing an adapter function which is executed before the requestor (by means of parseq.sequence) and combines its closure with the value arriving from the previous requestor. A default adapter function is provided in order to manage the most common cases.

function factory_maker(requestor, factory_name = "factory") {
//the adapter combines the online value passed to the requestor with the
// closure/context in which the factory is executed
// its return value is passed to the requestor
    return function factory(adapter) {

//a default adapter is provided in order to manage the most common cases
        function default_adapter(precomputed) {
            return function (value) {
//default: both values are object, so we give the requestor their merge
                if (
                    typeof precomputed === "object"
                    && !Array.isArray(precomputed)
                ) {
                    return Object.assign(
                        {},
                        precomputed,
                        value
                    );
                }
//otherwise, default behavior is to provide only the precomputed value
//unless it's nullish
                return precomputed ?? value;
            };
        }

        if (typeof adapter !== "function") {
            adapter = default_adapter(adapter);
        }
        return function req(cb, value) {
            check_callback(cb, factory_name);
            return parseq.sequence([
                requestorize(adapter),
                requestor
            ])(cb, value);
        };
    };
}

A simple example:

function readFileRequestor(callback, {path, options}) {
    fs.readFile(path, options, function (err, data) {
        if (err) {
            return callback(undefined, err);
        }
        return callback(data);
    });
}

const readFile = factory_maker(readFileRequestor);

//generic adapter
readFile(function (value_from_previous_requestor) {
    return {
        path: value_from_previous_requestor.default_path,
        options: {}
    };
});

//default adapter
//is merged with the value arriving from the previous requestor if the adapter argument is an object
readFile({path: "./file.ext", options: {}}); 
//default adapter
readFile(); //the value arriving from the previous requestor is passed to the requestor

In this way I can write a more readable code, and more testable too (I can write tests for the adapter function if they're not straightforward). 😄

[jamesdiacono]

It's like currying but with objects.

[calebmsword]

I want to comment on the factory_maker function. There is a simpler solution for managing sequences of requestors whose values don't "link up" properly. Just use the unary function to mutate the value between the two (although I rename the function to map or map_value when I use it for this purpose):

function map_value(mapper) {
    return function(callback) {
        try {
            callback(mapper(value));
        }
        catch(reason) {
            callback(undefined, reason);
        }
    }
}

For example, suppose you have requestor factories get and post which make GET and POST https requests, respectively. The requestor returned by get results in a value that is an object with the property data (an object representing the JSON response), and other properties containing headers, status code, etc. Meanwhile, the requestor returned by post takes a value that is the body for the post request and results in a value that is a similar object to that from get. You could use the two, in combination with map_value, to get data from one endpoint and save it in another endpoint in the following way:

import parsec from './parsec.js';
import { get, post } './http-factories.js';
import { map_value } './utility-factories.js';

const get_stuff_and_save_it = parsec.sequence([
    // make a GET request
    get('https://example.com/api/stuff'),
    
    // extract the JSON response data from the GET request
    map_value(function(value) {
        return value.data;
    }),
    
    // save the data in a database
    post('https://my-database.com/api/things')
]);

[bunglegrind]

yep, that's almost exactly the same. factory_maker is just inserting the requestorize function before the actual requestor (and doing that in a more structured way)

[douglascrockford]

I call a function that makes factories a 'factory factory'. Is that too meta? Too repetitive?

[jamesdiacono]

I have not yet written a function that makes a factory yet, but I maybe I would call it an 'industrialist'.

[jamesdiacono]

pool(throttle)

Runs a dynamic collection of requestors in parallel. It is like parseq.parallel, except that instead of providing any array of requestors up front, they can be added wun at a time. The optional throttle parameter limits how many requestors are run at wunce.

Unlike a regular requestor factory, it returns an object with two methods:

add(requestor, initial_value)

Adds the requestor to the pool. If the pool is not being throttled, it will be run immediately. Otherwise it will be put on a wait list. The requestor is given the initial_value when it is run.

requestor(callback) -> cancel(reason)

A requestor that completes wunce the pool has no more running requestors.

If every requestor succeeds, the callback is called with the array of values produced by the requestors, in the order they were added to the pool.

If any requestor fails, its reason is passed to the callback and all other requestors are cancelled.

Any attempt to run this requestor whilst it is already running will fail.

Example

Here, a pool is used to traverse the filesystem, running an operation on each file as it is discovered. No more than 10 file_requestor's are run simultaneously. The traverse requestor factory takes the pool's add method, and a requestor that takes a file path.

const {add, requestor} = pool(10);

add(traverse(file_requestor, add), "./src");
add(traverse(file_requestor, add), "./test");

requestor(function callback(value, reason) {
    // ...
});

On success, value is an array containing file_requestor's value for each file, in the order that the file was discovered.

An implementation is available here.

[bunglegrind]

I've recently faced an issue with parseq, concerning the amount of memory occupied by the requestor returned values during a parallel run. In this case the throttle parameter is not helpful, since the resource bottleneck is in the local process, and at the end of the parallel task, you still have the all the returned values in an array.
So, I wrote a reduce requestor factory in order to store only the accumulator between a throttle turn and another.

function reduce(
    reducer,
    initial_value,
    requestor_array,
    throttle
) {
    throttle = throttle || requestor_array.length;

    return parseq.sequence([
        parseq.parallel(requestor_array.slice(0, throttle)),
        requestorize(function (array) {
            return array.reduce(reducer, initial_value);
        }),
        if_else(
            () => throttle >= requestor_array.length,
            do_nothing,
            (callback, value) => reduce(
                reducer,
                value,
                requestor_array.slice(throttle),
                throttle
            )(callback, value)
        )
    ]);
}

The reducer is a two argument function, not a requestor, since I don't see why I should perform an eventual task in the reducer step (but it can be easily extended).

It's still pretty raw, though. I've discarded the time parameters and the optional requestors array.
Moreover, even if it's recursive, the setTimeout in the parseq run function avoids call stack exaustion...but if you have a very large requestor array and a very small throttle, well, you'll meet the heap exaustion (mostly due to the array slice shallow copy and the very long chain of closures). Possibly, a more imperative/iterative version shouldn't be affected (Anyway, in my use cases, there are no issues).
Straightforward extension are reduce_object, apply_reduce (when the requestor array is dependent on some previous eventual task).

I was also wondering if I can extend it in order to apply it to an infinite stream (pagination, UI events, etc.) as RxJs does. I still think reactive programming (the observer pattern) is the more natural way to deal with events/streams but I would like to give a try.

[douglascrockford]

Log the value of a sequence. You can put it anywhere in a sequence, and it will put the current value in the log, without disrupting the sequence progress.

function log(prefix) {
    return function (callback, value) {
        console.log(prefix + " " + JSON.stringify(value));
        return callback(value);
    };
}

[bunglegrind]

My receipt to requestorize a promise...quite convoluted, but because of how promises work:

1. you can't defer execution (therefore, I'm actually working with promise thunks)
2. the catch block catches too much
3. usually impossible to cancel. In the rare cases (see fetch) you can provide a cancel function to the factory

const is_thunk = (f) => typeof f === "function" && !f.length;
const is_promise = (p) => typeof p?.then === "function";//a thenable

function promise_requestorize(
    promise_thunk,
    action = "executing promise",
    cancel = undefined
) {
    if (!is_thunk(promise_thunk)) {
        throw parseq.make_reason(
            action,
            `Not a thunk when ${action}`,
            promise_thunk
        );
    }

    return function (callback) {
        parseq.check_callback(callback, action);
        let is_called = false;
        function promise_callback(value, reason) {
            if (!is_called) {
                is_called = true;
                if (value === undefined) {
                    return callback(
                        undefined,
//first callback call: promise has thrown
                        parseq.make_reason(
                            "promise_requestorize",
                            `Failed when ${action}`,
                            reason
                        )
                    );
                }
                return callback(value);
            }
//second callback call: callback has thrown
            const e = new Error(`Callback failed when ${action}`);
            e.evidence = reason;
            throw e;
        }
        const promise = promise_thunk();
        if (!is_promise(promise)) {
            const ee = new Error(`Not a promise when ${action}`);
            return promise_callback(
                undefined,
                ee
            );
        }
        promise.then(promise_callback).catch(function (e) {
//at this point we still don't know if the promise or the callback has thrown
            promise_callback(
                undefined,
                e
            );
        });
        if (typeof cancel === "function") {
            return cancel;
        }
    };
}

[jamesdiacono]

I also have a factory for working with Promises, but it doesn't support cancellation.

function unpromise(make_promise, undefined_value) {
    return function unpromise_requestor(callback, value) {
        try {
            return make_promise(value).then(function (result) {
                return callback(result ?? undefined_value);
            }).catch(function (error) {
                return callback(undefined, error);
            });
        } catch (exception) {
            return callback(undefined, exception);
        }
    };
}

// Example

const fetch_requestor = unpromise(fetch);
fetch_requestor(callback, "https://site.com/news");

The undefined_value dictates the output value when the Promise succeeds whilst producing undefined (which is generally not a failure).

I've tried to write a requestor factory around fetch that supports cancellation, but haven't found a good solution so far. It's difficult to make cancellation apply to both phases of operation:

• the request (obtaining a Response object)
• reading the response body (as text? as JSON? as a stream?)

Maybe the factory should take a function that is responsible for reading the response body, so that the main requestor's cancel function remains usable throughout the second phase.

[bunglegrind]

Simply because the response object doesn't provide a cancel function for its methods (json(), blob(), etc.)
The only built-in promise which provides a cancel procedure AFAIK is fetch, via signals. In other words, the cancel function works only if the response is not arrived yet, otherwise it simply doesn't do anything.

An old example of mine, not so general (There are a couple of underlying assumptions, and it's not so smart in dealing with the errors...)

import pq from "./parseq-extended.js";

const fetch_factory = function (uri, method = "get", request_body) {
    const options = Object.create(null);
    options.method = method;
    if (request_body) {
        options.body = JSON.stringify(request_body);
        options.headers = {
            "Content-Type": "application/json"
        };
    }
    const controller = new AbortController();
    options.signal = controller.signal;
    return pq.promise_requestorize(
        () => fetch(uri, options).then((body) => body.json()),
        "fetch",
        controller.abort
    );
}

export default Object.freeze({factory: fetch_factory});

[jamesdiacono]

I believe an AbortController does cancel the response body, even after the Response has been produced. fetch_factory would have ideal cancellation handling, except that I think controller.abort needs to be called as a method or it will throw an exception.

    return pq.promise_requestorize(
        () => fetch(uri, options).then((body) => body.json()),
        "fetch",
        () => controller.abort()
    );

[bunglegrind]

I believe an AbortController does cancel the response body, even after the Response has been produced. fetch_factory would have ideal cancellation handling, except that I think controller.abort needs to be called as a method or it will throw an exception.

I should verify both...

[bunglegrind]

Concerning the latter, you're right, abort needs to be called as a method. But there's another problem (possibly...I'm still investigating): calling abort results in executing the catch function with a DOMException as argument, so we might fall back to something related to #13

/*jslint devel, node*/
/*global fetch*/
import http from "node:http";
import assert from "node:assert";

const url = "http://localhost:8000";

const server = http.createServer(function (ignore, res) {
    setTimeout(function () {
        res.end();
        server.close();
    }, 5000);
});
server.listen(8000);

const controller = new AbortController();
const signal = controller.signal;

fetch(url, {signal});

setTimeout(function () {
    const abort = controller.abort;
    try {
        abort();
    } catch (e) {
        assert.fail(`should not throw: ${e.message}`);
    }
}, 2000);

[bunglegrind]

Concerning the former, yes, it can cancel also the .json (etc.) method, until the response object goes out of the scope.

/*jslint devel, node*/
/*global fetch*/
import http from "node:http";
import assert from "node:assert";

const url = "http://localhost:8000";

const server = http.createServer(function (ignore, res) {
    console.log("writing response...");
    res.writeHead(200, {"Content-Type": "application/json"});
    res.end(JSON.stringify("Hello, world!"));
    server.close();
});
server.listen(8000);

const controller = new AbortController();
const signal = controller.signal;

fetch(url, {signal}).then(function (res) {
    console.log("response received...");
    setTimeout(function () {
        console.log("decoding json...");
        res.json().then(function (r) {
            setTimeout(function () {
                console.log(r);
            }, 1400);
        }).catch(function (err) {
            assert.fail(`Not here! ${err}`);
        });
    }, 2900);
}).catch(function (err) {
    assert.fail(`Not here! ${err}`);
});

setTimeout(function () {
    console.log("aborting request...");
    controller.abort();
}, 3000);

[jamesdiacono]

Yes, it is necessary to ignore the abort exception.

.catch(function (reason) {
    if (!controller.signal.aborted) {
        return callback(undefined, reason);
    }
})

[bunglegrind]

Yes, it is necessary to ignore the abort exception.

Actually the run function already discards multiple execution of callbacks*:

parseq/parseq.js

Line 150 in 04aa2ee

if (

parseq/parseq.js

Line 165 in 04aa2ee

number = undefined;

number is used as a flag

*however, there's the loophole described in #13

[jamesdiacono]

You're right about that, but it's not safe to assume that a requestor will be run via parseq.