Реактивные объекты reactive-objects - подробное руководство с примерами

Введение

Реактивные объекты (reactive-objects) — это подход к проектированию кода, при котором объект строится вокруг реакций на события и изменения данных, а не вокруг последовательных вызовов методов. Вместо того чтобы «тянуть» данные, вы настраиваете объект так, чтобы он автоматически «реагировал» на входящие события, сигналы, запросы или изменения состояния.

Смотрите, здесь важно понять одну вещь: реактивный объект — это не просто объект, у которого есть пара callback-функций. Это объект, чье поведение описано через потоки событий и декларативные зависимости, а не через явный императивный контроль потока (if-ы, циклы, ручные вызовы по таймерам и т. д.).

В этой статье мы разберемся:

• что такое реактивный объект и из каких частей он обычно состоит;
• чем он отличается от «обычного» объекта с методами;
• как описывать состояние и события;
• как настроить подписки и обработку ошибок;
• как организовать жизненный цикл реактивных объектов;
• как интегрировать reactive-objects в реальное приложение.

Я буду опираться на общую парадигму реактивного программирования (FRP, Reactive Streams и т. д.), но объяснять так, чтобы вы могли адаптировать эти идеи под любой стек: Java/Kotlin (Project Reactor, RxJava), JavaScript/TypeScript (RxJS, Vue, MobX), C# (Reactive Extensions), Go, а также под собственные самописные реализации.

Что такое реактивный объект

Основная идея

Если упростить, реактивный объект — это объект, который:

1. Имеет явное состояние, доступное как поток значений (observable state).
2. Подписывается на один или несколько входных потоков событий.
3. Порождает выходные потоки (события, данные, сигналы).
4. Описывает свою логику декларативно: «когда приходит X, обнови состояние так, а затем опубликуй Y».

Обратите внимание: нам важно не столько то, как именно реализован поток (Observable, Flow, Channel, Signal), сколько сам принцип: данные «текут» через объект, а объект реагирует и трансформирует их.

Отличие от обычных объектов

Смотрите, давайте сравним.

Обычный объект:

• хранит состояние в полях;
• изменяется через явные вызовы методов;
• часто «тянет» данные: сам идет за ними, делает запросы, дергает API.

Реактивный объект:

• подписывается на внешние источники (стримы, события, очереди);
• пушит (push-модель) обновления состояния и события всем подписчикам;
• может не иметь ни одного «явного» публичного метода изменения состояния — только потоки.

Проще всего понять это через небольшой пример.

Базовый пример реактивного объекта (на псевдокоде)

Здесь я размещаю пример в стиле TypeScript/RxJS-подобного кода, но вы легко перенесете логику в любой язык:

// Условный интерфейс потока событий
interface Stream<T> {
  subscribe(handler: (value: T) => void): Subscription // Подписка на события
  map<R>(fn: (value: T) => R): Stream<R>               // Преобразование потока
  filter(fn: (value: T) => boolean): Stream<T>         // Фильтрация значений
  merge(other: Stream<T>): Stream<T>                   // Объединение потоков
}

// Реактивный объект - счетчик кликов по кнопке
class ClickCounter {
  // Публичный поток текущего значения счетчика
  public readonly count$: Stream<number>

  // Внутреннее состояние
  private count: number = 0

  constructor(clicks$: Stream<void>) {
    // Подписываемся на входной поток кликов
    clicks$.subscribe(() => {
      // Обновляем внутреннее состояние
      this.count++

      // Публикуем новое значение в публичный поток count$
      // Представим, что есть функция emit, которая пушит значения в Stream
      emit(this.count$, this.count)
    })
  }
}

Комментарии к примеру:

• Объект ClickCounter сам по себе не дергает никого, он реагирует на clicks$.
• Внешний мир видит только count$ — поток, из которого можно читать изменения счетчика.
• Вся логика обновления состояния сосредоточена внутри подписки на входной поток.

Давайте теперь формализуем архитектуру реактивного объекта.

Архитектура реактивного объекта

Основные элементы

У типичного reactive-object можно выделить несколько ключевых компонентов:

1. Входные потоки (inputs)
   Они представляют то, что «происходит снаружи»: пользовательские действия, сообщения из сети, таймеры, команды от других объектов.

