Использование RxJS в React-разработке для управления состоянием приложений

Автор материала, перевод которого мы сегодня публикуем, говорит, что здесь он хочет продемонстрировать процесс разработки простого React-приложения, использующего RxJS. По его словам, он не является экспертом в RxJS, так как сам занимается изучением этой библиотеки и не отказывается от помощи знающих людей. Его цель — привлечь внимание аудитории к альтернативным способам создания React-приложений, вдохновить читателя на самостоятельные исследования. Этот материал нельзя назвать введением в RxJS. Тут будет показан один из многих способов использования этой библиотеки в React-разработке.

С чего всё началось

Недавно мой клиент вдохновил меня на изучение использования RxJS для управления состоянием React-приложений. Когда я проводил аудит кода приложения этого клиента, он хотел узнать моё мнение о том, как ему развивать приложение, учитывая то, что до этого в нём использовалось исключительно локальное состояние React. Проект достиг такого уровня, когда неоправданно было полагаться исключительно на React. Сначала мы говорили об использовании в качестве более совершенных средств управления состоянием приложения Redux или MobX. Мой клиент создал прототип для каждой из этих технологий. Но этими технологиями он не ограничился, создав и прототип React-приложения, в котором используется RxJS. С этого момента наш разговор стал гораздо интереснее.

Приложение, о котором идёт речь, представляло собой торговую площадку для криптовалют. Оно имело множество виджетов, данные которых обновлялись в режиме реального времени. Разработчикам этого приложения, кроме прочих, надо было решить следующие непростые задачи:

• Управление множеством (асинхронных) запросов на получение данных.
• Обновление, в режиме реального времени, большого количества виджетов на панели управления.
• Решение проблемы связанности виджетов и данных, так как некоторые виджеты нуждались в данных не только из неких особых источников, но и из других виджетов.

В итоге основные сложности, с которыми столкнулись разработчики, не относились к самой библиотеке React, к тому же, я мог им в этой области помочь. Главная проблема заключалась в том, чтобы заставить правильно работать внутренние механизмы системы, те, что связывали данные о криптовалютах и интерфейсные элементы, созданные средствами React. Именно в этой области возможности RxJS оказались весьма кстати, и прототип, который они мне показали, выглядел весьма многообещающе.

Использование RxJS в React

Предположим, что у нас есть приложение, которое, после выполнения неких локальных действий, производит запросы к стороннему API. Оно позволяет выполнять поиск по статьям. Прежде чем выполнить запрос, нам необходимо получить текст, который используется для формирования этого запроса. В частности, мы, с использованием этого текста, формируем URL для обращения к API. Вот как выглядит код React-компонента, реализующий этот функционал

import React from 'react';

const App = ({ query, onChangeQuery }) => (
  <div>
    <h1>React with RxJS</h1>

    <input
      type="text"
      value={query}
      onChange={event => onChangeQuery(event.target.value)}
    />

    <p>{`http://hn.algolia.com/api/v1/search?query=${query}`}</p>
  </div>
);

export default App;

Этому компоненту не хватает системы управления состоянием. Состояние для свойства query нигде не хранится, функция onChangeQuery также не обновляет состояние. При обычном подходе такой компонент оснащают системой локального управления состоянием. Выглядит это так:

class App extends React.Component {
  constructor(props) {
    super(props);

    this.state = {
      query: '',
    };
  }

  onChangeQuery = query => {
    this.setState({ query });
  };

  render() {
    return (
      <div>
        <h1>React with RxJS</h1>

        <input
          type="text"
          value={this.state.query}
          onChange={event =>
            this.onChangeQuery(event.target.value)
          }
        />

        <p>{`http://hn.algolia.com/api/v1/search?query=${
          this.state.query
        }`}</p>
      </div>
    );
  }
}

export default App;

