Інтро до концепцій RxJS з «ванільним» JavaScript

В JavaScript є декілька API, які використовують колбек-функції майже з однаковою метою.

Нитки (Streams)

stream.on('data', data => {
   console.log(data)
})
stream.on('end', () => {
   console.log("Finished")
})
stream.on('error', err => {
   console.error(err)
})

Проміси (Promises)

somePromise()
  .then(data => console.log(data))
  .catch(err => console.error(err))

Слухачі подій (Event Listeners)

document.addEventListener('click', event => {
  console.log(event.clientX)
})

Орієнтовний шаблон, який ми тут спостерігаємо: існує об'єкт, всередині якого метод, що приймає функцію (тобто колбек). В усіх наведених прикладах вирішується та сама проблема, але різними способами. Тож ви змушені напружувати мозок, щоб запам'ятати особливості синтаксису кожного підходу. Тут на допомогу приходить RxJS, який об'єднує все це під загальними абстракціями.

Що таке observable? Цей шаблон — така ж абстракція, як і масиви, функції та об'єкти. Проміси можуть виконувати як resolve, так і reject, повертаючи одне значення. Натомість observable може «випускати» значення з плином часу. Ви можете використовувати потоки даних з сервера або прослуховувати події DOM.

Шаблон Observable

const observable = {
  subscribe: observer => {

  },
  pipe: operator => {

  },
}

Observables — звичайні об'єкти, що вміщують методи subscribe та pipe. Наведений код може вас заплутати. Варто розуміти, що observers — просто об'єкти, що мають в собі колбек-методи для next, error та complete. Метод subscribe використовує observer та передає йому значення. Тож observable видає певні дані, а observer – «споживає» їх.

Observer

const observer = {
  next: x => {
    console.log(x)
  },
  error: err => {
    console.log(err)
  },
  complete: () => {
    console.log("done")
  }
}

Всередині subscribe ви передаєте певний вид даних методам об'єкта Observer.

Метод subscribe

const observable = {
  subscribe: observer => {
    document.addEventListener("click", event => {
      observer.next(event.clientX)
    })
  },
  pipe: operator => {

  },
}

Тут ми додаємо слухача події кліку по всьому документу. Якщо запустити цей код та викликати observable.subscribe(observer), побачимо координати курсору у консолі. А що з приводу методу pipe? Метод pipe отримує operator, повертає функцію і викликає її з observable.

Метод pipe

const observable = {
  subscribe: observer => {
    document.addEventListener("click", event => {
      observer.next(event.clientX)
    })
  },
  pipe: operator => {
    return operator(this)
  },
}

Але навіщо нам operator? Він потрібен для перетворення даних. У масивів є оператори на зразок map. map дозволяє виконати певну функцію для кожного елементу масиву. Ви можете отримати два масиви: один — вихідний, а інший — у певний спосіб перетворений з map.

Напишемо map-функцію для observable.

map-оператор

const map = f => {
  return observable => {
    subscribe: observer => {
      observable.subscribe({
        next: x => {
          observer.next(f(x))
        },
        error: err => {
          console.error(err)
        },
        complete: () => {
          console.log("finished")
        }
      })
    },
    pipe: operator => {
      return operator(this)
    },
  }
}

У наведеному фрагменті багато всього, тому розбиратимемось поступово.

const map = f => {
  return observable => {

Тут ми передаємо функцію f та повертаємо функцію, яка очікує observable. Пам'ятаєте метод pipe?

pipe: operator => {
  return operator(this)
},

Щоб виконати operator в observable, його необхідно передати у pipe. pipe передасть викликаний observable (this) у функцію, яку повертає наш operator.

subscribe: observer => {
  observable.subscribe({

Далі ми визначаємо метод subscribe для observable — і одразу повертаємо його. Метод очікує на observer, який він отримає у майбутньому, коли викличемо .subscribe для поверненого observable (через інший operator або явно). Потім виконується observable.subscribe з об'єктом observer .

{
  next: x => {
    observer.next(f(x))
  },
  error: err => {
    console.error(err)
  },
  complete: () => {
    console.log("finished")
  }
}

У методі next можна спостерігати, як функція викликається для майбутнього observer з функцією, котру ми передали у map та аргументом х, який передали у next. Затестимо наш новий оператор map з observable!

observable
  .pipe(map(e => e.clientX))
  .pipe(map(x => x - 1000))
  .subscribe(observer)

Наприкінці обов'язково викликаємо subscribe, інакше жоден з операторів не виконається. Усе тому, що вони огорнуті у методи subscribe своїх observer. В цих методах здійснюється виклик subscribe попереднього observer у ланцюжку. Тож цей ланцюг повинен десь починатися.

Отже, простежимо, що відбувається, коли запускаємо наведений код.

1. Для observable викликається перший pipe, виконується прив'язка контексту до this.
2. У map передається e => e.clientX і повертається функція.
3. Функція викликається з початковим observable і повертає інший observable . 1. Ми називатимемо його observable2.
4. pipe викликається для observable2, а map прив'язується до this.
5. У map передається x => x - 1000 і повертається функція.
6. Функція викликається з observable2, повертає інший observable . 2. Ми називатимемо його observable3.
7. .subscribe викликається для observable3, у якості аргументу — observer.
8. .subscribe викликається для observable2, у якості аргументу — observer оператора. 9 . .subscribe викликається для початкового observable, у якості аргументу — observer оператора.
9. Відбувається клік по координаті clientX зі значенням 100.
10. Викликається observer2.next(100).
11. Викликається observer3.next(100).
12. Викликається observer.next(-900), а у консоль виводиться -900.
13. Готово!

Ми дослідили ланцюжок виконання покроково. Коли ви викликаєте subscribe, ви очікуєте певні дані, кожне посилання, у свою чергу, запитує попереднє посилання у ланцюжку, поки не отримає дані та поки не відбудеться виклик метода next в observer. Потім ці дані піднімаються вгору по ланцюжку, трансформуючись по дорозі, поки не досягнуть кінцевого observer.

А повний код буде ось таким:

const observable = {
  subscribe: observer => {
    document.addEventListener("click", event => {
      observer.next(event.clientX)
    })
  },
  pipe: operator => {
    return operator(this)
  }
}

const observer = {
  next: x => {
    console.log(x)
  },
  error: err => {
    console.log(err)
  },
  complete: () => {
    console.log("done")
  }
}

const map = f => {
  return observable => {
    subscribe: observer => {
      observable.subscribe({
        next: x => {
          observer.next(f(x))
        },
        error: err => {
          console.error(err)
        },
        complete: () => {
          console.log("finished")
        }
      })
    },
    pipe: operator => {
      return operator(this)
    },
  }
}