pipe() should work recursively, and the free standing pipe() should work inside the pipe chain method about ixjs HOT 4 CLOSED

Does this not work?

function operator1(source) {
  return source.pipe(map(x => x * 3), filter(x => x % 2 == 0));
}

function operator2(source) {
  return map(x => x + 100)(source);
}

from([1,2,3]).pipe(operator1, operator2);

from ixjs.

@trxcllnt yes that does work.
I suppose I could either

1. just use your operator1 style, and still use pipe for a single operator
2. have the abstract operator function return either op() or [opA(),opB()] and use lodash flatten(castArray(operator())) the array before calling pipe(...operators).

I don't find your operator2 style intuitive. But if I was to adopt that, I would expect
pipe(map(x => x + 100))(source)
to work, but it has to be
pipe(source,map(x => x + 100))

If you're not going to have chaining operators on the source object, then being able to use standalone pipe() recursively within chaining pipe() would make a smooth experience.

For now, this achieves what I need :

const { from, pipe } = require('ix/asynciterable');
const { map,filter } = require('ix/asynciterable/operators');
const _ = require('lodash');

function joinOps(...ops) {
  return _.flatten(ops);
}

(async function() {
  try {

    function operator1() {
      return [
        map(x => x * 3),
        filter(x => x % 2 != 0)
      ];
    }

    function operator2() {
      return map(x => x + 100);
    }

    function logger(data) {
      console.log(data);
      return data;
    }

    try{
      let operators = [operator1,operator2];
      operators = _(operators).map(op => op()).flatten().value();
      let result = await from([1,2,3]).pipe(...operators).forEach(logger);
      console.log('done');
    }
    catch(err) {
      console.error('pipeline failed with error:', err);
    }

    console.log('finished');
  } catch(e) {
    console.log(e);
  } finally {
    process.exit();
  }
})();

from ixjs.

The style I described is how all the Ix operators are implemented; functions which return functions that take a source iterable and return a result iterable. pipe() is a specialization of left-fold over a list of functions with the signature (source: TSource) => TResult.

It sounds like you want a curried version of pipe, which is valid but straightforward to write yourself:

// could also be:
// const compose = (...ops) => (source) => source.pipe(...ops);
const compose = (...ops) => (source) => ops.reduce((source, op) => op(source), source);

function operator1() {
  return compose(map(x => x * 3), filter(x => x % 2 == 0));
}

function operator2() {
  return map(x => x + 100);
}

from([1,2,3]).pipe(operator1(), operator2());

from ixjs.

Thanks for that, I haven't had enough functional experience to write that quickly.
Yes, your compose() is what I was expecting from the standalone pipe().
This can't be a special need - could you add something like compose to the library?

from ixjs.