2. Внутреннее состояние (state)
   Это модель данных, с которой работает объект. Она может быть:

   • примитивной (число, строка);
   • структурированной (объект, DTO);
   • сложной (дерево, граф, кэш).

3. Реакции (reducers / handlers / intents)
   Это функции, которые получают событие и текущее состояние, а затем:

   • вычисляют новое состояние;
   • порождают побочные эффекты (effects);
   • генерируют выходные события.

4. Выходные потоки (outputs)
   То, на что могут подписаться другие части системы:

   • состояние как поток;
   • отдельные события (ошибки, сигналы, команды).

5. Жизненный цикл (lifecycle)
   Подписки нужно правильно создавать и закрывать, чтобы:

   • не было утечек памяти;
   • не работали «мертвые» обработчики.

Схема взаимодействия

Давайте опишем поток данных словами:

• Внешний мир публикует событие в один из входных потоков.
• Реактивный объект принимает событие, прогоняет через обработчики.
• Обработчик обновляет состояние и/или порождает новые выходные события.
• Все подписчики на выходных потоках получают актуальную информацию.

Модель состояния реактивного объекта

Почему состояние — это поток

В реактивном подходе состояние почти всегда представляют как поток, а не просто как переменную в поле. Это дает:

• возможность наблюдать изменения в реальном времени;
• простую интеграцию с UI, который сам обновляется на каждое изменение;
• удобные операторы трансформации (map, distinctUntilChanged, combineLatest и т. д.).

Давайте разберемся на примере с состоянием пользователя.

// Описание состояния
interface UserState {
  id: string | null        // Текущий пользователь
  name: string | null      // Имя пользователя
  isLoading: boolean       // Флаг загрузки
  error: string | null     // Текст ошибки, если была
}

// Реактивный объект - профиль пользователя
class UserProfile {
  public readonly state$: Stream<UserState> // Поток состояния профиля

  private state: UserState = {
    id: null,
    name: null,
    isLoading: false,
    error: null,
  }

  constructor(
    private readonly loadUser$: Stream<string>,  // Входной поток - запросы на загрузку пользователя по id
    private readonly api: UserApi               // Некоторый сервис для запросов
  ) {
    // Подписываемся на запросы на загрузку
    loadUser$.subscribe((userId) => {
      // Обновляем состояние - начинаем загрузку
      this.setState({
        ...this.state,
        isLoading: true,
        error: null,
      })

      // Делаем асинхронный запрос
      this.api.getUser(userId)
        .then((user) => {
          // Успех - обновляем состояние с данными
          this.setState({
            id: user.id,
            name: user.name,
            isLoading: false,
            error: null,
          })
        })
        .catch((err) => {
          // Ошибка - сохраняем текст ошибки
          this.setState({
            ...this.state,
            isLoading: false,
            error: String(err),
          })
        })
    })
  }

  private setState(next: UserState) {
    this.state = next
    // Публикуем новое состояние в поток
    emit(this.state$, this.state)
  }
}

Обратите внимание:

• Состояние централизовано и неизменно из точки зрения внешнего мира — вы не мутируете его напрямую, а получаете обновления через state$.
• Вся логика обновления спрятана в приватном методе setState.

Входные потоки и события

Типы входных событий

Реактивный объект может принимать разные типы событий:

• команды (command) — «сделай что-то»;
• запросы (request) — «верни результат»;
• уведомления (event) — «что-то произошло»;
• интенты пользователя (user intent) — «пользователь нажал кнопку» и т. п.

Часто все эти типы объединяют в один «доменный» тип события.

Давайте посмотрим, как это выглядит в коде.

// Возможные события для списка задач
type TodoEvent =
  | { type: 'add'; text: string }         // Добавить задачу
  | { type: 'toggle'; id: string }        // Переключить статус задачи
  | { type: 'remove'; id: string }        // Удалить задачу
  | { type: 'load' }                      // Загрузить задачи с сервера
  | { type: 'loaded'; items: TodoItem[] } // Успешная загрузка
  | { type: 'error'; message: string }    // Ошибка

class TodoListReactive {
  public readonly state$: Stream<TodoState>

  private state: TodoState = {
    items: [],
    isLoading: false,
    error: null,
  }

