Angular observable 发布和订阅解释/它是如何工作的

Question

Observable Post 的解释

setup.component.ts

import { Component, EventEmitter, OnInit, Output } from '@angular/core';
import { FormGroup, FormBuilder, Validators } from '@angular/forms';

import { Post } from './model/post';
import { PostService } from './service/post.service';

@Component({
  selector: 'setup',
  templateUrl: './setup.component.html',
  styleUrls: ['./setup.component.scss']
})
export class SetupComponent implements OnInit {

  @Output()
  change: EventEmitter<string> = new EventEmitter();

  postUsers(input){
    this.postService.postUser(input)
      .subscribe(
        post => {
          this.post = post
        },
        err => {
          console.log(err);
        });
  }

  clicked(value) {
    console.log(value);
    this.postUsers(this.input)
    // this.change.emit(value);
  }

  complexForm : FormGroup;
  constructor(private postService: PostService) {}

  post: Post[];

  ngOnInit() {}

}

post.service.ts

import { Injectable } from '@angular/core';
import { Http, Response, Headers, RequestOptions } from '@angular/http';
import { Post } from '../model/post';
import { Observable } from 'rxjs/Rx';

// Import RxJs required methods
import 'rxjs/add/operator/map';
import 'rxjs/add/operator/catch';

@Injectable()
export class PostService {

  constructor (private http: Http) {}
  private registerUrl = 'http://localhost:2001/api/users/register';

  postUser(user:Post) : Observable<Post[]>{
    let headers = new Headers({ 'Content-Type': 'application/json' });
    let options = new RequestOptions({ headers: headers });
    return this.http.post(this.registerUrl, user, options)
      .map(this.extractData)
      .catch(this.handleError);
  }

  private extractData(res: Response) {
    let body = res.json();
    return body || { };
  }
  private handleError (error: Response | any) {
    // In a real world app, you might use a remote logging infrastructure
    let errMsg: string;
    if (error instanceof Response) {
      const body = error.json() || '';
      const err = body.error || JSON.stringify(body);
      errMsg = `${error.status} - ${error.statusText || ''} ${err}`;
    } else {
      errMsg = error.message ? error.message : error.toString();
    }
    console.error(errMsg);
    return Observable.throw(errMsg);
  }

}

model/post.ts

export class Post {
    constructor(
        public id: number, 
        public name: string, 
        public email: string, 
        public password: string
    ){}
}

我了解 model/post.ts 的作用，它定义了属性类型，我需要澄清 setup.components.ts 中的 subscribe 方法。 Observable 在clicked() 内部被调用，但我想了解的是，这是如何做到的，因为应用程序可以访问，这样我就可以在操作完成后继续进行this.postUsers(this.input)

通常情况下，我会执行以下操作

this.postUsers(this.input)
  .then(function(){
  });

如果有人能解释它的工作原理以及如何确认帖子已完成然后运行下一个功能，我真的很高兴

即我有这个

clicked(value) {
    console.log(value);
    this.postUsers(this.input)
    // this.change.emit(value);
  }

但我保证我会这样做

  clicked(value) {
    console.log(value);
    this.postUsers(this.input)
      .then(function(){
        this.change.emit(value);
      });
  }

我怎样才能让它与 Observables 一起工作？我试图查看通过执行返回的内容

  clicked(value) {
    console.log(value);
    const runThis = this.postUsers(this.input);
    console.log(runThis);
    // this.change.emit(value);
  }

但它返回undefined。

Answer 1

与 promise 类似，让您的 postUsers 方法返回一个可观察的（不是订阅）

postUsers(input){
    return this.postService.postUser(input);
}

然后你可以订阅这个方法，就像在 Promise 中使用 then 一样。

clicked(value) {
    console.log(value);
    this.postUsers(this.input).subscribe((response)=> {
        this.change.emit(value);
    });
}

---

您还可以将 observable 转换为 Promise。这样你就不会糊涂了。

import 'rxjs/add/operator/toPromise';

postUsers(input){
    return this.postService.postUser(input).toPromise();
}

clicked(value) {
    console.log(value);
    this.postUsers(this.input)
      .then((res)=>{
        this.change.emit(value);
      });
}

请注意，我没有在回调中使用 function 关键字。如果我这样做了，this.change 将引用回调 change，因为 this 不会引用组件。

Answer 2

如何确认帖子已完成

当流触发complete 回调时，您将知道请求已完成。如果您传递观察者对象而不是回调，这是您传递给 subscribe 方法或 complete 方法的第三个回调。

我认为您将从了解运算符toPromise 的工作原理中受益。然后，您可以直接使用它或模拟它的逻辑。你可以使用它：

this.postUsers(this.input).toPromise()
  .then(function(){
     ...
  });

基本上它所做的是创建一个承诺并返回它。然后它订阅 this.postUsers 返回的 observable 并等待直到这个 observable 发出 completed 事件。之后，它使用它返回的最后一个事件来解决承诺。大致如下：

toPromise(observable) {
   let lastReturnedValue = null;
   return new Promise() {
      observable.subscribe(
      (v)=>{
            lastReturnedValue = v;
      },
      (e)=>{
            reject(e);
      },      
      (c)=>{
            resolve(lastReturnedValue);
      });
   }
}

就像toPromise 一样，您可以订阅可观察对象并监听complete 事件：

  clicked(value) {
    console.log(value);
    this.postUsers(this.input).subscribe(
    (v)=>{ // here you get the value and can save it },
    (e)=>{ // here you know that observable has error, similar to `catch` callback },
    ()=>{ // here you know that observable has completed, similar to `then` callback}
    );
    console.log(runThis);
    // this.change.emit(value);
  }

或者你可以返回 observable 让其他人听：

  clicked(value) {
    return this.postUsers(this.input)
  }

  ...
  clicked().subscribe( // all the code from previous example)