Однако это не тот подход, о котором мы будем здесь говорить. Вместо этого мы хотим наладить систему управления состоянием приложения с помощью RxJS. Посмотрим на то, как это сделать с использованием компонентов высшего порядка (Higher-Order Component, HOC).

При желании вы можете реализовать подобную логику и в своём компоненте App, но вы, скорее всего, в некий момент работы над приложением, решите оформить такой компонент в виде HOC, который подходит для повторного использования.

React и компоненты высшего порядка RxJS

Разберёмся с тем, как управлять состоянием React-приложений с помощью RxJS, применяя для этой цели компонент высшего порядка. Вместо этого можно было бы реализовать шаблон «render props». В итоге, если вы не хотите самостоятельно создавать для этой цели компонент высшего порядка, вы можете воспользоваться наблюдаемыми компонентами высшего порядка Recompose с mapPropsStream() и componentFromStream(). В этом руководстве, однако, мы будем делать всё самостоятельно.

import React from 'react';

const withObservableStream = (...) => Component => {
  return class extends React.Component {
    componentDidMount() {}

    componentWillUnmount() {}

    render() {
      return (
        <Component {...this.props} {...this.state} />
      );
    }
  };
};

const App = ({ query, onChangeQuery }) => (
  <div>
    <h1>React with RxJS</h1>

    <input
      type="text"
      value={query}
      onChange={event => onChangeQuery(event.target.value)}
    />

    <p>{`http://hn.algolia.com/api/v1/search?query=${query}`}</p>
  </div>
);

export default withObservableStream(...)(App);

Пока компонент высшего порядка RxJS никаких действий не выполняет. Он лишь передаёт своё собственное состояние и свойства входному компоненту, который планируется расширить с его помощью. Как видите, управлением состоянием React, в итоге, будет заниматься компонент высшего порядка. Однако это состояние будет получено из наблюдаемого потока. Прежде чем мы приступим к реализации HOC и к использованию его с компонентом App, мы должны установить RxJS:

npm install rxjs --save

Теперь приступим к использованию компонента высшего порядка и к реализации его логики:

import React from 'react';
import { BehaviorSubject } from 'rxjs';

...

const App = ({ query, onChangeQuery }) => (
  <div>
    <h1>React with RxJS</h1>

    <input
      type="text"
      value={query}
      onChange={event => onChangeQuery(event.target.value)}
    />

    <p>{`http://hn.algolia.com/api/v1/search?query=${query}`}</p>
  </div>
);

const query$ = new BehaviorSubject({ query: 'react' });

export default withObservableStream(
  query$,
  {
    onChangeQuery: value => query$.next({ query: value }),
  }
)(App);

Сам компонент App не меняется. Мы лишь передали два аргумента компоненту высшего порядка. Опишем их:

• Наблюдаемый объект. Аргумент query является наблюдаемым объектом, имеющим начальное значение, но, кроме того, выдающим, со временем, новые значения (так как это — BehaviorSubject). На данный наблюдаемый объект может подписаться кто угодно. Вот что говорит об объектах типа BehaviorSubject документация по RxJS: «Одним из вариантов объектов Subject является объект BehaviorSubject, использующий понятие «текущего значения». Он хранит последнее значение, переданное его подписчикам, и, когда на него подписывается новый наблюдатель, он немедленно получает это «текущее значение» от объекта BehaviorSubject. Такие объекты хорошо подходят для представления данных, новые порции которых появляются с течением времени».
• Система выдачи новых значений для наблюдаемого объекта (триггер). Функция onChangeQuery(), передаваемая через HOC компоненту App, представляет собой обычную функцию, которая передаёт следующее значение наблюдаемому объекту. Эта функция передаётся в объекте, так как может понадобиться передать компоненту высшего порядка несколько таких функций, которые выполняют некие действия с наблюдаемыми объектами.

После создания наблюдаемого объекта и подписки на него, поток для запроса должен заработать. Однако до сих пор сам компонент высшего порядка выглядит для нас как чёрный ящик. Реализуем его:

