public interface | source

ObservableInput

ObservableInput 接口描述了所有值是一个 SubscribableOrPromise 或者此类型的值可以转化为能发出值的 Observable 。每个操作符都可以接收被该接口 annotate 过的参数，而不一定需要是 RxJS 的 Observables。

ObservableInput 继承了 SubscribableOrPromise ，拥有如下类型:

数组

数组可以被理解为，observables 会从左到右，一个接一个地发送数组的所有值，然后立即完成。

类数组

传递给操作符的数组也可以不是 JavaScript 内置的数组。它们还可以是其他的，例如，每个函数的 arguments 属性，DOM NodeList,或者实际上，拥有 length 属性（是个数字）的对象并且存储了大于0个数。

ES6 迭代器

操作符可以接收内置的和自定义的 ES6 迭代器，把它们当做 observables 通过顺序的发出迭代器的所有值，然后当迭代器完成的时候，触发完成。注意和数组的区别，迭代器不需要是有限的，应该创建一个永远不会完成的 Observables 使之成为可能。

注意，通过在 Symbol.iterator 方法中返回自身，你可以迭代迭代器实例。这意味着每个操作符接收可迭代对象，但是间接的，迭代器本身也可以。所有原生 ES6 的迭代器默认都是 Iterable 的实例，所以你不需要自己实现 Symbol.iterator 方法。

TypeScript Iterable 接口 issue

TypeScript ObservableInput 接口实际上缺乏 Iterables 的类型签名，由于一些项目造成的 issue (查看 this issue). 如果你想给操作符传递 Iterable, 首先将它转化为 any。当然要铭记，因为类型转化，你需要确保传递的参数确实实现了此接口。

Example:

用数组使用 merge 操作符

Rx.Observable.merge([1, 2], [4], [5, 6])
.subscribe(
  value => console.log(value),
  err => {},
  () => console.log('ta dam!')
);

// 日志：
// 1
// 2
// 3
// 4
// 5
// 6
// "ta dam!"

用类数组使用 merge 操作符

Rx.Observable.merge({0: 1, 1: 2, length: 2}, {0: 3, length: 1})
.subscribe(
  value => console.log(value),
  err => {},
  () => console.log('nice, huh?')
);

// 日志：
// 1
// 2
// 3
// "nice, huh?"

用 Iterable (Map) 使用 merge 操作符

const firstMap = new Map([[1, 'a'], [2, 'b']]);
const secondMap = new Map([[3, 'c'], [4, 'd']]);

Rx.Observable.merge(
  firstMap,          // 传递 Iterable
  secondMap.values() // 传递 iterator, 它本身就是 Iterable
).subscribe(
  value => console.log(value),
  err => {},
  () => console.log('yup!')
);

// 日志：
// [1, "a"]
// [2, "b"]
// "c"
// "d"
// "yup!"

用 generator（返回无限的 iterator）使用 from 操作符

// 无限的自增数列流
const infinite = function* () {
  let i = 0;

  while (true) {
    yield i++;
  }
};

Rx.Observable.from(infinite())
.take(3) // 因为是无限的，仅仅取前三个
.subscribe(
  value => console.log(value),
  err => {},
  () => console.log('ta dam!')
);

// 日志：
// 0
// 1
// 2
// "ta dam!"