  constructor(
    events$: Stream<TodoEvent>,         // Один объединенный поток входных событий
    private readonly api: TodoApi
  ) {
    events$.subscribe((event) => {
      this.handleEvent(event)
    })
  }

  private handleEvent(event: TodoEvent) {
    // Разбираем тип события
    switch (event.type) {
      case 'add':
        this.addItem(event.text)
        break
      case 'toggle':
        this.toggleItem(event.id)
        break
      case 'remove':
        this.removeItem(event.id)
        break
      case 'load':
        this.loadItems()
        break
      case 'loaded':
        this.setLoaded(event.items)
        break
      case 'error':
        this.setError(event.message)
        break
    }
  }

  // Далее идут приватные методы addItem, toggleItem и т.д.
}

Как видите, этот код выполняет простую, но важную роль: он превращает абстрактный поток событий в конкретные изменения состояния и операций.

Обработчики и редьюсеры в реактивных объектах

Императивный vs декларативный стиль

Есть два основных способа описать реакцию на событие:

1. Императивный (как в примере выше) — мы явно пишем switch и меняем состояние.
2. Декларативный / реактивный — мы описываем поток преобразований:

• «для событий типа X сделай Y»;
• «когда приходит событие Z, аналогично инициируй побочный эффект W».

Смотрите, как это можно описать более реактивно.

// Представим, что у нас есть исходный поток событий events$
const addTodo$ = events$.filter(e => e.type === 'add')               // Фильтруем только события добавления
const toggleTodo$ = events$.filter(e => e.type === 'toggle')         // Фильтруем переключения
const removeTodo$ = events$.filter(e => e.type === 'remove')         // Фильтруем удаления

// Для каждого типа события определяем поток изменений состояния
const addStateUpdates$ = addTodo$.map(e => (state: TodoState) => ({
  ...state,
  items: [
    ...state.items,
    {
      id: generateId(),   // Создаем новый ID
      text: e.text,
      done: false,
    }
  ]
}))

const toggleStateUpdates$ = toggleTodo$.map(e => (state: TodoState) => ({
  ...state,
  items: state.items.map(item =>
    item.id === e.id
      ? { ...item, done: !item.done }  // Переключаем флаг done
      : item
  )
}))

const removeStateUpdates$ = removeTodo$.map(e => (state: TodoState) => ({
  ...state,
  items: state.items.filter(item => item.id !== e.id) // Убираем задачу
}))

// Объединяем все потоки обновлений состояния
const stateUpdates$ = merge(addStateUpdates$, toggleStateUpdates$, removeStateUpdates$)

Здесь каждый поток событий addTodo$, toggleTodo$, removeTodo$ превращается в поток функций-редьюсеров (state) => nextState. А дальше мы просто сворачиваем (scan в Rx-стиле) эти редьюсеры в одно итоговое состояние.

Конструирование реактивного объекта через scan

Теперь давайте построим весь реактивный объект в более функциональном стиле. Это полезно, когда вы хотите минимизировать количество побочных эффектов и сделать состояние максимально предсказуемым.

// Начальное состояние
const initialState: TodoState = {
  items: [],
  isLoading: false,
  error: null,
}

// Поток редьюсеров - функций (state) => nextState
const reducers$: Stream<(state: TodoState) => TodoState> = stateUpdates$ // Из примера выше
  .merge(loadReducers$)       // Потоки обновлений для загрузки
  .merge(errorReducers$)      // Потоки обновлений при ошибках
  // ... при необходимости добавляем другие

// Поток состояний как результат применения редьюсеров по очереди
const state$: Stream<TodoState> = reducers$.scan(
  (state, reducer) => reducer(state),  // Применяем редьюсер к текущему состоянию
  initialState                         // Начальное состояние
)

Комментарии:

• Оператор scan (или аналогичный в вашей библиотеке) — это «реактивный reduce».
• Мы фактически описали реактивный объект как функцию от потока событий к потоку состояний.
• Это крайне удобно для тестирования: вы можете подавать события на вход и проверять последовательность состояний на выходе.

Управление побочными эффектами

Что такое эффект в контексте reactive-objects

Эффект (side effect) — это любое действие, которое:

• выходит за пределы чистой функции;
• не может быть воспроизведено только на основе входных аргументов (сетевой запрос, логирование, запись в БД).

В реактивных объектах эффекты нужно явно отделять от чистых преобразований состояния. Это важно, чтобы:

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

Один из популярных подходов:

1. Редьюсер, получая событие, возвращает:
   • новое состояние;
   • описание эффекта (например, «вызвать api.loadTodos»).
2. Отдельный обработчик эффектов превращает описание эффекта в реальный вызов.

Вот пример структуры:

// Тип описания эффекта
type Effect =
  | { type: 'none' }                        // Нет эффекта
  | { type: 'loadTodos' }                  // Загрузить список задач
  | { type: 'log'; message: string }       // Залогировать сообщение

// Результат работы редьюсера - новое состояние + эффект
interface ReduceResult<S> {
  state: S
  effect: Effect
}

// Чистый редьюсер
function todoReducer(
  state: TodoState,
  event: TodoEvent
): ReduceResult<TodoState> {
  switch (event.type) {
    case 'load':
      return {
        state: {
          ...state,
          isLoading: true,
          error: null,
        },
        effect: { type: 'loadTodos' }, // Запрашиваем загрузку
      }

    case 'loaded':
      return {
        state: {
          ...state,
          items: event.items,
          isLoading: false,
          error: null,
        },
        effect: { type: 'none' },
      }

    case 'error':
      return {
        state: {
          ...state,
          isLoading: false,
          error: event.message,
        },
        effect: {
          type: 'log',
          message: `Todo error: ${event.message}`, // Фиксируем ошибку
        },
      }

    default:
      // Для всех остальных событий изменяем только состояние без эффектов
      return {
        state: state,
        effect: { type: 'none' },
      }
  }
}

Теперь покажу вам, как это реализовано вместе с выполнением эффектов.

class TodoReactive {
  public readonly state$: Stream<TodoState>

  private state: TodoState = initialState

  constructor(
    private readonly events$: Stream<TodoEvent>, // Входной поток событий
    private readonly api: TodoApi,               // Сервис с методами загрузки
    private readonly logger: Logger              // Сервис логирования
  ) {
    this.state$ = createStateStream() // Создаем поток состояний (детали зависят от реализации)

    // Подписываемся на события
    events$.subscribe((event) => {
      // Применяем редьюсер
      const { state: nextState, effect } = todoReducer(this.state, event)

      // Обновляем локальное состояние
      this.state = nextState
      emit(this.state$, this.state) // Публикуем новое состояние

      // Выполняем эффект
      this.runEffect(effect)
    })
  }

  private runEffect(effect: Effect) {
    switch (effect.type) {
      case 'none':
        return
      case 'loadTodos':
        // Делаем асинхронный запрос
        this.api.getTodos()
          .then((items) => {
            // Публикуем событие успешной загрузки
            emit(this.events$, { type: 'loaded', items })
          })
          .catch((err) => {
            // Публикуем событие ошибки
            emit(this.events$, {
              type: 'error',
              message: String(err),
            })
          })
        break
      case 'log':
        // Логируем сообщение
        this.logger.info(effect.message)
        break
    }
  }
}

Обратите внимание, как этот фрагмент кода решает задачу разделения ответственности:

• todoReducer — чистая функция, которая не знает о сети и логере.
• runEffect — место, где сосредоточены все реальные побочные действия.

Жизненный цикл реактивного объекта

Подписка и отписка

Реактивный объект, как правило:

• подписывается на внешние потоки;
• порождает выходные потоки, на которые подписываются другие.

Если вы не позаботитесь о корректной отписке:

• обработчики будут продолжать работать, даже когда объект логически «умер»;
• ссылки на объект останутся в замыканиях и коллекциях подписок, что приведет к утечкам памяти.

Давайте рассмотрим типичный паттерн управления жизненным циклом.

class ReactiveObject {
  private subscriptions: Subscription[] = []   // Храним все подписки здесь

  constructor(input1$: Stream<Event1>, input2$: Stream<Event2>) {
    // Создаем подписку на первый поток
    const sub1 = input1$.subscribe((e) => this.handleEvent1(e))
    this.subscriptions.push(sub1)

    // Создаем подписку на второй поток
    const sub2 = input2$.subscribe((e) => this.handleEvent2(e))
    this.subscriptions.push(sub2)
  }