const withObservableStream = (observable, triggers) => Component => {
  return class extends React.Component {
    componentDidMount() {
      this.subscription = observable.subscribe(newState =>
        this.setState({ ...newState }),
      );
    }

    componentWillUnmount() {
      this.subscription.unsubscribe();
    }

    render() {
      return (
        <Component {...this.props} {...this.state} {...triggers} />
      );
    }
  };
};

Компонент высшего порядка получает наблюдаемый объект и объект с триггерами (возможно, этот объект, содержащий функции, можно назвать каким-то более удачным термином из лексикона RxJS), представленные в сигнатуре функции.

Триггеры лишь передаются через HOC входному компоненту. Именно поэтому компонент App напрямую получает функцию onChangeQuery(), которая непосредственно работает с наблюдаемым объектом, выполняя передачу ему новых значений.

Наблюдаемый объект использует метод жизненного цикла componentDidMount() для подписывания и метод componentDidMount() для отмены подписки. Отмена подписки нужна для предотвращения утечек памяти. В подписке наблюдаемого объекта функция лишь отправляет все входящие данные из потока в локальное хранилище состояния React с помощью команды this.setState().

Выполним небольшое изменение компонента App, которое позволит избавиться от проблемы, возникающей в том случае, если в компоненте высшего порядка не установлено начальное значение для свойства query. Если этого не сделать, то, в начале работы, свойство query окажется равным undefined. Благодаря данному изменению это свойство получает значение по умолчанию.

const App = ({ query = '', onChangeQuery }) => (
  <div>
    <h1>React with RxJS</h1>

    <input
      type="text"
      value={query}
      onChange={event => onChangeQuery(event.target.value)}
    />

    <p>{`http://hn.algolia.com/api/v1/search?query=${query}`}</p>
  </div>
);

Ещё один способ борьбы с этой проблемой выглядит как установка исходного состояния для query в компоненте высшего порядка:

const withObservableStream = (
  observable,
  triggers,
  initialState,
) => Component => {
  return class extends React.Component {
    constructor(props) {
      super(props);

      this.state = {
        ...initialState,
      };
    }

    componentDidMount() {
      this.subscription = observable.subscribe(newState =>
        this.setState({ ...newState }),
      );
    }

    componentWillUnmount() {
      this.subscription.unsubscribe();
    }

    render() {
      return (
        <Component {...this.props} {...this.state} {...triggers} />
      );
    }
  };
};

const App = ({ query, onChangeQuery }) => (
  ...
);

export default withObservableStream(
  query$,
  {
    onChangeQuery: value => query$.next({ query: value }),
  },
  {
    query: '',
  }
)(App);

Если вы сейчас испытаете это приложение, то поле ввода должно работать так, как ожидается. Компонент App получает из HOC, в виде свойств, лишь состояние query и функцию onChangeQuery для изменения состояния.

Получение и изменение состояния происходят через наблюдаемые объекты RxJS, несмотря на то, что внутри компонента высшего порядка используется внутреннее хранилище состояния React. Мне не удалось найти очевидного решения задачи по потоковой отправке данных из подписки наблюдаемых объектов напрямую в свойства расширенного компонента (App). Именно поэтому мне пришлось использовать локальное состояние React в виде промежуточного слоя, который, кроме того, удобен в том плане, что вызывает повторный рендеринг. Если вам известен другой способ достижения тех же целей — можете поделиться им в комментариях.

Комбинирование наблюдаемых объектов в React

Создадим второй поток значений, с которым, так же, как и со свойством query, можно работать в компоненте App. Позже мы будем пользоваться обоими значениями, работая с ними с помощью ещё одного наблюдаемого объекта.

const SUBJECT = {
  POPULARITY: 'search',
  DATE: 'search_by_date',
};

