Angular Observables 和 Http

Question

我很难将我的大脑围绕在 Angular 中的 observables 上。我来自 PHP 的世界，那里的东西绝对不是异步的。

我有一个组件，它只显示一般主题的消息列表。现在，我有一个所有消息都属于的主题。如果主题不存在，则应该创建它。消息和主题调用都是通过 REST api 完成的。

在非异步世界中，我会按顺序对其进行编程。消息服务将查看主题是否存在。如果没有，则由主题服务创建它。获得主题后，它会获取该主题中的所有消息。

我知道您订阅了 observable，但是当需要按顺序发生一系列事情时会发生什么？ angular 2 http documentation 提供了一个非常简单的示例，即在拨打电话时更新英雄列表。

组件

export class NotificationListComponent implements OnInit {
    constructor(private _notificationService:NotificationService) {
}

***

ngOnInit() {
    this.getNotifications();
}

getNotifications() {
      this._notificationService.getNotifications()
        .subscribe(
            notifications => this.notifications = notifications,
            error => this.errorMessage = <any>error);
}

通知服务

...    
 getNotifications() {
     //  call the topic service here for general topic??

    return this.http.get('/messages?order[datesent]=DESC')
        .map((res) => {
            return res.json()["hydra:member"];
        })
        .map((notifications:Array<any>) => {
            let result:Array<Notification> = [];
            notifications.forEach((jsonNotification) => {
                var Notification:Notification = {
                    message: jsonNotification.message,
                    topic: jsonNotification.topic,
                    datesent: new Date(jsonNotification.datesent),
                };
                result.push(Notification);
            });
            return result;
        })
        .catch(this.handleError);
}
...

主题服务

  ...
 getGeneralTopic() {
    var generalTopic;

    this.http.get('/topics?name=general')
        .map((res) => {
            return res.json()["hydra:member"][0];
        })
        .do(data => console.log(data))
        .subscribe(generalTopic => res);
}
...

Answer 1

如何推理Observables？

Observables 处理流。流几乎可以是任何东西，但您可以将它们视为异步事件的抽象数组。这很有用，因为您现在可以更清楚地推理它们：

• 抽象 因为 Observable 可以产生任何类型的值：String, Boolean, Object
• array 因为 Observable 有 Operators，其工作方式类似于 JavaScript 的数组方法：map(), filter(), reduce()
• of 因为 Observable 是值的包装器
• 异步因为 Observable 可能或可能不执行
• 事件因为 Observable 需要被触发

什么时候使用Observables？

当您需要执行涉及多个步骤的任务或操作时，您通常希望使用 Observables。这可以作为您的起点，您可以稍后根据需要简化或增加复杂性。

您应该有一个“计划”，或者至少有一个模糊的想法，这些步骤应该是什么。听起来很明显，但是由于您不知道自己想要什么而出现许多问题/问题（;

一旦您计划了一项行动（作为一系列步骤），您就可以从任一端开始，但我认为最好从端开始。至少在您了解更多信息之前。

我有一个组件，它只显示一般主题的消息列表。现在，我有一个所有消息都属于的主题。如果主题不存在，则应该创建它。消息和主题调用都是通过 REST api 完成的。

在非异步世界中，我会按顺序对其进行编程。消息服务将查看主题是否存在。如果没有，则由主题服务创建它。获得主题后，它会获取该主题中的所有消息。

对于您的用例计划将是：["(create topic)", "select topic", "show messages"]。 messages 是抽象数组，select 和create 是异步事件。

如何使用Observable？

正如我上面所说，让我们从头开始 - "show messages"。

<div *ngFor="#message of messages"></div>

我们知道我们正在处理Observable.of(messages)（这是您手动创建它的方式）。接下来，您需要“填充”消息的流，您可以使用返回Observable 的Http 服务来完成。由于您从服务器获得的消息被Http 服务包裹在几个“层”中，我们可以利用 Observable 的能力链接操作符（操作符返回 Observable）并获取我们需要的消息：

