javascript - 对流中的每个元素应用多个函数

Question

我正在尝试获取流元素，并一次对一个元素应用一系列函数。每个函数都是多个异步操作之一(下载、图像转换、通过网络上传等)。这些库使用不同的 API(一些 Promises，一些回调)，因此如果我能弄清楚如何实现这项工作，那么使用 Observables 可能会简化 API。

使用rxjs我正在尝试找出正确执行此操作的语法，这是我到目前为止所尝试的:

const Rx = require('rxjs');
const { download } = require('./lib/images');
const { transform } = require('./lib/operations');
const program = require('./lib/bin');
const util = require('./lib/util');

const files = util.fixFileExtensions(program.args);
const files$ = Rx.Observable.fromPromise(getMetadata(files));
const download$ = Rx.Observable.bindCallback(download);
const transform$ = Rx.Observable.fromPromise(transform);

// Take the files, apply the download$ and transform$ to each element in the stream
files$
    .map(download$)
    .map(transform$)
    .subscribe();

但是我的语法错误并且出现错误。使其工作的语法是什么？

编辑根据当前的评论和答案，这是更新的，但它仍然没有将 download 的输出传递给 transform。这是我所拥有的:

function download({id, name}, callback) {
    const credentials = {....};
    const filePath = `/path/to/tmp/${name}`;
    const stream = fs.createWriteStream(filePath);
    authenticate(credentials, err {
        if (err) throw err;
        files.download(...., (err, res) => {
            res.data
                .on('end', () => callback(filePath))
                .on('error', err => throw err)
                .on('data', /* write to file */)
                .pipe(stream)
        });
    });
}

function transform(filename, callback) {
     const transformations = /* load transformations */;
     const result = transformations.map(t => applyTransformation(filename, t));
     callback(result); 
}

const download$ = Rx.Observable.bindCallback(download);
const transform$ = Rx.Observable.bindCallback(transform);
files$
    .flatMap(file => file)
    .switchMap(download$)
    .map(transform$)
    .subscribe(console.log, console.error, () => console.log('map is done'));

这会从我的 download 函数中抛出一个错误，提示 TypeError:callback is not a function，因此 download 被正确调用，但其输出没有被传递给 transform

Answer 1

最后需要使用switchMap来应用异步操作。 Switchmap将订阅内部Observable并解析它:

files$
    .switchMap(download$)
    .switchMap(transform$)
    .subscribe();

如果订阅中需要下载值，则必须传递内部值:

files$
    .switchMap(download$)
    .switchMap(transform$, (outerValue, innerValue) => {outerValue, innerValue})
    .subscribe(valueOfDonwload => console.log(valueOfDonwload));