const App = ({
  query = '',
  subject,
  onChangeQuery,
  onSelectSubject,
}) => (
  <div>
    <h1>React with RxJS</h1>

    <input
      type="text"
      value={query}
      onChange={event => onChangeQuery(event.target.value)}
    />

    <div>
      {Object.values(SUBJECT).map(value => (
        <button
          key={value}
          onClick={() => onSelectSubject(value)}
          type="button"
        >
          {value}
        </button>
      ))}
    </div>

    <p>{`http://hn.algolia.com/api/v1/${subject}?query=${query}`}</p>
  </div>
);

Как видите, параметр subject может быть использован для уточнения запроса при построении URL, используемого для обращения к API. А именно, материалы можно искать, ориентируясь на их популярность или на дату публикации. Далее, создадим ещё один наблюдаемый объект, который может быть использован для изменения параметра subject. Этот наблюдаемый объект может быть использован для организации связи компонента App и компонента высшего порядка. В противном случае свойства, переданные компоненту App, работать не будут.

import React from 'react';
import { BehaviorSubject, combineLatest } from 'rxjs/index';

...

const query$ = new BehaviorSubject({ query: 'react' });
const subject$ = new BehaviorSubject(SUBJECT.POPULARITY);

export default withObservableStream(
  combineLatest(subject$, query$, (subject, query) => ({
    subject,
    query,
  })),
  {
    onChangeQuery: value => query$.next({ query: value }),
    onSelectSubject: subject => subject$.next(subject),
  },
)(App);

Триггер onSelectSubject() не является чем-то новым. Он, посредством кнопки, может быть использован для переключения между двумя состояниями subject. Но наблюдаемый объект, переданный компоненту высшего порядка, представляет собой нечто новое. Он использует функцию combineLatest() из RxJS для комбинирования последних выданных значений из двух (или большего количества) наблюдаемых потоков. После того, как оформлена подписка на наблюдаемый объект, при изменении любого из значений (query или subject) подписчик получит оба значения.

Дополнением к механизму, реализуемому функцией combineLatest(), является её последний аргумент. Здесь можно задать порядок возврата значений, генерируемых наблюдаемыми объектами. В нашем случае нам нужно, чтобы они были представлены в виде объекта. Это позволит, как и прежде, деструктурировать их в компоненте высшего порядка и записать в локальное состояние React. Так как необходимая структура у нас уже имеется, мы можем опустить шаг оборачивания объекта наблюдаемого объекта query.

...

const query$ = new BehaviorSubject('react');
const subject$ = new BehaviorSubject(SUBJECT.POPULARITY);

export default withObservableStream(
  combineLatest(subject$, query$, (subject, query) => ({
    subject,
    query,
  })),
  {
    onChangeQuery: value => query$.next(value),
    onSelectSubject: subject => subject$.next(subject),
  },
)(App);

Исходный объект, { query: '', subject: 'search' }, а так же все остальные объекты, выдаваемые комбинированным потоком наблюдаемых объектов, подходят для деструктурирования их в компоненте высшего порядка и для записи соответствующих значений в локальное состояние React. После обновления состояния, как и прежде, выполняется рендеринг. Когда вы запустите обновлённое приложение, у вас должна быть возможность изменять оба значения, пользуясь полем ввода и кнопкой. Изменённые значения влияют на URL, используемый для доступа к API. Даже если меняется лишь одно из этих значений, другое значение сохраняет своё последнее состояние, так как функция combineLatest() всегда комбинирует самые свежие значения, выданные из наблюдаемых потоков.

Axios и RxJS в React

Теперь в нашей системе URL для доступа к API конструируется на основе двух значений из комбинированного наблюдаемого объекта, который включает в себя два других наблюдаемых объекта. В этом разделе мы воспользуемся URL для загрузки данных из API. Возможно, вы хорошо умеете пользоваться системой загрузки данных React, но, при использовании наблюдаемых объектов RxJS, в наше приложение необходимо добавить ещё один наблюдаемый поток.

Прежде чем мы займёмся работой над следующим наблюдаемым объектом, установим axios. Это — библиотека, которую мы будем использовать для загрузки данных из потоков в программу.

