数组上的可观察 concat 方法返回相同的数组

Question

在我的服务中，我尝试解决下一个问题。我有一个包含其他 json 文件（卡片）名称的 json 文件：

{
  "filename1" : ... ,
  "filename2" : ... ,
  ...
  "filenameN" : ...
}

我需要加载所有这些文件。 “filenameX”文件有一些卡片数据：

{
  dataX
}

我需要在对象中合并加载的数据：

{
  "filename1" : { data1 },
  "filename2" : { data2 },
...
  "filenameN" : { dataN }
}

我为任何文件加载创建观察者，并尝试将它们组合成一个高级单个观察者，当所有相应的观察者都存在时，该观察者被解析。这是我的代码：

import { Observable } from "rxjs/Observable";
import "rxjs/add/observable/concat";
import "rxjs/add/operator/map";
...
    _loadCards(dir, cardsListFile) {
        var http = ...
        var url = ...
        return Observable.create(function (observer) {
            http.get(url + dir + "/" + cardsListFile + ".json").map(res => { return res.json(); }).subscribe(list => {
                let data = {};
                let observers = [];
                for(var card in list) {
                    if(list.hasOwnProperty(card)) {
                        let obs = http.get(url + dir + "/cards/" + card + ".json").map(res => { return res.json(); });
                        observers.push(obs);
                        let getData = function(card) {
                            obs.subscribe(cardData => {
                                data[card] = cardData;
                            });
                        };
                        getData(card);
                    }
                }

                let concatResult = Observable.concat(observers);
                console.log(concatResult);
                concatResult.subscribe(result => {
                    observer.onNext(data);
                    observer.onCompleted();
                });         
            });
        });
    };

但是，Observer 的 concat 运算符并不能像描述的那样工作 - 它返回相同的 Observers 数组作为其输入。问题出在哪里，我可以使用哪些其他运算符来使我的解决方案更直接（因为现在它肯定很难看）？

Answer 1

我不完全确定我理解您想要做什么，但 concat() 运算符将其参数解包而不是作为数组。

这意味着 Observable.concat(observables) 将重新发送 Observable，就像它们在 observables 数组中一样。

您想要的是传递解压缩的数组Observable.concat(...observables)，它与以下内容相同：

Observable.concat(observable1, observable2, observable3 ,...)

我认为这模拟了您正在尝试做的事情。它使用 forkJoin 而不是 concat 所以它会等到所有 Observables 完成。

var input = {
  "filename1": 'data1',
  "filename2": 'data2',
  "filenameN": 'dataN'
};

let keys = Object.keys(input);

let observables = keys.map(key => {
  return Observable.of(input[key]).map(s => s.split("").reverse().join(""));
});

Observable.forkJoin(observables, (...results) => {
    let combined = {};
    keys.forEach((key, i) => {
      combined[key] = results[i];
    });
    return combined;
  })
  .subscribe(val => console.log(val));

这会打印到控制台：

{ filename1: '1atad', filename2: '2atad', filenameN: 'Natad' }

观看现场演示：https://jsbin.com/coruyu/4/edit?js,console

Answer 2

带有解决方案的JSBIN：http://jsbin.com/dahaqif/edit?js,console

我假设您有两个 observable，一个用于获取文件名列表，另一个用于获取特定文件名的数据：

const filenamesObs = Observable.of({
  "filename1" : null,
  "filename2" : null,
  "filename3" : null
});
// This is actually a function that returns an observable.
const filedataObs = (filename) => {
  return Observable.of(`This is the data for ${filename}`);
}

现在，您可以通过以下方式组合两个 observable 以获得您描述的数据结构：

const source = filenamesObs

  // Extract the list of filenames as an array.
  .map(obj => Object.keys(obj))

  // Flatten the array, i.e. emit each filename individually
  // vs a SINGLE array containing ALL filenames.
  .mergeMap(val => val)

  // Get the data for each filename.
  .mergeMap(filename =>
    filedataObs(filename).map(data => Object.assign({}, { filename: filename, data: data }))
  )

  // At this point, you have a stream of { filename, data } objects.
  // Reduce everything back to a single object.
  .reduce((acc, curr) => {
    acc[curr.filename] = curr.data;
    return acc;
  }, {});

如果您订阅并将结果记录到控制台，您将看到：

[Object] {
  filename1: "This is the data for filename1",
  filename2: "This is the data for filename2",
  filename3: "This is the data for filename3"
}

Arseniy 发表评论后的补充说明

• 第一个mergeMap() 是一个“技巧”，可以将我们一开始拥有的单个值数组转换为多个单独的值。由于您希望能够单独处理每个文件名（以获取相应的数据），因此更方便的是逐个接收文件名而不是作为包含所有文件名的大数组。为了清楚起见，您应该转到我的 JSBIN 并添加行 .do(console.log) before 和 after 第一个 mergeMap()。您会立即明白其中的区别。
• 第二个mergeMap() 将源 observable 的值“投影”到目标 observable 中。这听起来很花哨，但这只是意味着我们正在将文件名（第一个 observable）转换为 HTTP 请求以获取文件的数据（第二个 observable）。
• 最后，Object.assign() 让我们将所有东西重新组合在一起。由于我们有来自两个 observable 的数据——一个发出文件名，一个发出文件数据——我们需要将所有内容合并到一个容器中，然后才能使用它（或进一步转换它）。为此，我使用Object.assign() 创建具有两个属性filename 和data 的临时对象。请注意，这是 vanilla JavaScript，它与 observables 无关。同样，您可以在第二个 mergeMap() 之后添加 .do(console.log) 行，以将此时流包含的内容打印到控制台。

Answer 3

规则 #1：切勿在其他订阅中使用自定义订阅 - 可能有 0.1% 的情况确实需要这样做。

我已经稍微改写了你的流，它是如何完成的：

_loadCards(dir, cardsListFile) {
    var http = ...
    var url = ...
    return http.get(url + dir + "/" + cardsListFile + ".json")
        .map(res => res.json())
        .switchMap(fileMap => Rx.Observable.from(Object.keys(fileMap)))
        .concatMap(card => http.get(url + dir + "/cards/" + card + ".json")
               .map(res => ({[card]: res.json()})))
        .map(arrayOfCardResultObjects => Object.assign.apply({}, arrayOfCardResultObjects));
};

---

调整后的版本，可以作为 sn-p 实时运行：

getMockedCardMap()
    .switchMap(fileMap => Rx.Observable.from(Object.keys(fileMap)))
    .concatMap(card => getMockedCardFromFile(card).map(res => ({[card]: res})))
    .toArray()
    .map(arrayOfCardResultObjects => Object.assign.apply({}, arrayOfCardResultObjects))
    .subscribe(console.log);


// Mocked helper-methods
function getMockedCardMap() {
    return Rx.Observable.of({
        "file1": "foo",
        "file2": "bar",
        "file3": "baz"
    });
}

function getMockedCardFromFile(file) {
    return Rx.Observable.of("Result from: " + file);
}

<script src="https://unpkg.com/rxjs/bundles/Rx.min.js"></script>