lyyourc/RxJS 5 Operators By Example.md

## RxJS 5 Operators By Example.md

      
    Raw
  

              RxJS 5 Operators By Example.md
            
          
    用例子讲解 RxJS 5 的 Operators 。
原文链接：RxJS 5 Operators By Example by @btroncone
目录


buffer
bufferCount
bufferTime
bufferToggle
bufferWhen
combineAll
combineLatest
concat
concatAll
concatMap
concatMapTo
count
debounce
debounceTime
defaultIfEmpty
delay
delayWhen
dematerialize
distinctUntilChanged
do
every
expand
filter
first
groupBy
ignoreElements
last
map
mapTo
merge
mergeMap
pluck
publish
race
repeat
retry
retryWhen
sample
scan
share
single
skip
skipUntil
skipWhile
startWith
switchMap
window
windowCount
windowTime
windowToggle
windowWhen
withLatestFrom
zip

buffer

signature: buffer<T>(closingNotifier: Observable<any>): Observable<T[]>

缓冲所有输出的值，直到他们被提交。重复之。。。
(demo | official docs)
// 创建一个 observable ，每秒 emit 一个值
const myInterval = Rx.Observable.interval(1000)

// 创建一个 observable ，每当点击 document 的时候，emit 一个值
const bufferBy = Rx.Observable.fromEvent(document, 'click')

// 缓冲 interval observable emit 的所有值
// 直到我们点击了 document ，这会使得 bufferBy Observable emit 一个值
// 最后，把缓冲好的值，当作一个数组输出
const myBufferedInterval = myInterval.buffer(bufferBy)

// 打印到控制台
// [1,2,3]...[4,5,6,7,8]
const subscribe = myBufferedInterval.subscribe(val => console.log('Buffered Values', val))
bufferCount

signature: bufferCount<T>(bufferSize: number, startBufferEvery: number = null): Observable<T[]>

缓冲所有的输出值，直到某个数字被 fullfilled ，然后把它们 emit 。重复之。。。
(demo | docs)
// 创建一个 Observable ，每秒 emit 一个 value 。
const myInterval = Rx.Observable.interval(1000)

// 当 emit 了三个值之后，把它们作为数组，输出。
const bufferThree = myInterval.bufferCount(3)

// 打印到控制台
// 比如，输出 [0,1,2]...[3,4,5]
const subscribe = bufferThree.subscribe(val => console.log(`Buffered Values:`, val))

/*
`bufferCount` 的第2个参数：什么时候开始下一个 buffer
比如，bufferCount(3, 1)

第一个时间间隔后：
buffer 1: [0]

第二个时间间隔后：
buffer 1: [0,1]
buffer 2: [1]

第三个时间间隔后：
buffer 1: [0,1,2]
buffer 2: [1,2]
buffer 3: [2]

第四个时间间隔后：
buffer 2: [1,2,3]
buffer 3: [2,3]
buffer 4: [3]
*/

const bufferEveryOne = myInterval.bufferCount(3, 1)
// 打印到控制台
const secondSubscribe = bufferEveryOne.subscribe(val => console.log('Start Buffer Every 1:', val))
bufferTime

signature: bufferTime(bufferTimeSpan: number, bufferCreationInterval: number, scheduler: Scheduler): Observable<T[]>

缓冲输出的值，直到达到指定的时间点，然后把他们 emit 。重复之。。。
(demo | docs)
// 创建一个 observable ，每隔 500ms 就会 emit 一个值
const myInterval = Rx.Observable.interval(500)

// 2秒后，把所有缓冲的值，作为一个数组输出
const bufferTime = myInterval.bufferTime(2000)

// 输出值到控制台
// 比如，输出 [0,1,2]...[3,4,5,6]
const subscribe = bufferTime.subscribe(val => console.log(`Buffered with Time:`, val))

/*
类似地，`bufferTime` 的第2个参数：什么时候开始下一个 buffer
比如，bufferTime(2000, 1000)
可能会输出：[0,1,2]...[1,2,3,4,5]...[3,4,5,6,7]
*/
const bufferTimeTwo = myInterval.bufferTime(2000, 1000)
// 输出到控制台
const secondSubscribe = bufferTimeTwo.subscribe(val => console.log(`Start buffer Every 1s`, val))
bufferToggle

signature: bufferToggle(openings: Observable<O>, closingSelector: Function): Observable<T[]>

打开 buffer ，使其捕捉 source emit 的值，关闭 buffer ，使其 emit 缓冲了的值。
(demo | docs)
// 每隔1秒 emit 值
const sourceInterval = Rx.Observable.interval(1000)
// 在5秒后，emit 值。接着每隔5秒 emit 值
const startInterval = Rx.Observable.interval(5000)

// 在3s后 emit 值，并且关闭对应的 buffer
const closingInterval = val => {
  console.log(`Value ${val} emitted, starting buffer! Closing in 3s!`)
  return Rx.Observable.interval(3000)
}

// 每隔5s就会开始一个新的 buffer ，缓冲3s内的所有 emit 过的值，然后 emit 缓冲好的值
const bufferToggleInterval = sourceInterval.bufferToggle(startInterval, closingInterval)

// 打印到控制台
// Emitted Buffer：[4,5,6]...[9,10,11]
const subscribe = bufferToggleInterval.subscribe(val => console.log('Emitted Buffer:', val))
bufferWhen

signature: bufferWhen(closingSelector: function): Observable<T[]>

缓冲所有的值直到 closing selector emit 值，接着 emit 缓冲好的值
(demo | docs)
// 每隔1s emit 一个值
const oneSecondInterval = Rx.Observable.interval(1000)
// 每隔5秒 emit 一个值
const fiveSecondInterval = Rx.Observable.interval(5000)

// 每隔5秒 emit 缓冲了的值
const bufferWhenExample = oneSecondInterval.bufferWhen(fiveSecondInterval)

