Returning Errors From sequence()

[paulxtiseo]

The sequence() function has a track parameter that allows the function to return resolved data. I would like to propose that one adds functionality to track/capture rejected data too.

My particular scenario would be in file streaming. I use sequence() to process the data in large files into a database. Some rows fail from being inserted due to a variety of reasons. When an error is thrown/rejected, the whole run ends. In some cases, I would rather have an option that the sequence continues (add a skip boolean param) and/or that the problem rows be collected and reported (add a trackErrors param).

Looking at the code, my first idea would be to implement via checking in each catch() in sequence() for those params. The skip == true would result in no rejections and the trackErrors == true would, much like track, push data into the results.

Any suggestions or alternatives? I'd be willing to put in a pull request if so desired.

[vitaly-t]

If a promise request within the source of your sequence may reject but not fail, then simply do not use that request as a return value. Instead, wrap it up in a promise chain that never fails:

function source(index, data, delay) {
    if (/*condition*/) {
        return this.query('insert into users(...) values(...)')
            .catch(function (error) {
                return promise.resolve({
                    success: false,
                    result: error
                });
            });
    }
}

This will override any failure. You can also add override for the resolve, if you want to uniform the result for easier processing after the resolution, like this:

function source(index, data, delay) {
    if (/*condition*/) {
        return this.query('insert into users(...) values(...)')
            .then(function (data) {
                return {
                    success: true,
                    result: data
                };
            })
            .catch(function (error) {
                return promise.resolve({
                    success: false,
                    result: error
                });
            });
    }
}

With this you will have your sequence carry on regardless of the insert result, while within your processing function you can easily see which request was successful and which failed.

[paulxtiseo]

Actually, now that I consider your thoughts and look again, I see that the problem lies one layer above:

    api.parsers.processCsvStream = function(passedStream, processor) {
      return spex.stream.read(passedStream.pipe(csv.parse()), function(index, data, delay) {
        return spex.sequence(function(index) {
          if (index < data.length) {
            return processor(data[index]);
          }
        });
      });
    }

The read() function does not have a way to return anything but the {calls, reads, length, duration} construct? Assuming I catch() and return a resolve() per your example above, this still gets ignored in the read(), right?

[vitaly-t]

The read function doesn't have anything to return, except either a failure or success statistics.

You can bring it up one level, if you want, make your code ignore the result from the sequence:

api.parsers.processCsvStream = function (passedStream, processor) {
    return spex.stream.read(passedStream.pipe(csv.parse()), function (index, data, delay) {
        return spex.sequence(function (index) {
                if (index < data.length) {
                    return processor(data[index]);
                }
            })
            .catch(function () {
                return promise.resolve();
            });
    });
}

And by the way, I'm not sure if your csv.parse() ever returns more than one row at a time, and if so, then you do not even need to use sequence at all, just return the value:

api.parsers.processCsvStream = function (passedStream, processor) {
    return spex.stream.read(passedStream.pipe(csv.parse()), function (index, data, delay) {
        if (data.length) {
            return data[0];
        }
    });
}

[paulxtiseo]

Actually, I would like to end up with a success "report", and the {calls, reads, length, duration} object currently returned by read() is perfect, or a log of failures (array of objects, in this case the individual rows that fail in processor()) from processCsvStream(). I think read() is currently not designed todo anything but return the "report" object?

PS: The function processor() does the save-to-the-db, so you second, slimmer version does skip the call and so does not apply for my particular use case, although I do see what you mean about making sure to not over-nest calls.

[vitaly-t]

Method read doesn't deal with the data resolution, only with the data reading. It is method sequence that deals with the sequence of resolved data.

If you want method sequence to report the data it either succeeded or failed to process, have it push the results in an array either within this context or an external one, and then you can access it when method read resolves.

[paulxtiseo]

But, read wraps sequence. No matter what sequence returns, read resolves to {calls, reads, length, duration} or an error based on whether read resolves or not, correct?

[paulxtiseo]

I ended up tracking it myself, per your suggestion, although I still think it (returning errors) would be a great addition to any function that has a track param (which returns wanted data).

return promise.map(this.mappedFiles, function(file) {
  var errLog = [];
  return api.parsers.processCsvStream(stream, query, errLog)
    .then(function(data) {
      api.log('ImportFilemakerDataFromCsv > Finished processing file: ' + file.file_name, 'info');
      data.file = file;
      data.errors = errLog;
      return data;
    }, function(reason) {
      api.log('ImportFilemakerDataFromCsv > Error processing file: ' + file.file_name, 'info', reason);
      reason.file = file;
      data.errors = errLog;
      return reason;
    });
});

and

  api.parsers.processCsvStream = function(passedStream, processor, errLog) {
    return spex.stream.read(passedStream.pipe(csv.parse()), function(index, data, delay) {
      return spex.sequence(function(index) {
        if (index < data.length) {
          return processor(data[index])
            .catch(function(qryErr){
              // catch  and record errors, but otherwise don't stop the scanning
              errLog.push({ line: index, error: qryErr })
              return promise.resolve();
            });
        }
      });
    });

[vitaly-t]

Method read does return the error, as per its documentation, parameter receiver:

If the function throws an error or returns a rejected promise, the method rejects with the same error / rejection reason.