Изучай observable, создавая observable

Эта статья — перевод оригинальной статьи Ben Lesh “Learning Observable By Building Observable”. Также я веду телеграм канал “Frontend по-флотски”, где рассказываю про интересные вещи из мира разработки интерфейсов ​

Вступление

Это повторение старой статьи, написанной мною в 2016 году, и доклада, с которым я выступал несколько раз. Я хочу немного модернизировать контент и, надеюсь, упростить его. Цель — помочь людям понять, что такое observable. Не только observable из RxJS, но и любой observable (да, их больше одного) как тип.

Observables — просто функции

Чтобы понять это, я хочу разместить рядом два примера. Один является observable из RxJS, а другой — просто функцией. Оба этих примера имеют одинаковый результат:

RxJS Observable

Это пример простого observable, созданного с помощью RxJS, который выдает три числа и завершается. Если вы уже знакомы с RxJS, это эквивалент (1, 2, 3).

import { Observable } from 'rxjs';
 
const source = new Observable<number>((subscriber) => {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.next(3);
  subscriber.complete();
});
 
// Usage
console.log('start');
source.subscribe({
  next: console.log,
  complete: () => console.log('done'),
});
console.log('stop');

Функция “observable”

/**
 * A simple object with a `next` and `complete` callback on it.
 */
interface Observer<T> {
  next: (value: T) => void;
  complete: () => void;
}
 
/**
 * A function that takes a simple object with callbacks
 * and does something them.
 */
const source = (subscriber: Observer<number>) => {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.next(3);
  subscriber.complete();
};
 
// Usage
console.log('start');
source({
  next: console.log,
  complete: () => console.log('done'),
});
console.log('stop');

Вывод (обоих!)

"start"
1
2
3
"done"
"stop"

Я хочу, чтобы вы обратили внимание на сходство. В обоих случаях вы передаете объект с помощью метода next и complete. В обоих случаях вы вызываете next и complete в теле функции. В обоих случаях тело функции не выполняется до тех пор, пока вы не вызовете source.subscribe() или просто вызовете функцию напрямую как source() в другом примере. Это потому, что observables — это просто специализированные функции.

Почему бы просто не использовать функции? Что «особенного» в observable?

Отлично подмечено. Если бы вы были очень осторожны, вы, вероятно, могли бы использовать обычные функции для некоторых из этих случаев. Но проблема в том, что версия с функциями не совсем «безопасна». На основе всего лишь одного примера, если семантика observable такова, что next никогда не должен вызываться после завершения, нам понадобятся некоторые гарантии для этого.

const source = function (subscriber: Observer<number>) {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.next(3);
  subscriber.complete();
  subscriber.next(4); // Упс, этого не должно быть
};

В примере выше будет написано «done», а затем сразу 4. Такого быть вообще не должно! Поэтому мы хотим предоставить способ гарантировать, что следующий метод нашего подписчика не будет вызван после завершения. Это можно сделать, заключив нашу функцию в класс.

/**
 * A class used to wrap a user-provided Observer. Since the
 * observer is just a plain objects with a couple of callbacks on it,
 * this type will wrap that to ensure `next` does nothing if called after
 * `complete` has been called, and that nothing happens if `complete`
 * is called more than once.
 */
class SafeSubscriber<T> {
  closed = false;
 
  constructor(private destination: Observer<T>) {}
 
  next(value: T) {
    // Check to see if this is "closed" before nexting.
    if (!this.closed) {
      this.destination.next(value);
    }
  }
 
  complete() {
    // Make sure we're not completing an already "closed" subscriber.
    if (!this.closed) {
      // We're closed now.
      this.closed = true;
      this.destination.complete();
    }
  }
}
 
/**
 * A class to wrap our function, to ensure that when the function is
 * called with an observer, that observer is wrapped with a SafeSubscriber
 */
class Observable<T> {
  constructor(private _wrappedFunc: (subscriber: Observer<T>) => void) {}
 