// 打印到控制台
//[0,1,2,3]...[4,5,6,7,8]
const subscribe = bufferWhenExample.subscribe(val => console.log('Emitted Buffer', val))
combineAll

signature: combineAll(project: function): Observable

当 outer Observable complete 的时候，输出 inner Observable 最新的值
(demo | docs)
// 5s 后 emit 值，然后 complete
const fiveSecondTimer = Rx.Observable.timer(5000)

// 当 timer(outer Observable) 触发以及完成时，
// 输出 inner Observable 最新的值，在例子中就是一个单独的值
const example = fiveSecondTimer.mapTo(Rx.Observable.of('Hello', 'World'))
const combined = example.combineAll()

// 输出：['Hello']...['World']
const subscribe = combined.subscribe(val => console.log(`Values from inner observable`, val))

// 我们也可以传一个 project function 去接收 emit 的值
const fiveSecondTimer = Rx.Observable.timer(5000)
const example = fiveSecondTimer.mapTo(Rx.Observable.of('Hello', 'Goodbye'))
const combined = example.combineAll(val => `${val} Friend!`)

// 输出：`Hello Friend!' ... 'Goodbye Friend!'
const subscribeProjected = combined.subscribe(val => console.log(`Values Using Projection`, val))
combineLatest

signature: combineLatest(observables: ...Observable, project: function): Observable

给定一组 Observable ，当其中一个 emit 的时候，其他的 emit 最新的值
(demo | docs)
// `timerOne` 会在第1s的时候 emit 第一个值，接着每隔4s emit
const timerOne = Rx.Observable.timer(1000, 4000)
// `timerTwo` 会在第2s的时候 emit 第一个值，接着每隔4s emit
const timerTwo = Rx.Observable.timer(2000, 4000)
// `timerThree` 会在第3s的时候 emit 第一个值，接着每隔4s emit
const timerThree = Rx.Observable.timer(3000, 4000)

// 当其中一个 timer emit 的时候，其他的也同时 emit 最新的值，作为一个数组返回
const combined = Rx.Observable
  .combineLatest(
    timeOne,
    timeTwo,
    timeTrhee
  )
  
const subscribe = combined.subscribe(latestValues => {
  const [timerValOne, timerValTwo, timerValThree] = latestValues
  
  /*
  * timerOne first tick: timer One Latest: 1, Timer Two Latest: 0, Timer Three Latest: 0
  * timerOne first tick: timer One Latest: 1, Timer Two Latest: 1, Timer Trhee Latest: 0
  * timerOne first tick: timer One Latest: 1, Timer Two Latest: 1, Timer Trhee Latest: 1
  */
  console.log(
    `Timer One Latest: ${timerValOne}`,
    `Timer Two Latest: ${timerValTwo}`,
    `Timer Three Latest: ${timerValThree}`
  )
})