  // Метод для освобождения ресурсов
  dispose() {
    // Отписываемся от всех потоков
    for (const sub of this.subscriptions) {
      sub.unsubscribe() // Завершаем подписку
    }
    this.subscriptions = []
  }

  private handleEvent1(e: Event1) {
    // Обработка первого типа событий
  }

  private handleEvent2(e: Event2) {
    // Обработка второго типа событий
  }
}

Здесь важно:

• В UI-фреймворках (Angular, React, Vue) метод dispose обычно вызывается автоматически при размонтировании компонента.
• В backend-приложениях вы можете вызывать его при завершении обработки запроса или закрытии соединения.

Композиция реактивных объектов

Объединение нескольких реактивных объектов

В реальных системах вы редко ограничиваетесь одним reactive-object. Обычно их много, и они взаимодействуют друг с другом через потоки.

Есть два популярных подхода к композиции:

1. Через события (event bus / message bus)
   Объекты подписываются на «шину событий» и публикуют в нее собственные события.
2. Через зависимости (внедрение потоков)
   Один объект получает на вход потоки, которые формирует другой объект.

Давайте посмотрим, как можно связать два реактивных объекта: поиск и список результатов.

// Реактивный объект - поле поиска
class SearchBox {
  public readonly query$: Stream<string> // Поток введенного текста

  constructor(inputChanges$: Stream<string>) {
    // Здесь мы просто пробрасываем входной поток наружу,
    // но можем добавлять debounce, trim и т.п.
    this.query$ = inputChanges$
      .map(text => text.trim())        // Убираем лишние пробелы
      .filter(text => text.length > 0) // Пропускаем только непустые строки
  }
}

// Реактивный объект - поиск по API
class SearchResults {
  public readonly results$: Stream<SearchResult[]>   // Поток результатов
  public readonly isLoading$: Stream<boolean>        // Поток флага загрузки

  constructor(query$: Stream<string>, api: SearchApi) {
    // Когда приходит новый запрос - загружаем результаты
    const loading$ = new Subject<boolean>()   // Условный поток булевых значений
    const results$ = new Subject<SearchResult[]>()

    query$.subscribe((q) => {
      emit(loading$, true) // Ставим флаг загрузки

      api.search(q)
        .then((items) => {
          emit(results$, items)   // Публикуем результаты
        })
        .finally(() => {
          emit(loading$, false)   // Снимаем флаг загрузки
        })
    })

    this.results$ = results$
    this.isLoading$ = loading$
  }
}

// Композиция
const searchBox = new SearchBox(userInput$)                      // userInput$ - поток ввода пользователя
const searchResults = new SearchResults(searchBox.query$, api)   // Передаем поток запросов в SearchResults

Теперь вы увидите, как это выглядит в использовании:

• UI подписывается на searchBox.query$, чтобы подсветить текущий запрос.
• Другой компонент подписывается на searchResults.results$, чтобы рендерить список.
• Все элементы связаны потоками, а не прямыми вызовами методов.

Ошибки и ретраи в реактивных объектах

Обработка ошибок

В реальности любые реактивные объекты работают с ненадежной средой:

• сеть может падать;
• пользователь может ввести неверные данные;
• внешние сервисы могут отвечать некорректно.

В реактивном подходе типичные варианты:

1. Представлять ошибки как обычные события (TodoEvent с типом error).
2. Или использовать механизмы ошибок самих потоков (например, onError в Rx).

Чаще всего безопаснее оборачивать ошибки в доменные события, чтобы:

• объект не «падал» полностью при первой ошибке;
• можно было отображать сообщения пользователю.

Мы уже видели пример с событием error, давайте дополним его логикой ретрая.

Повторные попытки (retry)

Допустим, вы хотите, чтобы при неудачной загрузке:

• пользователь мог нажать «Повторить»;
• или система автоматически сделала N повторных попыток.

Смотрите, как можно реализовать ручной retry через событие.

type TodoEvent =
  | { type: 'load' }          // Загрузить задачи
  | { type: 'retry' }         // Повторить загрузку
  | { type: 'loaded'; items: TodoItem[] }
  | { type: 'error'; message: string }

