registerWorker call feels awkward

[iarna]

(Unless you're using the payload as a stream.)

I find that using the payload as a string is quite awkward as it means another layer of indirection to either promise or concat the stream. It would be nice to be able to register the worker such that you get the payload at initial call time. In that scenario the task wouldn't be readable, but it would still be writable. There are a few ways we might do this... a task.payload property or an additional argument to the callback both come to mind.

I kind of want to just trigger off of the callback's function signature, even though my "maybe this is too magical" alarm is going off:

gm.registerWorker('test', function (task) {
    // Task is exactly as today, a readable, writable stream that can
    // also be resolved as a promise
});
gm.registerWorker('test', function (task,payload) {
    // Task is a writable stream only, with no promise
});

Alternatively we could pass in an option, for example:

gm.registerWorker('test',{nostream: true}, function (task,payload) {

[amv]

Have I understood correctly that the "non-streaming" mode is pretty much the default in many other libraries?

A very non-magical way to do this would be to have a different function call (like submitJob and submitJobBg) for the streaming worker:

gm.registerStreamingWorker( 'test', function( task ) {
    // current functionality
} );
gm.registerWorker( 'test', function( task ) {
    // promised functionality with payload in task.payload
}

I think some other libraries also implement this with a reverse attribute order:

gm.registerWorker( 'test', function( payload, task/worker ) {} )

At least my current preference would be for the task object as the sole parameter, which would keep the interface similar to the streaming worker (it could also still be "thenable"). The implementation of registerStreamingWorker could just be a call to gm.registerWorker('test', { streaming : true }, ..) inside. The payload could be accessible as task.payload or as task.payload( [ encoding ] ).

[iarna]

Why would you want it to be thenable outside of streaming? Wouldn't that just be a no op?

I'll have to meditate more on this, as I'm not really happy with any of the solutions so far.

[amv]

Yes, it would be a "no op" (immediately fulfilling promise) and I would consider that valuable just to keep the interfaces consistent for streaming and non-streaming workers.

If you used .payload() instead of .payload, you could also throw an error in case of someone trying to request payload from a streaming worker. I am not sure how to guard against that in the case of a streaming worker if a simple variable is used.

Is the payload supposed to be a Buffer or a string btw? If it is a Buffer, what would be the easiest way to use it as an utf8 string (probably the most common scenario(?)) while still being able to request also other encodings?

BTW. Is it possible for the worker to receive an error while reading or writing streaming data? How are those errors handled if they are possible?

[iarna]

The payload is whatever your encoding was set to. If you don't set one, it's the default. If you didn't set a default it's Buffer. Setting it is done via options, for exaple gm.registerWorker('foo',{encoding: 'base64'},…) Of course, you can change the default when setting up the connection with the defaultEncoding option. (As you are in your code.)

Tasks are DuplexCombination objects which emits both read and write errors as error events, so you can catch them that way. However at this time its not possible to distinguish the two– some experimentation/testing around how this works out in various scenarios is called for.

If you look in task.js, you can see how it converts from stream to promise, that is:

var concat = require('concat-stream');
Task.prototype._makePromise = function () {
    var self = this;
    this.promise = new Promise(function(resolve,reject) {
        self.pipe(concat(function(body) { resolve(body) }));
        self.once('error', function (err) { reject(err) });
    });
}

[amv]

What if we try to split this into smaller questions?

The first one for me is that I find that the streaming workers and the non-streaming workers seem to have a big distinction on how they deal with the payload: Either it is immediately synchronously available or you might still need to wait for it through a pipe or a promise.

Question 1: Do you think this is a big enough difference in how the workers should be implemented that this would warrant a different interface to register handlers with these two types of expectations?

Question 2: If you think one of these options could be considered the primary interface, which one would it be and why?

Question 3: Are there other differences in these two approaches? For example is there a need to have an different type of way to handle errors for these two approaches? Or is there a need to return/pass on the response in a different way?