RxJS - 연산자 4

유용한 연산자들

sequenceEqual Operator

타이밍에 관계없이, 두 스트림 발행물들의 순서와 값 동일 여부 반환

두 스트림이 끝나는 시점이 달라도 같은 값을 같은 순서로 발행하면 true 를 발행한다.

<input type="number" />

const { from, fromEvent } = rxjs
const { sequenceEqual, mergeMap, map, take } = rxjs.operators
 
const num$ = from([3, 1, 4, 7, 5, 8, 2])
 
const key$ = fromEvent(document, 'keyup').pipe(
  map(e => Number(e.code.replace('Digit', ''))),
  take(7),
  sequenceEqual(num$)
).subscribe(console.log)

distinctUntilChanged Operator

같은 값이 연속되는 것만 제외

distinct는 전체 중에서 중복된것을 제외하고 한 번씩만 발행한다.

distinctUntilChanged는 같은 값이 연속해서 발행되면 그때는 중복을 제거하고 첫번째 값만 발행 연속되지 않고 뛰엄뛰엄 같은 값이 나오면 중복 제거가 안된다.

객체 형태에도 적용가능

const { of } = rxjs
const { distinctUntilChanged } = rxjs.operators

of(1, 1, 2, 2, 2, 1, 1, 2, 3, 3, 3, 4, 4, 1).pipe(
  distinctUntilChanged(),
).subscribe(console.log)// 1 2 1 2 3 4 1 

const { from } = rxjs
const { distinctUntilChanged } = rxjs.operators

const students = [
    { name: '홍길동', sex: 'male' },
    { name: '전우치', sex: 'male' },
    { name: '아라치', sex: 'female' },
    { name: '성춘향', sex: 'female' },
    { name: '임꺽정', sex: 'male' },
]
from(students).pipe(
  distinctUntilChanged((a, b) => a.sex === b.sex),
).subscribe(console.log)

combineLatest Operator

두 스트림을 각 최신 값들끼리 결합 ( zip과 비교)

const { combineLatest, interval, fromEvent } = rxjs
const { pluck } = rxjs.operators

combineLatest(
  interval(2000),
  fromEvent(document, 'click').pipe(pluck('x'))
).subscribe(console.log)

• zip은 두 스트림의 각각 순서대로 짝을 맞춰서 이어준다.
• combineLatest는 꼭 순서가 맞지 않아도 두 스트림의 최신 값들을 이어준다.

buffer Operator

• 두 번째 스트림이 발생하기 전까지 발행된 첫 번째 스트림 값들을 배열로 묶어서 발행한다.

const { interval, fromEvent } = rxjs
const { buffer } = rxjs.operators

interval(1000).pipe(
  buffer(fromEvent(document, 'click'))
).subscribe(console.log)

bufferCount Operator