// `combineLastest` 也可以接收一个可选的参数：projection function
const combineProject = Rx.Observable(
  timerOne,
  timerTwo,
  timerThree,
  (one, two, three) => {
    return `Timer One (Proj) Latest: ${one},
      Timer Two (Proj) Latest: ${two},
      Timer Three (Proj) Latest: ${three}
    `
  )
  
const subscribe = combineProject.subscribe(latestValuesProject => console.log(latestValuesProject))
concat

signature: concat(observables: ...*): Observable

就像在 ATM 交易的队伍一样，直到上一个交易完毕，下一个才能开始
(demo | docs)
// emit 1,2,3
const sourceOne = Rx.Observable.of(1,2,3)
// emit 4,5,6
const sourceTwo = Rx.Observable.of(4,5,6)

// emit `sourceOne`，直到他 complete ，才开始 subscribe `sourceTwo`
const concatSource = sourceOne.concat(sourceTwo)

// 输出：1,2,3,4,5,6
const subscribe = concatSource.subscribe(val => console.log('Example 1: Basic concat:', val))

// 延迟3s再 emit
const delayedSourceOne = sourceOne.delay(3000)
// 等到 `delaySourceOne` complete 之后才开始 subscibe `sourceTwo`
const concatDelayedSource = delayedSourceOne.concat(sourceTwo)

// 输出：1,2,3,4,5,6
const subscribeDelayed = concatDelayedSource.subscribe(val => console.log('Example 2: Delayed source one:', val))


// 如果 `sourceOne` 永远不 complete ，那么永远也不会 subcribe 第二个 Observable
const sourceOneNeverComplete = Rx.Observable
  .concat(
    Rx.Observable.interval(1000),
    Rx.Observable.of('Thiis', 'Never', 'Runs')
  )
  .delay(5000)

// 输出：1,2,3,4....
const subscribeNeverComplete = sourceOneNeverComplete.subscribe(val => 
  console.log('Example 3: Source never complete, Second Observable never run', val)
)
concatAll

signature: concatAll(): Observable

用于 nestet Observalbe(observable of observables) ，当上一个 complete 的时候，subscribe 每个，然后合并（merge）每个 emit 的值
(demo | docs)
// 每隔2s emit 一个值
const sourceOne = Rx.Observable.interval(2000)
const example = sourceOne
  // 加10，并且返回一个 Observable
  .map(val => Rx.Observable.of(val + 10)
  // 合并从 inner Observable 返回的值
  .concatAll()
  
// 输出：'Example with Basic Observable 10', `Example with Basic Observable 11'...
const subscribe = example.subscribe(val => console.log('Exmaple with Basic Observable', val))


// 创建一个只有 resolve 的 promise
const samplePromise = val => new Promise(resolve => resolve(val))

const exampleTwo = sourceOne
  .map(val => samplePromise(val))
  // 合并从 samplePromise 来的值
  .concatAll()
  
// 输出：`Example with Promise 0`, `Example with Promise 1`
const subscribeTwo = exampleTwo.subscribe(val => console.log('Example with Promise', val))
concatMap

signature: concatMap(project: function, resultSelector: function): Observable

把 source Observable 的值映射到 inner Observable ，接着按顺序 subscribe 和 emit inner Observable 的值
相等于：map -> concat
(demo | docs)
// emit 'Hello' 和 'Goodbye'
const source = Rx.Observable.of('Hello', 'Goodbye')
// 把 source value 映射到 inner Observable ，当 complete 时，emit 结果，接着下一个。
const exampleOne = source.concatMap(val => Rx.Observable.of(`${val} World!`))

// 输出：`Example One: 'Hello World', Example One: 'Goodbye World'
const subscribe = exampleOne.subscribe(val => console.log('Example One:', val))


// 结合 promise 的例子
const examplePromise = val => new Promise(resolve => resolve(`${val} World`))
// 把 source value 映射到 inner Observable ，当 complete 时，emit 结果，接着下一个。
const exampleTwo = source.concatMap(val => examplePromise(val))

// 输出：`Example w/ Promise: 'Hello World', Exmaple w/ Promise: 'Goodbye World'
const subscribeTwo = exampleTwo.subscribe(val => console.log('Exmaple w/ Promise', val))


// 第一个参数返回的值，会传给第二个参数
const exampleWithSelector = source.concatMap(val => examplePromise(val), result => `${result} w/ selector!`)

// 输出：Example w/ Selector: 'Hello w/ Selector', Example w/ Selector: 'Goodbye w/ Selector'
const subscribeThree = exampleWithSelector
  .delay(2000)
  .subscribe(val => console.log('Exmaple w/ Selector', val))
concatMapTo

signature: concatMapTo(observable: Observable, resultSelector: function): Observable

当 source emit 的时候，总是 subscribe 相同的 Observable ，当 complete 时合并结果
(demo | docs)
// 每隔2s emit 一个值
const interval = Rx.Observable.interval(2000)
const message = Rx.Observable.of(`Second(s) elapsed!`)

// 当 `interval` emit 时，subscribe `message` 直到他 complete ，最后合并结果
const example = interval.concatMapTo(message, (time, msg) => `${time} ${msg}`)

// 输出：`0 Second(s) elapsed! 1 Second(s) elapsed!`
const subscribe = example.subscribe(val => console.log(val))

// 每隔1s emit 一个值，取前5个
const basicTimer = Rx.Observable.interval(1000).take(5)

/*
* ***注意*** 
* 像这种 source Observable emit 的速度比 inner Observable complete 快的情景下，
* 就会出现内存问题
* (`interval` emit 时间间隔为1s ，而 `basicTimer` complete 则需要5s)
*/

// `basicTimer` 会在 5s 后 complete ，emit 的值有：0,1,2,3,4
const exampleTwo = interval
  .concatMapTo(basicTimer, (firstInterval, secondInterval) => `${fisrtInterval} ${secondInterval}`)

/* 输出：
* 0 0
* 0 1
* 0 2
* 0 3
* 0 4
* 1 0
* 1 1
* continue...
*/

const subscribe = exampleTwo.subscribe(val => console.log(val))
count

signature: count(predicate: function): Observable

计算 source emit 值的数量，直到 complete 。
(demo | docs)
// emit 1,2,3 然后 complete
const threeItems = Rx.Observable.of(1, 2, 3)
// 当 `threeItems` complete 的时候，计算他 emit 值得数量
const exampleOne = threeItems.count()

// 输出：`Count from Example One: 3'
const subscribe = exampleOne.subscribe(val => console.log(`Count from Example One: `, val))


// 数组的计数
const myArray = [1, 2, 3, 4, 5]
const myObsArray = Rx.Observable.from(myArray)
const exampleTwo = myObsArray.count()

// 输出：'Count of Basic Array: 5'
const subscribeTwo = exampleTwo.subscribe(val => console.log(`Count of Basic Array: ${val}`))


// 可选参数：计数符合 predicate function 的值
const evensCount = myObsArray.count(val => val % 2 === 0)
// 输出：'Count of Even Numbers: 2'
const subscribeThree = evensCount.subscribe(val => console.log(`Count of Even Number: ${val}`))
debounce

signature: debounce(durationSelector: function): Observable

忽略 source Observable 某段时间内的 emit 的值。
(demo | docs)
// emit 4个字符串
const example = Rx.Observable.of('WAIT', 'ONE', 'SECOND', 'Last will display')

/*
* 仅仅 emit 最后一个 emission 过后的1s的值
* 其他的值丢掉
*/
const debouncedExmaple = example.debounce(() => Rx.Observable.timer(1000))

/*
* 在这个例子中，所有的值除了最后一个之外全部都忽略掉
* 输出：'Last will display'
*/
const subscribe = deboundedExmaple.subscibe(val => console.log(val))


// 每隔1s emit 一个值
const interval = Rx.Observable.interval(1000)
// 每隔1s 增加我们的 dobounce time
const debouncedInterval = interval.debounce(val => Rx.Observable.timer(val * 200))

/*
* 在5s后，debounce time 就大于 interval time
* 因此，未来的值都会被丢掉
* 输出：0...1...2...3...4......(此时 debounce time 大于1s ，再也没有值 emit)
*/

const subscribeTwo = debounceInterval.subscribe(val => console.log(`Example Two: ${val}`))
debounceTime

signature: debounceTime(dueTime: number, scheduler: Scheduler): Observable

忽略 source Observable 某段时间内的 emit 的值。
(demo | docs)
const input = document.getElementById('example')

// 对于每个 keyup 事件，映射为当前 input 的值
const example = Rx.observable
  .fromEvent(input, 'keyup')
  .map(i => i.currentTarget.value)
  
// 在 keyups 和 emit 当前值之间，等待 .5s
const debouncedInput = example.debounceTime(500)

// 输出
const subscribe = debouncedInput.subscribe(val => {
  console.log(`Debounced Input: ${val}`)
}
defaultIfEmpty

signature: defaultIfEmpty(defaultValue: any): Observable

当 Observable 为空时，使用这个设定的默认值，否则值为 null 。
(demo | docs)
const empty = Rx.Observable.of()
// 当 source Observable 的值为空时，使用这个默认值
const exampleOne = empty.defaultIfEmpty('Observable.of() Empty!')
// 输出：'Observable.of() Empty!'
const subscribe = exampleOne.subscribe(val => console.log(val))

// 空的 Observable
const emptyTwo = Rx.Observable.empty()
const exampleTwo = emptyTwo.defaultIfEmpty('Observable.empty()!')
// 输出：'Observable.empty()!'
const subscribe = exampleTwo.subscribe(val => console.log(val))
delay

signature: delay(delay: number | Date, scheduler: Scheduler): Observable

延迟 Observable emit 的时间。
(demo | docs)
const example = Rx.Observable.of(null)

const merge = example.merge(
  example.mapTo('Hello'),
  example.mapTo('World').delay(1000),
  example.mapTo('Goodbye').delay(2000),
  example.mapTo('World!').delay(3000)
)

// 输出：'Hello'...'World'...'Goodbye'...'World!'
const subscribe = merge.subscribe(val => console.log(val))
delayWhen

signature: delayWhen(selector: Function, sequence: Observable): Observable

根据指定的函数，延迟 emit 的时间。
(demo | docs)
// 每隔1s emit 一个值
const message = Rx.Observable.interval(3000)
// 在5s后 emit 值
const delayForFiveSeconds = () => Rx.Observable.timer(5000)
// 在5s后，开始 emit 值
const delayWhenExample = message.delayWhen(delayForFiveSeconds)

// 输出：延迟5s后，
// 5s....1...2...3
const subscribe = delayWhenExample.subscribe(val => console.log(val))
dematerialize

sigature: dematerialize(): Observable

把 notification object 变成 notification values
(demo | docs)
// emit next notification 和 error notification
const source = Rx.Observable
  .from([
    Rx.Notification.createNext('SUCCESS'),
    Rx.Notification.createError('ERROR!')
  ])
  // 把 notification object 变成 notification values
  .dematerialize()
  
// 输出：`NEXT VALUE: SUCCESS' 'ERROR VALUE: 'ERROR!'
const subscriptionn = source.subscribe({
  next: val => console.log(`NEXT VALUE: ${val}`),
  error: val => console.log(`ERROR VALUE: ${val}`),
})
distinctUntilChanged

signature: distinctUntilChanged(compare: function): Observable

只有当前的值与上个值不同，才 emit 。
(demo | docs)
// 仅仅输出那些与上个值不同的值
const myArrayWithDuplicatesInARow = Rx.Observable
  .from([1, 1, 2, 2, 3, 1, 2, 3])
  
const distinctSub = myArrayWithDuplicatesInARow
  .distinctUntilChanged()
  // 输出：1, 2, 3, 1, 2, 3
  .subscribe(val => console.log('DISTINCT SUB: ', val))
  
const nonDistinctSub = myArrayWithDuplicatesInARow
  // 输出：1, 1, 2, 2, 3, 1, 2, 3
  .subscribe(val => console.log('NON DISTINCT SUB:', val))
  
const sampleObject = { name: 'Test' }
const myArrayWithDuplicateObjects = Rx.Observable.from([sampleObject, sampleObject])

const nonDisinctObjects = myArrayWithDuplicateObjects
  .distinctUntilChanged()
  // 输出：`DISTINCT OBJECTS: { name: 'Test' }`
  .subscribe(val => console.log('DISTINCT OBJECTS', val))
do

signature: do(nextOrObserver: function, error: function, complete: function): Observable

显式地进行某些 action ，比如 logging
(demo | docs)
const source = Rx.Observable.of(1, 2, 3, 4, 5)

// 通过 do ，显式地打印某些值
const example = source
  .do(val => console.log(`BEFORE MAP: ${val}`)
  .map(val => val + 10)
  .do(val => console.log(`AFTER MAP: ${val}`)
  
// `do` 是不会 emit 值
const subscribe = example.subscribe(val => console.log(val))
every

signature: every(predicate: function, thisArg: any): Observable

检测是否「所有 emit 的值」都符合某个条件。
(demo | docs)
// emit 5个值
const source = Rx.Observable.of(1, 2, 3, 4, 5)
const example = source
  .every(val => val % 2 == 0)
  
// 输出：false
const subscribe = example.subscribe(val => console.log(val))


// emit 5个值
const allEvens = Rx.Observable.of(2, 4, 6, 8, 10)
const exampleTwo = allEvens
  // 是否每个值都是偶数
  .every(val => val % 2 === 0)
  
// 输出：true
const subscribe = exampleTwo.subscribe(val => console.log(val))
expand

signature: expand(project: function, concurrent: number, scheduler: Scheduler): Observable

递归地调用给定的函数。
(demo | docs)
// emit 2
const source = Rx.Observable.of(2)

const example = source
  .expand(val => {
    // 2, 3, 4, 5
    console.log(`Passed value: ${val}`)
    // 3, 4, 5, 6
    return Rx.Observable.of(1 + val)
  })
  // 只调用 5 次
  .take(5)
  
/*
* RESULT: 2
* Passed value: 2
* RESULT: 3
* Passed value: 3
* RESULT: 4
* Passed value: 4
* RESULT: 5
* Passed value: 5
* RESULT: 6
* Passes value: 6
*/

// 输出：2, 3, 4, 5, 6
const subscribe = exmaple.subscribe(val => console.log(`RESULT: ${val}`))
filter

signature: filter(select: Function, thisArg: any): Observable

只返回符合给定条件的值。
(demo | docs)
// emit (1, 2, 3, 4, 5)
const source = Rx.Observable.from([1, 2, 3, 4, 5])
// 过滤掉不是偶数的值
const example = source.filter(num => num % 2 === 0)
// 输出：Even Number: 2, Even Number: 4
const subscribe = example.subscribe(val => console.log(`Even number: ${val}`))

// emit ({ name: 'Joe', age: 31 }, { name: 'Bob', age: 25 })
const sourceTwo = Rx.Observable.from([{ name: 'Joe', age: 31 }, { name: 'Bob', age: 25 }]
// 过滤掉30岁以下的
const exampleTwo = sourceTwo.filter(person => person.age >= 300)
// 输出：Over 30: Joe
const subscribeTwo = exampleTwo.subscribe(val => console.log(`Over 30: ${val.name}`))
first

signature: first(predicate: function, select: function)

emit 第一个值，或者第一个符合给定条件的值。
(demo | docs)
const source = Rx.Observable.from([1, 2, 3, 4, 5])
// 没有传任何参数，那么就 emit 第一个值
const example = source.first()
// 输出：First Value: 1
const subscribe = example.subscribe(val => console.log(`First Value: ${val}`))

// emit 第一个符合给定条件的值
const exampleTwo = source.first(num => num === 5)
// 输出：First to pass test: 5
const subscribe = exampleTwo.subscribe(val => console.log(`First to pass test: ${val}`))

// 传递可选参数 project function
const exampleThree = source.first(
  num => num % 2 === 0,
  (result, index) => `First even: ${result} is at index: ${index}`
)
// 输出：First even: 2 is at index 1
const subscribeThree = exampleThree.subscribe(val => console.log(val))
groupBy

signature: groupBy(keySelector: Function, elementSelector: Function): Observable

按照给定的值，分组到 Observable。
(demo | docs)
const people = [{name: 'Sue', age:25},{name: 'Joe', age: 30},{name: 'Frank', age: 25}, {name: 'Sarah', age: 35}];
const source = Rx.Observable.from(people)
// 按年龄分组
const example = source
  .groupBy(person => person.age)
  // 每个分组作为数组返回
  .flatMap(group => group.reduce((acc, curr) => [...acc, curr], []))
  
/*
* 输出：
* [{age: 25, name: 'Sue'}, {age: 25, name: 'Frank'}]
* [{age: 30, name: 'joe'}]
* [{age: 35, name: 'Sarah'}]
*/

const subscribe = example.subscribe(val => console.log(val))
ignoreElements

signature: ignoreElements(): Observable

忽略所有的东西，除了 complete 和 error 。
(demo | docs)
// 每隔500ms emit 值
const source = Rx.Observable.interval(100)
// 忽略一切东西除了 complete
const example = source
  .take(5)
  .ignoreElements()
// 输出：'COMPLETE!'
const subscribe = example.subscribe(
  val => console.log(`NEXT: ${val}`),
  val => console.log(`ERROR: ${val}`),
  () => console.log(`COMPLETE`)
)

// 忽略一切东西除了 error
const error = source
  .flatMap(val => {
    if (val === 4) return Rx.Observable.throw(`ERROR AT ${val}`)
    return Rx.Observable.of(val)
  })
  .ignoreElements()
  
// 输出：ERROR: ERROR at 4
const subscribeTwo = error.subscribe(
  val => console.log(`NEXT: ${val}`),
  err => console.log(`ERROR: ${err}`),
  () => console.log('SECOND COMPLETE')
)
last

signature: last(predicate: function): Observable

emit 最后一个值，或者最后一个通过 test 的值。
(demo | docs)
const source = Rx.Observable.from([1, 2, 3, 4, 5])
// 没有传参数，emit 最后一个值
const example = source.last()
// 输出："Last value: 5"
const subscribe = example.subscribe(val => console.log(`Last value: ${val}`))

// emit 最后一个偶数
const exampleTwo = source.last(num => num % 2 === 0)
// 输出：Last to pass test: 4
const subscribeTwo = exampleTwo.subscribe(val => console.log(`Last to pass test: ${val}`))
map

signature: map(project: Function, thisArg: any): Observable

把每个值都应用到 project function ，映射为新的值。
(demo | docs)
// emit (1, 2, 3, 4, 5)
const source = Rx.Observable.of([1, 2, 3, 4, 5])
// 每个值都加10
const example = source.map(val => val + 10)
// 输出：11, 12, 13, 14, 15
const subscribe = example.subscribe(val => console.log(val))


// emit ({name: 'Joe', age: 30}, {name: 'Frank', age: 20},{name: 'Ryan', age: 50})
const sourceTwo =  Rx.Observable.from([{name: 'Joe', age: 30}, {name: 'Frank', age: 20},{name: 'Ryan', age: 50}]);
// 获取每个人的名字
const exmapleTwo = sourceTwo.map(person => person.name)
// 输出："Joe", "Frank", "Ryan"
const subscribeTwo = exampleTwo.subscribe(val => console.log(val))
mapTo

signature: mapTo(value: any): Observable

每一次都映射（map）为一个常量。
(demo | docs)
const source = Rx.Observable.interval(2000)
// 每个值都映射为一个常量
const example = source.mapTo('HELLO WORLD!')
// 输出：'HELLO WORLD!'...'HELLO WORLD!'...'HELLO WORLD!'...
const subscribe = example.subscribe(val => console.log(val))


// 每次点击 document 时都 emit
const clickSource = Rx.Observable.fromEvent(document, 'click')
// 把所有的 emission 都设为一个值
const exampleTwo = clickSource.mapTo('GOODBYE WORLD!')
// 输出：(click)'GOODBYE WORLD!'...
const subscribeTwo = exampleTwo.subscribe(val => console.log(val))
merge

signature: merge(input: Observable): Observable

把多个 Observable 压扁为一个 Observable 。
(demo | docs)
// 每隔2.5s emit 值
const first = Rx.Observable.interval(2500)
// 每隔2s emit 值
const second = Rx.Observable.interval(2000)
// 每隔1.5s emit 值
const third = Rx.Observable.interval(1500)
// 每隔1s emit 值
const fourth = Rx.Observable.interval(1000)

// 把多个 Observable 合并为一个单独的 Observable
const example = Rx.Observable.merge(
  first.mapTo('FIRST'),
  second.mapTo('SECOND'),
  third.mapTo('THIRD'),
  fourth.mapTo('FOURTH')
)

// 输出：'FOURTH', 'THIRD', 'SECOND', 'FOURTH', 'FIRST', 'THIRD', 'FOURTH'...
const subscribe = example.subscribe(val => console.log(val))

// merge 可以当作 Observable instance 的一个方法
const exampleTwo = first.merge(fourth)
// 输出：0, 1, 0, 2...
const subscribeTwo = exampleTwo.subscribe(val => console.log(val))
mergeMap

signature: mergeMap(project: function: Observable, resultSelector: function: any, concurrent: number): Observable

把 source 的值先 map 到 inner Observable ，最后压扁返回。简而言之：map => mergeAll 。
(demo | docs)
const source = Rx.Observable.of('Hello')
// map 到 inner Observable ，并且 flatten 。
const example = source.mergeMap(val => Rx.Observable.of(`${val} World!`)
// 输出：Hello World!
const subscribe = example.subscribe(val => console.log(val))

// mergeMap 也可以 emit Promise
const myPromise = val => new Promise(resolve => resolve(`${val} World from Promise!`))
// map 到 promise ，然后 emit 最后结果
const exampleTwo = source.mergeMap(val => myPromise(val))
// 输出：Hello World From Promise!
const subscribeTwo = exampleTwo.subscribe(val => console.log(val))

/*
* 你提供第2个参数，他可以接收 source 传来的值，然后在 inner Observable emit
*/
const exampleThree = source
  .mergeMap(val => myPromise(val),
    (valueFromSource, valueFromPromise) => {
      return `Source: ${valueFromSource}, Promise: ${valueFromPromise}`
    }
  )
  
// 输出：Source: Hello, Promise: Hello World From Promise!
const subscribeThree = exampleThree.subscribe(val => console.log(val))
pluck

signature: pluck(properties: ...args): Observable

挑选出嵌套的属性（nested property）。
(demo | docs)
const source = Rx.Observable.from([
  { name: 'Joe', age: 30 },
  { name: 'Sarah', age: 35 },
])

// 抓取所有的名字（name）
const example = source.pluck('name')
// 输出："Joe", "Sarch"
const subscribe = example.subscribe(val => console.log(val))

const sourceTwo = Rx.Observable.from([
  { name: 'Joe', age: 30, job: { title: 'Developer', language: 'JavaScript' }},
  // 如果没有找到 `job` 的话，就会返回 undefined
  { name: 'Sarah', age: 25 },
])
// 抓取在 `job` 里面的 `title`
const exampleTwo = sourceTwo.pluck('job', 'title')
// 输出：'Developer', undefined
const subscribeTwo = exampleTwo.subscribe(val => console.log(val))
publish

signature: publish() : ConnectableObservable

什么事情都不做，直到调用 connect ，共享 source 。
(demo | docs)
const source = Rx.Observable.interval(1000)
const example = source
  // side effect
  .do(() => console.log('Do something'))
  // 直到调用 connect() ，否则什么事情都不会发生
  .publish()
  
/*
 * source 不会 emit 值，直到调用 connect 方法
 * 输出：（5s 后）
 * 'Do Something'
 * 'Subscriber One: 0'
 * 'Subscriber Two: 0'
 * 'Do something'
 * 'Subscriber One: 1'
 * 'Subscriber Two: 1'
 */
const subscribe = example.subscribe(val => console.log(val))
const subscribeTwo = example.subscribe(val => console.log(val))

// 5秒后调用 connect 方法，这会引起 source 开始 emit 值
setTimeout(() => {
  example.connect()
}, 5000)
race

signature: race(): Observable

让第一个 Observable emit 。
(demo | docs)
// 让第一个 Observable emit
const example = Rx.Observable.race(
  Rx.Observable.interval(1500),
  Rx.Observable.interval(1000).mapTo('1s won!'),
  Rx.Observable.interval(2000),
  Rx.Observable.interval(2500)
)
// 输出：'1s won!'...'1s won!'...
const subscribe = example.subscribe(val => console.log(val))
repeat

signature: repeat(scheduler: Scheduler, count: number): Observable

指定 source 重复的次数。
(demo | docs)
// emit 'Repeat this!'
const source = Rx.Observable.of('Repeat this')
// 重复 source emit 的值3次
const example = source.repeat(3)
// 输出：Repeat this! ... Repeat this! ... Repeat this
const subscribe = example.subscribe(val => console.log(val))

const sourceTwo = Rx.Observable.interval(1000)
// 取前5个值，重复2次
const exampleTwo = source.take(5).repeat(2)
// 输出：0,1,2,3,4 ... 0,1,2,3,4
const subscribeTwo = exampleTwo.subscribe(val => console.log(val))
retry

signature: retry(number: number): Observable

当发生 error 的时候，指定 retry 的次数。
(demo | docs)
const source = Rx.Observable.interval(1000)
const example = source
  .flatMap(val => {
    // 抛出 error
    if (val > 5) {
      return Rx.Observable.throw('Errro!')
    }
    return Rx.Observable.of(val)
  })
  // retry 2 times on error
  .retry(2)
  
/*
 * 输出：
 * 0..1..2..3..4..5..
 * 0..1..2..3..4..5..
 * 0..1..2..3..4..5..
 * Error! Retried 2 times when quit!
 */
const subscribe = example
  .subscribe({
    next: val => console.log(val),
    error: val => console.log(`${va}: Retried 2 times when quit!`)
  })
retryWhen

signture: etryWhen(receives: notificationHandler, the: scheduler): Observable

在额外的逻辑的 retry 。
(demo | docs)
const source = RxObservable.interval(1000)
const example = source
  .map(val => {
    if (val > 5) {
      throw val
    }
    return val
  })
  .retryWhen(errors => errors
    .do(val => console.log(`Value ${val} was too high!`)
    .delayWhen(val => Rx.Observable.timer(val * 1000))
  )
  
/*
 * 输出：
 * 0
 * 1
 * 2
 * 3
 * 4
 * 5
 * Value 6 was too high!
 * ... wait 5s then repeat
 */
const subscribe = example.subscribe(console.log)
sample

signature: sample(sampler: Observable): Observable

当给定的 Observable emit 的时候，返回 source 最新的一个样本（sample）。
(demo | docs)
// 每隔1s emit 值
const source = Rx.Observable.interval(1000)
// 每隔2s ，返回 source 最新的值
const example = source.sample(Rx.Observble.interval(2000))
// 输出：2..4..6..8
const subscribe = example.subscribe(val => console.log(val))


const sourceTwo = Rx.Observable.zip(
  Rx.Observable.from(['Joe', 'Frank', 'Bo']),
  Rx.Observable.interval(2000)
)
// 每隔2.5s ，返回 sourceTwo 的最新值
const exampleTwo = sourceTwo.sample(Rx.Observable.interval(2500))
// 输出：['Joe', 0]...['Frank', 1]...
const subscribeTwo = exampleTwo.subscribe(console.log)
scan

signature: scan(accumulator: function, seed: any): Observable

reducer / 累加器
(demo | docs)
const testSubject = new Rx.Subject()
// 最简单的 scan 的例子，从零开始累加
const basicScan = testSubject
  .startWith(0)
  .scan((acc, curr) => acc + curr)
// 输出累加后的值
const subscribe = basiScan.subscribe(val => console.log('Accumulated total: ', val))

testSubject.next(1) // 1
testSubject.next(2) // 3
testSubject.next(3) // 6

const testSubjectTwo = new Rx.Subject()
const objectScan = testSubject.scan((acc, curr) => Object.assign({}, acc, curr), {})
const subscribe = objectScan.subscribe(val => console.log('Accumulated object: ', val))

testSubject.next({ name: 'Joe' }) // { name: 'Joe' }
testSubject.next({ age: 30 }) // { name: 'Joe', age: 30 }
testSubject.next({ favoriteLanguage: 'JavaScript' }) // { name: 'Joe', age: 30, favoriteLanguage: 'JavaScript' }
share

signature: share(): Observable

在多个 subscriber 中共享 observable 。
(demo | dcos)
const source = Rx.Observable.timer(1000)
const example = source
  .do(() => console.log('*** SIDE EFFECT ***'))
  .mapTo('*** RESULT ***')
  
/*
 * 如果不 share 的话，SIDE EFFECT 会执行两次
 * 输出：
 * *** SIDE EFFET ***
 * *** RESULT ***
 * *** SIDE EFFET ***
 * *** RESULT ***
*/
const subscribe = example.subscribe(console.log)
const subscribeTwo = example.subscribe(console.log)

// 在 subscriber 中共享 observable
const shareExample = example.share()

/*
 * 如果 share 的话，SIDE EFFECT 只会执行一次
 * 输出：
 * *** SIDE EFFECT ***
 * *** RESULT ***
 * *** RESULT ***
 */
const subscribeThree = shareExample.subscribe(console.log)
const subscribeFour = shareExample.subscribe(console.log)
single

signature: signature: single(a: Function): Observable

emit 符合条件的一个单独的值。
(demo | docs)
// emit (1,2,3,4,6)
const source = Rx.Observable.of([1,2,3,4,5])
// emit 符合条件的一个 value 
const example = source.single(val => val === 4)
// 输出：4
const subscribe = example.subscribe(console.log)
skip

signature: skip(the: Number): Observable

跳过指定数量的 emitted value 。
(demo | docs)
const source = Rx.Observable.interval(1000)
// 跳过前5个 emitted value 
const example = source.skip(5)
// 输出：5..6..7..8...
const subscribe = example.subscribe(console.log)
skipUntil

signature: skipUntil(the: Observable): Observable

跳过 source emit 的值，直到 inner Observable emit 。
(demo | docs)
const source = Rx.Observable.interval(1000)
// 跳过 source emit 的值，直到 inner Observable emit 。
const example = source.skipUntil(Rx.Observable.timer(6000))
// 输出：5...6...7...8...
const subscribe = example.subscribe(console.log)
skipwhile

signature: skipWhile(predicate: Function): Observable

跳过 source emit 的值，直到给定的条件为 false 。
(demo | docs)
const source = Rx.Observable.interval(1000)
const example = source.skipWhile(val => val < 5)
// 输出：5...6...7...8.......
const subscribe = example.subscribe(console.log)
startWith

signature: startWith(an: Values): Observable

指定第一个 emit 的值。
(demo | docs)
const source = Rx.Observable.of([1, 2, 3])
const example = source.startWith(0)
// 输出：0,1,2,3
const subscribe = example.subscribe(console.log)

const sourceTwo = Rx.Observable.of('World! ', 'Goodbye', 'World!')
const exampleTwo = sourceTwo
  .startWith('Hello')
  .scan((acc, curr) => `${acc} ${curr}`)
  
// 输出：
// Hello
// Hello Wrold!
// Hello World! Goodybe
// Hello World! Goodbye World!
const subscribe = exmapleTwo.subscribe(console.log)
switchMap

signature: switchMap(a: Observable): Observable

当 source emit 的时候，切换到 inner Observable ，并且 emit 他已经 emit 过的值。
(demo | docs)
const source = Rx.Observable.timer(0, 5000)
// 切换到 inner Observable ，emit 已经 emit 过的值
const example = source.switchMap(() => Rx.Observable.interval(500))
// 输出：0,1,2,3,4,5,6,7,8,9.....0,1,2,3,4,5,6,7,8,9
const subscribe = example.subscribe(console.log)

const sourceTwo = Rx.Observable.fromEvent(document, 'click')
// 如果下一个 click 在 3s 内发生的话，不会产生新的 msg('Hello, I made it!')
const exampleTwo = sourcwTwo.switchMap(val => Rx.Observable.interval(3000).mapTo('Hello, I made it!'))
// 输出：(click)...3s...'Hello, I made it!'...(click)...2s(click)...
const subscribeTwo = exampleTwo.subscribe(console.log)
window

signature: window(windowBoundaries: Observable): Observable

类似于 buffer ，但是返回的是 nested Observable 。
(demo | docs)
// 马上 emit ，然后每隔1s emit
const source = Rx.Observable.timer(0, 1000)
const example = source
  .window(Rx.Observable.interval(3000))
  
const count = example.scan((acc, curr) => acc + 1, 0)
  
/*
 * window 1:
 * 0
 * 1
 * 2
 * window 2:
 * 3
 * 4
 * 5
*/
const subscribe = count.subscribe(val => console.log(`Window ${val}:`))
const subscribeTwo = example.mergeAll().subscribe(val => console.log(val))
windowCount

signature: windowCount(windowSize: number, startWindowEvery: number): Observable

source emit 的值是 Observable ，emit 的间隔是指定的时间。
(demo | docs)
// 每隔1s emit
const source = Rx.Observable.interval(1000)
const example = source
  // 每隔4个值就开始新的 window
  .windowCount(4)
  .do(() => console.log('NEW WINDOW!'))
  
const subscribeTwo = example
  // window emit 的是 nested Observable
  .mergeAll()
  .sbuscribe(console.log)
/*
 * 输出：
 * NEW WINDOW!
 * 0
 * 1
 * 2
 * 3
 * NEW WINDOW!
 * 4
 * 5
 * 6
 * 7
*/
windowTime

signature: windowTime(windowTimeSpan: number, windowCreationInterval: number, scheduler: Scheduler): Observable

跟 bufferTime 一样，除了 emit 的值是 nested Observable 而不是一个 array 。
(demo | docs)
const source = Rx.Observable.timer(0, 1000)
const example = source
  // 每隔3s开始一个新的 window
  .windowTime(3000)
  .do(() => console.log('New Window!'))
  
const subscribe = example
  // window emit 的值为 nested Observable
  .mergeAll()
  
/*
 * 输出
 * New Window:
 * 0
 * 1
 * 2
 * New Window:
 * 3
 * 4
 * 5
 */
 .subscribe(console.log)
windowToggle

signature: windowToggle(openings: Observable, closingSelector: function(value): Observable): Observable

跟 bufferTime 一样，除了 emit 的值是 nested Observable 而不是一个 array 。
(demo | docs)
const source = Rx.Observable.timer(0, 1000)
// 在第5s的时候 toggle window
const toggle = Rx.Observable.interval(5000)
const example = source
  // 每隔5s 打开 window
  .windowToggle(toggle, val => Rx.Observable.interval(val * 1000))
  .do(() => console.log('New Window!'))
  
const subscribe = example
  // window emit 的值为 nested Observable
  .mergeAll()
  
/*
 * New Window!
 * New Window!
 * 10
 * New Window!
 * 15
 * 16
 * ...
 */
windowWhen

signature: windowWhen(closingSelector: function(): Observable): Observable

跟 bufferWhen 一样，除了 emit 的值是 nested Observable 而不是一个 array 。
(demo | docs)
const source = Rx.Observable.timer(0, 1000)
const example = source
  // 每隔5s就关闭wndow，并且 emit 从 source 中缓冲好的值
  .windowWhen(val => Rx.Observable.interval(5000))
  .do(() => console.log('New Window!'))
  
const subscribe = example
  .mergeAll()
  /*
   * New Window!
   * 0
   * 1
   * 2
   * 3
   * 4
   * New Window!
   * 5
   * 6
   * 7 
   * 8
   * 9
   */
    .subscribe(console.log)
withLatestFrom

signature: withLatestFrom(other: Observable, project: Function): Observable

当 source emit 的时候，同时也返回另一个 Obserable 最近 emit 的那个值。
(demo | docs)
const source = Rx.Observable.interval(5000)
const secondSource = Rx.Observable.interval(1000)
const example = source
  .withLatestFrom(secondSource)
  .map(([first, second]) => `First Source(5s): ${first} Second Source(1s): ${second}`)
  
/*
 * First Source(5s): 0 Second Source(1s): 4
 * First Source(5s): 1 Second Source(1s): 9
 * First Source(5s): 2 Second Source(1s): 14
 */
const subscribe = example.subscribe(console.log)


// withLastest 比 source 慢
const exampleTwo = secondSource
  .withLatestFrom(source)
  .map(([first, second]) => `Source(1s): ${first} Latest from(5s): ${second}`)
  
/*
 * Source(1s): 4 Latest Lastest from(5s): 0
 * Source(1s): 5 Latest Lastest from(5s): 0
 * Source(1s): 6 Latest Lastest from(5s): 0
 */
const subscribeTwo = exampleTwo.subscribe(console.log)
zip

signature: zip(observables: *): Observable

等到所有的 Observable 都 emit 之后，才作为数组返回。
(demo | docs)
const sourceOne = Rx.Observable.of('Hello')
const sourceTwo = Rx.Observable.of('World')
const sourceThree = Rx.Observable.of('Goodbye')
const sourceFour = Rx.Observable.of('World!')
// 等到所有的 Observable 都 emit 之后，才把它们作为数组 emit 出去
const example = Rx.Observable
  .zip(
    sourceOne,
    sourceTwo.delay(1000),
    sourceThree.delay(2000),
    sourceFour.delay(3000)
  )
// 输出：['Hello', 'World', 'Goodbye', 'World!']
const subscribe = example.subscribe(console.log)

// 每隔1s emit
const interval = Rx.Observable.interval(1000)
// 当一个 Observable complete 后，再也不会 emit 任何值
const exampleTwo = Rx.Observable
  .zip(
    interval,
    iterval.take(2)
  )
// 输出：[0, 0]...[1,1]
const subscribeTwo = exampleTwo.subscribe(console.log)