// Фрагмент редьюсера
function todoReducer(
  state: TodoState,
  event: TodoEvent
): ReduceResult<TodoState> {
  switch (event.type) {
    case 'load':
    case 'retry':
      return {
        state: {
          ...state,
          isLoading: true,
          error: null,
        },
        effect: { type: 'loadTodos' }, // В обоих случаях загружаем
      }

    case 'error':
      return {
        state: {
          ...state,
          isLoading: false,
          error: event.message,
        },
        effect: { type: 'none' },
      }

    default:
      return { state, effect: { type: 'none' } }
  }
}

Далее в UI вы можете:

• показывать кнопку «Повторить» только если есть error;
• при нажатии отправлять событие { type: 'retry' } во входной поток объекта.

Реактивные объекты и UI

Связь с интерфейсом

Реактивные объекты идеально ложатся на архитектуры типа MVVM, MVI, Redux-подобные модели:

• состояние объекта (state$) можно напрямую привязать к UI;
• события пользователя (clicks, inputs) превращать в поток входных событий.

Смотрите, на примере React-подобного подхода:

// Псевдо-React хук для подписки на поток состояний
function useStream<T>(stream: Stream<T>, initial: T): T {
  const [value, setValue] = useState(initial)

  useEffect(() => {
    // Подписываемся на поток
    const sub = stream.subscribe((v) => setValue(v)) // Обновляем состояние компонента

    // Отписываемся при размонтировании
    return () => sub.unsubscribe()
  }, [stream])

  return value
}

// Использование в компоненте
function TodoListView({ todoReactive }: { todoReactive: TodoReactive }) {
  // Подписываемся на поток состояния
  const state = useStream(todoReactive.state$, initialState)

  // Отображаем UI на основе состояния
  // Здесь только комментарий вместо JSX,
  // чтобы не отвлекать вас деталями разметки
  // // Рендер списка задач, спиннера и ошибок...
}

Важная мысль: UI-компонент становится тонким «проектором» состояния в разметку, а вся логика реакций сосредоточена в реактивном объекте.

Практические рекомендации по дизайну reactive-objects

Когда использовать реактивные объекты

Реактивный подход особенно полезен, когда:

• у вас много асинхронности и событий (UI, WebSocket, интеграции);
• состояния зависят друг от друга и пересчитываются;
• нужны простой дебаунс, буферизация, throttling, merge/zip нескольких источников.

Если код сводится к простым последовательным операциям, реактивный объект может быть избыточен.

Правила, которые помогают избежать хаоса

1. Один ответственный за состояние
   Для каждой доменной области (профиль пользователя, список задач, корзина) старайтесь иметь один главный реактивный объект, управляющий состоянием.

2. События — неизменяемые
   Не изменяйте входные события. Считайте их «фактами», которые уже произошли.

3. Разделяйте состояние и эффекты
   Как мы делали через ReduceResult и Effect. Это сильно упрощает тестирование.

4. Явно описывайте типы событий
   Не плодите «магические» строки, старайтесь иметь четкие типы, enum-ы, интерфейсы.

5. Документируйте входы и выходы
   Для каждого reactive-object полезно явно указать:

   • какие потоки он принимает;
   • какие потоки он отдает;
   • какие типы событий ожидает и генерирует.

Минимальный каркас своей реализации reactive-objects

Если вы не используете готовую библиотеку, покажу вам, как можно реализовать простой каркас потока и реактивного объекта.

Простейший Stream и Subject

// Тип подписчика
type Handler<T> = (value: T) => void

class SimpleStream<T> {
  private handlers: Handler<T>[] = [] // Список обработчиков

  subscribe(handler: Handler<T>): { unsubscribe: () => void } {
    this.handlers.push(handler)

    // Возвращаем объект с методом отписки
    return {
      unsubscribe: () => {
        this.handlers = this.handlers.filter(h => h !== handler)
      }
    }
  }

  protected emit(value: T) {
    // Вызываем всех подписчиков
    for (const handler of this.handlers) {
      handler(value)
    }
  }

  map<R>(fn: (value: T) => R): SimpleStream<R> {
    const result = new SimpleStream<R>()

    this.subscribe((value) => {
      // Преобразуем значение и отправляем в новый поток
      result.emit(fn(value))
    })

    return result
  }