npm install axios --save

Теперь представим, что у нас есть массив статей, который должен вывести компонент App. Тут, в качестве значения соответствующего параметра по умолчанию, мы используем пустой массив, поступив так же, как мы уже поступали с другими параметрами.

...

const App = ({
  query = '',
  subject,
  stories = [],
  onChangeQuery,
  onSelectSubject,
}) => (
  <div>
    ...

    <p>{`http://hn.algolia.com/api/v1/${subject}?query=${query}`}</p>

    <ul>
      {stories.map(story => (
        <li key={story.objectID}>
          <a href={story.url || story.story_url}>
            {story.title || story.story_title}
          </a>
        </li>
      ))}
    </ul>
  </div>
);

Для каждой статьи в списке предусмотрено использование запасного значения из-за того, что API, к которому мы обращаемся, неоднородно. Теперь — самое интересное — реализация нового наблюдаемого объекта, который отвечает за загрузку данных в React-приложение, которое будет их визуализировать.

import React from 'react';
import axios from 'axios';
import { BehaviorSubject, combineLatest } from 'rxjs';
import { flatMap, map } from 'rxjs/operators';

...

const query$ = new BehaviorSubject('react');
const subject$ = new BehaviorSubject(SUBJECT.POPULARITY);

const fetch$ = combineLatest(subject$, query$).pipe(
  flatMap(([subject, query]) =>
    axios(`http://hn.algolia.com/api/v1/${subject}?query=${query}`),
  ),
  map(result => result.data.hits),
);

...

Новый наблюдаемый объект — это, опять же, комбинация наблюдаемых объектов subject и query, так как, для построения URL, с помощью которого мы будем обращаться к API для загрузки данных, нам нужны оба значения. В методе pipe() наблюдаемого объекта мы можем применять так называемые «операторы RxJS» для того, чтобы выполнять со значениями некие действия. В данном случае мы выполняем мэппинг двух значений, помещаемых в запрос, который используется axios для получения результата. Мы тут используем оператор flatMap() а не map() для доступа к результату успешно разрешённого промиса а не к самому возвращённому промису. В итоге, после подписки на этот новый наблюдаемый объект, каждый раз, когда в систему, из других наблюдаемых объектов, поступает новое значение subject или query, выполняется новый запрос, а результат оказывается в функции подписки.

Теперь мы, снова, можем предоставить новый наблюдаемый объект компоненту высшего порядка. В нашем распоряжении имеется последний аргумент функции combineLatest(), это даёт возможность напрямую отобразить его на свойство с именем stories. В конце концов, это представляет собой то, как эти данные уже используются в компоненте App.

export default withObservableStream(
  combineLatest(
    subject$,
    query$,
    fetch$,
    (subject, query, stories) => ({
      subject,
      query,
      stories,
    }),
  ),
  {
    onChangeQuery: value => query$.next(value),
    onSelectSubject: subject => subject$.next(subject),
  },
)(App);

Триггера тут нет, так как наблюдаемый объект косвенно активируется двумя другими наблюдаемыми потоками. Каждый раз, когда меняется значение в поле ввода (query) или осуществляется щелчок по кнопке (subject), это воздействует на наблюдаемый объект fetch, в который попадают самые свежие значения из обоих потоков.

Однако, возможно, нам не нужно, чтобы каждый раз при изменении значения в поле ввода это воздействовало бы на наблюдаемый объект fetch. Кроме того, нам не хотелось бы, чтобы на fetch оказывалось бы воздействие в том случае, если значение представлено пустой строкой. Именно поэтому мы можем расширить наблюдаемый объект query с использованием оператора debounce, который позволяет устранить слишком частые изменения запроса. А именно, благодаря этому механизму новое событие принимается лишь по прошествии заданного времени после предыдущего события. Кроме того, мы используем здесь оператор filter, который отфильтровывает события потока в том случае, если строка query оказывается пустой.