  subscribe(observer: Observer<T>): void {
    // We can wrap our observer in a "safe subscriber" that
    // does the work of making sure it's not closed.
    const subscriber = new SafeSubscriber(observer);
    this._wrappedFunc(subscriber);
  }
}
 
// Usage
// Now 4 won't be nexted after we complete.
const source = new Observable((subscriber) => {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.next(3);
  subscriber.complete();
  subscriber.next(4); // this does nothing.
});

Обработка Partial Observers

Другой возможный сценарий — «partial» observer. Другими словами, observable, у которого есть только метод next или complete (или, возможно, метод ошибки, но мы вернемся к этому позже). Теперь мы можем легко разобраться с этим сценарием имея наш тип observable, указанный выше, потому что мы можем реализовать это в нашем SafeSubscriber:

class SafeSubscriber<T> {
  closed = false;
 
  constructor(private destination: Partial<Observer<T>>) {}
 
  next(value: T) {
    if (!this.closed) {
      this.destination.next?.(value); // Note the ?. check here.
    }
  }
 
  complete() {
    if (!this.closed) {
      this.closed = true;
      this.destination.complete?.(); // And here.
    }
  }
}

Уведомления об ошибке

Уведомить нашего подписчика об ошибке так же просто, как добавить дополнительный обработчик к нашим Observer и SafeSubscriber. Семантика очень похожа на complete выше. И ошибка, и complete считаются прекращением наблюдения.

Observer просто немного изменится, чтобы появился обработчик ошибок:

interface Observer<T> {
  next: (value: T) => void;
  complete: () => void;
  error: (err: any) => void;
}

После этого мы можем добавить в наш SafeSubscriber error метод:

class SafeSubscriber<T> {
  closed = false;
 
  constructor(private destination: Partial<Observer<T>>) {}
 
  next(value: T) {
    if (!this.closed) {
      this.destination.next?.(value);
    }
  }
 
  complete() {
    if (!this.closed) {
      this.closed = true;
      this.destination.complete?.();
    }
  }
 
  error(err: any) {
    if (!this.closed) {
      this.closed = true;
      this.destination.error?.(err);
    }
  }
}

Источники данных и кейс для отмены подписки

Основной вариант использования observable — обертывание асинхронного источника данных, например WebSocket.

Чтобы сделать что-то подобное, вы можете использовать наш самодельный observable созданный выше, например:

const helloSocket = new Observable<string>((subscriber) => {
  // Open a socket.
  const socket = new WebSocket('wss://echo.websocket.org');
 
  socket.onopen = () => {
    // Once it's open, send some text.
    socket.send('Hello, World!');
  };
 
  socket.onmessage = (e) => {
    // When it echoes the text back (in the case of this particular server)
    // notify the consumer.
    subscriber.next(e.data);
  };
 
  socket.onclose = (e) => {
    // Oh! we closed!
    if (e.wasClean) {
      // ...because the server said it was done.
      subscriber.complete();
    } else {
      // ...because of something bad. Maybe we lost network or something.
      subscriber.error(new Error('Socket closed dirty!'));
    }
  };
});
 
// Start the websocket and log the echoes
helloSocket.subscribe({
  next: console.log,
  complete: () => console.log('server closed'),
  error: console.error,
});

Но теперь у нас появилась проблема. Пользователь, который подписывается на наш observable, не имеет возможности отменить его и закрыть сокет. Нам нужен способ прекратить работу. Если бы мы просто использовали функцию, мы могли бы вернуть функцию, содержащую нашу логику прекращения подписки.

const source = (subscriber: Observer<string>) => {
  const socket = new WebSocket('wss://echo.websocket.org');
 
  socket.onopen = () => {
    socket.send('Hello, World!');
  };
 
  socket.onmessage = (e) => {
    subscriber.next(e.data);
  };
 
  socket.onclose = (e) => {
    if (e.wasClean) {
      subscriber.complete();
    } else {
      subscriber.error(new Error('Socket closed dirty!'));
    }
  };
 
  return () => {
    if (socket.readyState <= WebSocket.OPEN) {
      socket.close();
    }
  };
};
 
