import { Component } from '@angular/core';
import { Observable, Subscription } from 'rxjs';

@Component({
 selector: 'app-home',
 template: `<input type="text"> `
})
export class HomeComponent {
 ngOnInit() {
  const node = document.querySelector('input[type=text]');

  // 第二个参数 input 是事件名，对于input元素有一个 oninput 事件用于接受用户输入
  const input$ = Observable.fromEvent(node, 'input');
  input$.subscribe({
   next: (event: any) => console.log(`You just typed ${event.target.value}!`),
   error: (err) => console.log(`Oops... ${err}`),
   complete: () => console.log(`Complete!`)
  });
 }
}

input$.subscribe((event: any) => {

});

Observable.fromEvent(node, 'input')
 .map((event: any) => event.target.value)
 .filter(value => value.length >= 2)
 .subscribe(value => { console.log(value); });

ngOnDestroy() {
  this.inputSubscription.unsubscribe();
}

private alive: boolean = true;
ngOnInit() {
 const node = document.querySelector('input[type=text]');

 this.s = Observable.fromEvent(node, 'input')
  .takeWhile(() => this.alive)
  .map((event: any) => event.target.value)
  .filter(value => value.length >= 2)
  .subscribe(value => { console.log(value) });
}

ngOnDestroy() {
 this.alive = false;
}

@Injectable()
export class MessageService {
  private subject = new Subject<any>();

  send(message: any) {
    this.subject.next(message);
  }

  get(): Observable<any> {
    return this.subject.asObservable();
  }
}

constructor(public srv: MessageService) { }

ngOnInit() {
  this.srv.send('w s k f m"htmlcode">

constructor(private srv: MessageService) {}

message: any;
ngOnInit() {
  this.srv.get().subscribe((result) => {
    this.message = result;
  })
}



EventEmitter
其实EventEmitter跟RxJS没有直接关系，因为他是Angular的产物，而非RxJS的东西。或者我们压根没必要去谈，因为EventEmitter就是Subject。
EventEmitter的作用是使指令或组件能自定义事件。


@Output() changed = new EventEmitter<string>();

click() {
  this.changed.emit('hi~');
}







@Component({
 template: `<comp (changed)="subscribe($event)"></comp>`
})
export class HomeComponent {
 subscribe(message: string) {
   // 接收：hi~
 }
}



上面示例其实和上一个示例中 MessageService 如出一辙，只不过是将 next() 换成 emit() 仅此而已。
结论
RxJS最难我想就是各种operator的应用了，这需要一些经验的积累。
RxJS很火很大原因我认还是提供了丰富的API，以下是摘抄：
创建数据流：

  
单值：of, empty, never
  
多值：from
  
定时：interval, timer
  
从事件创建：fromEvent
  
从Promise创建：fromPromise
  
自定义创建：create

转换操作：

  
改变数据形态：map, mapTo, pluck
  
过滤一些值：filter, skip, first, last, take
  
时间轴上的操作：delay, timeout, throttle, debounce, audit, bufferTime
  
累加：reduce, scan
  
异常处理：throw, catch, retry, finally
  
条件执行：takeUntil, delayWhen, retryWhen, subscribeOn, ObserveOn
  
转接：switch

组合数据流：

  
concat，保持原来的序列顺序连接两个数据流
  
merge，合并序列
  
race，预设条件为其中一个数据流完成
  
forkJoin，预设条件为所有数据流都完成
  
zip，取各来源数据流最后一个值合并为对象
  
combineLatest，取各来源数据流最后一个值合并为数组

另，最好使用 $ 结尾的命名方式来表示Observable，例：input$。
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。