• 발행물의 갯수를 따지는 연산자
• 첫 번째 인자로 몇 개씩 묶을지를 정하고 두 번째 인자로 몇 개씩 shift할지를 정한다.
• 두 번째 인자는 생략 가능하다.
• bufferCount(10, 5)
  • 10개씩 값을 묶고 5씩 shift한다
  • [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
  • [6, 7, 8, 9, 10, 11, 12, 13, 14, 15]
  • …

const { range } = rxjs
const { bufferCount } = rxjs.operators

range(1, 100).pipe(
  bufferCount(10, 10) // = bufferCount(10) 동알
).subscribe(console.log)

// 클릭 3번중 한 번만 반응하기
const { fromEvent } = rxjs
const { bufferCount } = rxjs.operators

fromEvent(document, 'click').pipe(
  bufferCount(3)
).subscribe(_ => console.log('FIRE'))

bufferTime Operator

• 특정 시간 간격으로 끊어 주는 것

const { interval } = rxjs
const { bufferTime } = rxjs.operators

interval(200).pipe(
  bufferTime(2000)
).subscribe(console.log)

groupBy Operator

• 조건에 해당하는 값들 끼리 묶어서 (그룹 지어서) 발행한다.

const { range } = rxjs
const { groupBy, mergeMap, toArray } = rxjs.operators

range(1, 50).pipe(
  groupBy(x => x % 3),
  mergeMap(groups$ => groups$.pipe(
    toArray())
  )
).subscribe(console.log)

startWith/endWith : 맨 앞/뒤에 1~N개 요소 추가

• 인자로 들어온 값들을 맨 앞/뒤에 붙여서 발행한다.

const { of } = rxjs
const { startWith } = rxjs.operators

const obs$ = of (1, 2, 3)

obs$.pipe(startWith(0)).subscribe(console.log)
// obs$.pipe(startWith(-2, -1, 0)).subscribe(console.log)

every : 모든 발행물들이 주어진 조건에 부합하는가 여부

• 모든 값들이 조건에 만족하면 true 하나라도 만족하지 않으면 false 를 발행한다.

const { of } = rxjs
const { every } = rxjs.operators

of(1, 3, 5, 7, 9, 11, 13, 15).pipe(
  every(x => x % 2 !== 0)
).subscribe(console.log)

defaultIfEmpty : 발행물이 없을 시 기본값 발행

const { fromEvent, timer } = rxjs
const { defaultIfEmpty, pluck, takeUntil } = rxjs.operators

fromEvent(document, 'click').pipe(
  takeUntil(timer(5000)),
  pluck('x'),
  defaultIfEmpty('NO CLICK')
).subscribe(console.log)

retry : 발행 실패시 N회 재시도

• 서버와 api 통신할 때 retry 연산자의 인자로 들어간 값만큼 재요청을 시도한다.

const { range } = rxjs
const { ajax } = rxjs.ajax
const { mergeMap, pluck, retry, } = rxjs.operators

range(1, 20).pipe(
  mergeMap(keyword => ajax(
      `http://127.0.0.1:3000/people/quarter-error/${keyword}`
    ).pipe(
      pluck('response', 'first_name'),
      retry(3)
    )
  )
).subscribe(console.log)

defer : 조건에 따라 스트림 발행

• 구독하는 순간에 조건에 따른 스트림을 생성
• 💡 옵저버블이 해당 코드가 실행되는 부분시점에서 생성되기 때문에 당시의 상태에 따라 만들어질 옵저버블이 결정되도록 할 수 있습니다.

<input type="checkbox" id="check" />

const { defer, fromEvent, of } = rxjs
const { pluck } = rxjs.operators

fromEvent(document.querySelector('#check'), 'change').pipe(
  pluck('target', 'checked')
).subscribe(checked => {
  defer(_ =>
    checked ? of('CHECKED') : of('UNCHECKED')
  ).subscribe(console.log)
})

iif : 단순화된 defer: 조건에 따라 두 스트림 중 하나 발행

• false시의 스트림이 주어지지 않으면 false시 빈 스트림이 발행되고 바로 complete된다.

<input type="checkbox" id="check" />

const { iif, fromEvent, of } = rxjs
const { pluck } = rxjs.operators

fromEvent(document.querySelector('#check'), 'change').pipe(
  pluck('target', 'checked')
).subscribe(checked => {
  iif(
    _ => checked,
    of('CHECKED'),
    of('UNCHECKED')
  ).subscribe(
    console.log,
    err => console.log(err),
    _ => console.log('COMPLETE')
  )
})

empty

• 빈 스트림을 발행

const { empty } = rxjs

empty().subscribe(console.log, console.error, _ => console.log('COMPLETE'))

throwError

• 에러를 발생

const { throwError } = rxjs

throwError('ERROR').subscribe(console.log, console.error, _ => console.log('COMPLETE'))

share : 스트림을 여러 구독자들간 공유

• 스트림의 부작용(tap 등)이 한 번만 발생
• 여러 스트림이 같은 값을 발행한다. (Subject와 비슷하다)

const { interval } = rxjs
const { take, tap, takeLast, share } = rxjs.operators

const obs$ = interval(1000).pipe(
  take(20),
  tap(x => console.log(`side effect: ${x}`)),
  share()
)

obs$.subscribe(x => console.log(`subscriber 1: ${x}`))

setTimeout(_ => {
  obs$.subscribe(x => console.log(`subscriber 2: ${x}`))
}, 5000)
setTimeout(_ => {
  obs$.subscribe(x => console.log(`subscriber 3: ${x}`))
}, 10000)

shareReplay : share 된 스트림의 마지막 N개 발행물을 새 구독자에게 발행

const { interval } = rxjs
const { take, tap, takeLast, shareReplay } = rxjs.operators

const obs$ = interval(1000).pipe(
  take(20),
  tap(x => console.log(`side effect: ${x}`)),
  shareReplay(3)
)

obs$.subscribe(x => console.log(`subscriber 1: ${x}`))

setTimeout(_ => {
  obs$.subscribe(x => console.log(`subscriber 2: ${x}`))
}, 5000)
setTimeout(_ => {
  obs$.subscribe(x => console.log(`subscriber 3: ${x}`))
}, 10000)