const teardown = source({
  next: console.log,
  complete: () => console.log('done'),
  error: console.error,
});
 
// Decide you really don't want to keep the socket open.
teardown();

Еще одна проблема, которая возникает не обязательно связанная с WebSocket, но и с другими ситуациями, — это ситуации, когда автор observable решает, что он столкнулся с ошибкой или состоянием завершения, и он хочет уведомить пользователя, а затем удалить его. Было бы неплохо иметь единое место для этого, чтобы при вызове subscriber.error или subscriber.complete удаление выполнялось как можно быстрее.

Мы можем добиться всего этого с помощью некоторых изменений SafeSubscriber и добавления типа Subscription.

/**
 * Our subscription type. This is to manage teardowns.
 */
class Subscription {
  private teardowns = new Set<() => void>();
 
  add(teardown: () => void) {
    this.teardowns.add(teardown);
  }
 
  unsubscribe() {
    for (const teardown of this.teardowns) {
      teardown();
    }
    this.teardowns.clear();
  }
}
 
class SafeSubscriber<T> {
  closed = false;
 
  constructor(
    private destination: Partial<Observer<T>>,
    private subscription: Subscription,
  ) {
    // Make sure that if the subscription is unsubscribed,
    // we don't let any more notifications through this subscriber.
    subscription.add(() => (this.closed = true));
  }
 
  next(value: T) {
    if (!this.closed) {
      this.destination.next?.(value);
    }
  }
 
  complete() {
    if (!this.closed) {
      this.closed = true;
      this.destination.complete?.();
      this.subscription.unsubscribe();
    }
  }
 
  error(err: any) {
    if (!this.closed) {
      this.closed = true;
      this.destination.error?.(err);
      this.subscription.unsubscribe();
    }
  }
}
 
class Observable<T> {
  constructor(private _wrappedFunc: (subscriber: Observer<T>) => () => void) {}
 
  subscribe(observer: Observer<T>) {
    const subscription = new Subscription();
    const subscriber = new SafeSubscriber(observer, subscription);
    subscription.add(this._wrappedFunc(subscriber));
    return subscription;
  }
}
 
const helloSocket = new Observable<string>((subscriber) => {
  const socket = new WebSocket('wss://echo.websocket.org');
 
  socket.onopen = () => {
    socket.send('Hello, World!');
  };
 
  socket.onmessage = (e) => {
    subscriber.next(e.data);
  };
 
  socket.onclose = (e) => {
    if (e.wasClean) {
      subscriber.complete();
    } else {
      subscriber.error(new Error('Socket closed dirty!'));
    }
  };
 
  return () => {
    if (socket.readyState <= WebSocket.OPEN) {
      socket.close();
    }
  };
});
 
const subscription = helloSocket.subscribe({
  next: console.log,
  complete: () => console.log('server closed'),
  error: console.error,
});
 
// Later, we can unsubscribe!
subscription.unsubscribe();

Это всё! Observables — это просто специализированные функции!

Я знаю, что это было очень МНОГО кода, надеюсь, пошаговое руководство помогло вам понять, что входит в тип observable (например, тип из rxjs), и помогло немного прояснить его. Это уж точно не волшебство.

Тогда я бы посоветовал вам спросить себя: «Для чего нужны observables?» и, честно говоря, самый короткий ответ: вы можете использовать observables для всего, для чего вы можете использовать функцию. Самая большая разница — это небольшой набор гарантий в отношении обратных вызовов и то, что, возможно, наиболее важно, в отношении прекращения работы.

Важно отметить, что описанная выше реализация НЕ является чем-то, что вы должны воссоздавать или использовать в продакшене. Я бы рекомендовал использовать observable из RxJS, поскольку он охватывает гораздо больше ситуаций, чтобы помочь сохранить ваш код надежным и безопасным при составлении сложных потоков данных.