  getMessages(topic) {
    this.http.get("/messages?topic=" + topic)
      .map(response => response.json())
      // chain other operators here...
      .filter(message => !message.private)
  }

你可以在这里使用任何你需要的Operators...这导致了 Observables 的下一件大事：

“热”和“冷”Observables

Observables 默认是冷。这意味着当您创建一个可观察对象时，您只需描述它应该做什么。它不会立即执行这些操作（如 Promises 那样），它需要被触发。

您可以通过订阅来触发它，或者使用 subscribe() 方法手动订阅，或者您可以让 Angular 使用 async 管道使其hot（订阅给你）。

  getMessages(topic) {
    this.http.get("/messages?topic=" + topic)
      .map(response => response.json())
      .subscribe(messages => this.messages = messages);
  }

观察变化

接下来要做的事情（或之前，因为我们要在计划中倒退）是"select topic"。观察所选主题的值并通过加载新消息来响应它的变化会很好。这可以通过Subject 完成。

Subject 是一种桥接器或代理，在 ReactiveX 的某些实现中可用，它既充当观察者又充当 Observable。因为它是一个观察者，所以它可以订阅一个或多个 Observable，并且因为它是一个 Observable，它可以通过重新发射它观察到的项目来传递它们，它也可以发射新的项目。

我们可以设置topic$ 这样做：

class ListComponent {
  public messages;
  public topic$ = new Subject();

  constructor(private http: Http) {
    this.getMessages('posts');
    this.topic$
      .subscribe(topic => this.getMessages(topic));
  }

  getMessages(topic: string) {....}

  selectTopic(topic: string) {
    this.topic$.next(topic)
  }
}

总结

如果不存在，最后一步是"(create topic)"。假设如果主题不存在，服务器将返回错误：

  getMessages(topic: string) {
    this.http.get(API_URL + topic)
      .map(response => response.json())
      .subscribe(
        messages => this.messages = messages, 
        error => this.createTopic(topic)
      );
  }

  createTopic(topic: string) {
    this.http.post(API_URL + topic, { body: JSON.stringify(topic) })
      .map(response => response.json())
      .subscribe();
  }

这是working plunker 的示例。如您所见，这并不难（50 行代码......）。您可以轻松地四处移动并在需要的地方创建服务。

Answer 2

事实上，响应式编程允许创建异步数据流。这意味着您可以利用

Angular2 的 HTTP 支持利用运算符（flatMap，...）将流链接在一起并实现复杂的处理链。这意味着您可以使用响应式编程/RxJS 的所有概念来使 HTTP 请求成为异步数据流的一部分。

这是一个简单的示例，它允许根据表单输入的值（发生更新时）执行请求：

this.inputControl.valueChanges.debounceTime(500).flatMap(
  val => {
    // Execute ab 
    return this.http.get(`http://...?filter=${val}`);
  }).subscribe((data) => {
    this.places = data;
  });

操作员还可以允许在 HTTP 上下文中轻松实现几个问题：

• 确保在按顺序执行请求时接收最新的请求（例如基于用户输入）。 switchMap 运算符的示例，在触发新请求时取消先前正在进行的请求

  this.ctrl.valueChanges.switchMap(
    val => {
      return this.http.get(`http://...?filter=${val}`);
    });
  

• 缓冲事件并在一段时间后触发最后一个事件。在根据输入的最新值执行请求之前等待 500 毫秒不活动的示例：

  this.ctrl.valueChanges.debounceTime(500).flatMap(
    val => {
      return this.http.get(`http://...?filter=${val}`);
    });
  

• 请求失败时重试。每 500ms 和 2s 内重试的示例

  var params = new URLSearchParams();
  params.set('placename_startsWith', searchText);
  params.set('username', 'XXX');
  
  return this.http.get('http://api.geonames.org/postalCodeSearchJSON',
        { search: params })
    .retryWhen(error => error.delay(500))
    .timeout(2000, return new Error('delay exceeded'))
    .map(res => res.json().postalCodes);
  

这里有一些关于反应式编程的文章可以帮助你：

• André Staltz 所缺少的响应式编程简介 - https://gist.github.com/staltz/868e7e9bc2a7b8c1f754
• Rob Wormald 的一切都是流 - http://slides.com/robwormald/everything-is-a-stream

Answer 3

虽然我不确定是否理解您的问题，但Observable.selectMany 和Observable.and/thenDo/when 可能会为您提供帮助。

https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/selectmany.md

https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/when.md

在你的情况下，我认为它会让你创建一个获取和插入的事务。