import React from 'react';
import axios from 'axios';
import { BehaviorSubject, combineLatest, timer } from 'rxjs';
import { flatMap, map, debounce, filter } from 'rxjs/operators';

...

const queryForFetch$ = query$.pipe(
  debounce(() => timer(1000)),
  filter(query => query !== ''),
);

const fetch$ = combineLatest(subject$, queryForFetch$).pipe(
  flatMap(([subject, query]) =>
    axios(`http://hn.algolia.com/api/v1/${subject}?query=${query}`),
  ),
  map(result => result.data.hits),
);

...

Оператор debounce делает своё дело в процессе ввода данных в поле. Однако при щелчке по кнопке, воздействующей на значение subject, запрос должен быть выполнен немедленно.

Теперь первоначальные значения для query и subject, которые мы видим, когда компонент App выводится в первый раз, не являются теми же, которые получены из начальных значений наблюдаемых объектов:

const query$ = new BehaviorSubject('react');
const subject$ = new BehaviorSubject(SUBJECT.POPULARITY);

В subject записано undefined, в query — пустая строка. Происходит это из-за того, что именно эти значения мы предоставили в качестве параметров по умолчанию для деструктурирования в сигнатуре функции компонента App. Причина подобного заключается в том, что нам необходимо ждать первоначального запроса, выполняемого наблюдаемым объектом fetch. Так как я не знаю точно, как немедленно получать значения из наблюдаемых объектов query и subject в компоненте высшего порядка для того, чтобы записать их в локальное состояние, я решил опять настроить исходное состояние для  компонента высшего порядка.

const withObservableStream = (
  observable,
  triggers,
  initialState,
) => Component => {
  return class extends React.Component {
    constructor(props) {
      super(props);

      this.state = {
        ...initialState,
      };
    }

    componentDidMount() {
      this.subscription = observable.subscribe(newState =>
        this.setState({ ...newState }),
      );
    }

    componentWillUnmount() {
      this.subscription.unsubscribe();
    }

    render() {
      return (
        <Component {...this.props} {...this.state} {...triggers} />
      );
    }
  };
};

Теперь исходное состояние можно предоставить компоненту высшего порядка в качестве третьего аргумента. В дальнейшем мы можем убрать значения параметров по умолчанию для компонента App.

...

const App = ({
  query,
  subject,
  stories,
  onChangeQuery,
  onSelectSubject,
}) => (
  ...
);

export default withObservableStream(
  combineLatest(
    subject$,
    query$,
    fetch$,
    (subject, query, stories) => ({
      subject,
      query,
      stories,
    }),
  ),
  {
    onSelectSubject: subject => subject$.next(subject),
    onChangeQuery: value => query$.next(value),
  },
  {
    query: 'react',
    subject: SUBJECT.POPULARITY,
    stories: [],
  },
)(App);

Сейчас меня беспокоит то, что начальное состояние задано так же в объявлении наблюдаемых объектов query$ и subject$. Такой подход подвержен ошибкам, так как инициализация наблюдаемых объектов и начальное состояние компонента высшего порядка совместно используют одни и те же значения. Мне больше понравилось бы, если бы, вместо этого, начальные значения извлекались бы из наблюдаемых объектов в компоненте высшего порядка для установки начального состояния. Возможно, кто-нибудь из читателей этого материала сможет поделиться в комментариях советом о том, как это сделать.

Код программного проекта, которым мы здесь занимались, можно найти здесь.

Итоги

Основная цель этого материала — продемонстрировать альтернативный подход к разработке React-приложений с использованием RxJS. Надеемся, он дал вам пищу для размышлений. Иногда в Redux и MobX необходимости нет, но, возможно, в подобных ситуациях RxJS окажется как раз тем, что подойдёт для конкретного проекта.

Уважаемые читатели! Пользуетесь ли вы RxJS при разработке React-приложений?