如何在 Angular2 中的可观察对象中创建可观察对象

Question

我可能会离开这个过程，但这里是：

我有一个 angular2 服务。该服务的数据源将是本地存储...稍后在使用 http 的数据库调用返回时可选地更新。因为我想在各种来源返回时更新返回的数据，所以我想使用 observables。现在，我只是想把这个概念放下来，所以我跳过了本地存储方面......但我包括了“背景故事”，所以它（有些）对我为什么想要做this 在多种方法中。

我的想法是我将有一个“getHTTPEvents()”方法，该方法将返回一个可观察对象，其有效负载是来自数据库的事件。 （理论是在未来的某个时候我也会有一个'getLSEvents（）'方法可以在那里捎带）

为了模拟它，我有这个代码：

private eventsUrl = 'app/mock-events.json';
getHTTPEvents() : Observable<Array<any>> { 
    return this._http.get(this.eventsUrl)
        .map(response => response.json()['events'])
        .catch(this.handleError); // handle error is a logging method
}

我的目标是创建一个方法，允许过滤返回的事件，但仍向服务用户返回一个可观察对象。这就是我的问题所在。有了这个目标，我有一个公共方法将由​​服务的用户调用。 （尝试从这里使用模式https://coryrylan.com/blog/angular-2-observable-data-services）

public getEvents(key:string,value:string) : Observable<Array<any>> {
    var allEventsObserve : Observable<Array<any>> = this.getHTTPEvents();
    var filteredEventsObserve : Observable<Array<any>>;


    allEventsObserve 
    .subscribe(
        events => {
            for(var i=0;i<events.length;i++) {
                if(events[i][key]==value) {
                    console.log('MATCH!!!' + events[i][key]); // THIS WORKS!
                    return new Observable(observer => filteredEventsObserve = observer);  // what do I need to return here?  I want to return an observable so the service consumer can get updates
                }
            }
            return allEventsObserve
        },
        error =>  console.error("Error retrieving all events for filtering: " + error));

}

以上行不通。我观看了很多视频并阅读了很多关于 observables 的教程，但除了创建和使用 http observable 之外，我发现没有什么比这更深入了。

我进一步尝试了这种制作新可观察对象的方法：

var newObs = Observable.create(function (observer) {
                    observer.next(events[i]);
                    observer.complete(events[i]);
                });

虽然至少可以编译，但我不确定如何在正确的时间“返回”它......因为我无法在 allEventsObserve.subscribe 方法之外“创建”它（因为“事件”没有t 存在）并且不能（似乎）从订阅中“返回”它。我也不完全确定我将如何“触发”“下一个”......？

我是否需要修改 allEventsObserve 中的数据并以某种方式简单地返回它？我是否使用正确的有效载荷创建了一个新的 observable（如上面所尝试的） - 如果是，我如何触发它？等等...我在这里检查过：How to declare an observable on angular2 但似乎无法理解“第二个”可观察对象是如何被触发的。也许我的整个范式都错了？

Answer 1

您似乎误解了 RxJS 运算符（如 map、filter 等）实际返回的内容，我认为纠正这一点将使解决方案变得清晰。

考虑这个简短的例子：

allEventsObserve
  .map(events => {
    return 'this was an event';
  })

当然，这是一个非常无用的示例，因为来自events 的所有数据都丢失了，但现在让我们忽略它。上面代码的结果不是字符串数组或其他任何东西，它实际上是另一个Observable。这个 Observable 只会为 allEventsObserve 发出的每个事件数组发出字符串 'this was an event' 这允许我们在 observables 上链接运算符 - 链中的每个运算符返回一个新的 Observable ，它发出已修改的项目以某种方式成为前一个运算符。

allEventsObserve 
  .map(events => {
    return 'this was an event';
  }) 
  .filter(events => typeof events !== 'undefined') 

allEventsObserve 显然是 Observable，allEventsObserve.map() 的计算结果为 Observable，allEventsObserve.map().filter() 也是如此。

所以，既然您希望您的函数返回一个Observable，那么您现在还不想调用订阅，因为这样做会返回一些不是真正的Observable。

考虑到这一点，您的代码可以通过以下方式重写：

public getEvents(key:string,value:string) : Observable<Array<any>> {
    var allEventsObserve : Observable<Array<any>> = this.getHTTPEvents();

    return allEventsObserve 
      .map(events => {
          var match = events.filter(event => event[key] == value);
          if (match.length == 0) {
            throw 'no matching event found';
          } else {
            return match[0];
          }
      })
      .catch(e => {
        console.log(e);
        return e;
      });

}

由于getEvents 返回Observable，因此您可以在代码中的其他位置执行getEvents().subscribe(events => processEvents()) 之类的操作来与它们交互。此代码还假定this.getHTTPEvents() 返回Observable。

另外，请注意我将您的 for 循环更改为对 filter 的调用，它对数组进行操作。在这种情况下，events 是一个普通的 JavaScript Array，因此被调用的 filter 与 RxJS 运算符 filter 不同的 filter。