  // Для краткости другие операторы (filter, merge) опускаем
}

class Subject<T> extends SimpleStream<T> {
  // Публичный метод для отправки значений
  next(value: T) {
    this.emit(value)
  }
}

Теперь давайте построим на этом реактивный объект.

// Реактивный объект на основе нашего примитивного потока
class Counter {
  public readonly value$: SimpleStream<number> // Поток значений счетчика

  private value: number = 0
  private valueSubject: Subject<number>

  constructor(increment$: SimpleStream<void>, decrement$: SimpleStream<void>) {
    this.valueSubject = new Subject<number>()
    this.value$ = this.valueSubject

    // Подписка на поток инкремента
    increment$.subscribe(() => {
      this.value++
      this.valueSubject.next(this.value) // Публикуем новое значение
    })

    // Подписка на поток декремента
    decrement$.subscribe(() => {
      this.value--
      this.valueSubject.next(this.value) // Публикуем новое значение
    })
  }
}

Теперь давайте посмотрим, что происходит в следующем примере использования:

// Создаем субъекты для событий
const inc$ = new Subject<void>()
const dec$ = new Subject<void>()

// Создаем реактивный объект
const counter = new Counter(inc$, dec$)

// Подписываемся на изменения
const sub = counter.value$.subscribe((v) => {
  console.log('Counter value =', v) // Логируем новое значение
})

// Генерируем события
inc$.next()
inc$.next()
dec$.next()

// Позже можно отписаться
sub.unsubscribe()

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

Заключение

Реактивные объекты (reactive-objects) — это способ организовать код вокруг потоков событий и состояний. Вместо того чтобы вручную управлять последовательностью вызовов и состоянием, вы описываете:

• какие события может получать объект;
• как они меняют состояние;
• какие побочные эффекты и новые события порождаются.

Ключевые моменты, которые стоит вынести:

• состояние реактивного объекта удобно представлять как поток;
• входные события лучше типизировать и обрабатывать через редьюсеры;
• побочные эффекты полезно отделять от чистой логики;
• жизненный цикл подписок нужно контролировать явно;
• композиция reactive-objects через потоки позволяет строить модульные и хорошо тестируемые системы.

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

Частозадаваемые технические вопросы по теме и ответы

Как протестировать реактивный объект без реальных сетевых запросов

1. Вынесите вызовы сети в отдельный слой (интерфейс Api).
2. В тестах используйте фейковую реализацию этого интерфейса:
   • заранее определите, какие данные она вернет;
   • либо дайте ей возможность управлять ответами вручную.
3. Вместо реальных потоков используйте управляемые тестовые стримы:
   • вручную отправляйте события (например, subject.next(...));
   • считывайте последовательность состояний из state$ и сравнивайте с ожидаемой.

Как избежать «залипания» подписок при динамическом создании объектов

1. Для каждого реактивного объекта держите список подписок.
2. Реализуйте метод dispose, который:
   • отписывается от всех потоков;
   • очищает ссылки на обработчики.
3. При удалении объекта из структуры (UI-компонент, сессия пользователя) обязательно вызывайте dispose.

Как связать несколько реактивных объектов, чтобы не было циклических зависимостей

1. Вводите «шину событий» как отдельный объект, не зависящий ни от одного из reactive-objects.
2. Каждый объект:
   • подписывается на нужные типы событий из шины;
   • публикует свои события обратно в шину.
3. Благодаря этому объекты зависят только от шины, а не друг от друга напрямую.

Как ограничить частоту событий (например, ввод в поле поиска)

1. Между источником событий (ввод пользователя) и реактивным объектом используйте оператор debounce или throttle выбранной библиотеки.
2. Оборачивайте «сырой» поток событий в адаптер, который:
   • принимает частые события;
   • наружу отдает сглаженный поток с подходящей частотой.

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

1. Внутри реактивного объекта используйте «горячий» поток с возможностью записи (Subject, BehaviorSubject и т. п.).
2. Наружу отдавайте только интерфейс «только для чтения»:
   • тип Observable/Stream без методов next/emit;
   • либо обертку, скрывающую методы записи.
3. Все изменения состояния проводите только внутри объекта через его внутренние обработчики событий.