Раздел 46 · Effect: Тихая Академия

Поток вместо массива

middle~25 мин

без входаВойди, чтобы отмечать прогресс, вести заметки и решать задачи в редакторе. войти

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

В уроке 84 башня разбирала ночь функцией nightIntake(lines: readonly string[]). Подпись честная, но она врёт о мире. Чтобы получить массив строк, надо дождаться конца ночи. А у Магистрали конца нет: строки идут и идут, утром, днём и следующей ночью. Башня, которая ждёт конца массива, не отвечает никогда. Даже если конец бывает, массив целой ночи лежит в памяти, хотя каждой строке нужна только она сама. И первый кадр ждёт, пока придёт последний.

Нужна та же разборка, только по одной строке, по мере прихода, без конца. Для этого в Effect есть поток, Stream.

Эффект, у которого много ответов

Эффект Effect<A, E, R> это рецепт, который при запуске даёт один ответ A или ошибку E. Поток Stream<A, E, R> это рецепт, который при запуске даёт сколько угодно ответов A, один за другим, а потом кончается, падает с ошибкой E или не кончается вовсе. R значит то же, что у эффекта: какие сервисы нужны.

import { Effect, Stream } from 'effect';

const shutters = Stream.fromIterable(['T-1', 'T-2', 'T-3']);

const night = Stream.runCollect(shutters);
//    ^? Effect<Array<string>, never, never>

console.log(await Effect.runPromise(night)); // [ 'T-1', 'T-2', 'T-3' ]

Первая строка ничего не делает: Stream.fromIterable только описывает поток из трёх значений. Stream.runCollect превращает поток в эффект, который соберёт все значения в массив. Работа начинается, только когда этот эффект запустят. Всё как в уроке 36: описание отдельно, исполнение отдельно. Если запустить night дважды, поток пройдёт дважды, с начала.

runCollect годится только для потоков, которые кончаются: из бесконечного он соберёт массив тогда же, когда nightIntake получит свой, то есть никогда. Для ночи Магистрали нужен запуск, который делает работу с каждым значением и ничего не копит:

const program = Stream.runForEach(shutters, (code) => Effect.log(`башня ${code} на связи`));

Семейство run это край потока, как runPromise край эффекта. Сегодня нужны два: runCollect для тестов и коротких потоков и runForEach для настоящей работы. Остальные (runFold, runDrain, run со стоком) появятся в уроке 88.

Кто кого тянет

Строки приходят к башне двумя способами, и между ними разница больше, чем кажется.

Провод, из которого берут. Башня сама просит следующую строку, когда готова: «читай». Если башня занята, провод ждёт. Так устроен read() у сокета, так устроен курсор базы данных, так устроена очередь, из которой берут задания.

Линия, которая толкает. Строка приходит, когда пришла, и вызывает твой обработчик: onLine(line). Готова ли башня, линия не спрашивает. Так устроены события браузера, веб-сокеты, EventEmitter в Node.

Поток Effect устроен первым способом: он тянущий (pull). Каждый шаг потока просит следующее значение у шага перед ним, только когда сам готов его принять. Это главное, что надо унести из урока: в тянущем потоке медленный конец сам замедляет быстрое начало, и никто не тонет. В следующем уроке это получит имя.

Провод, из которого берут

Пусть у провода есть эффект read, который отдаёт следующую строку или undefined, если провод замолчал. Поток из него:

import { Cause, Effect, Stream } from 'effect';

const fromWire = (read: Effect.Effect<string | undefined>): Stream.Stream<string> =>
  Stream.fromEffectRepeat(
    Effect.flatMap(read, (line) => (line === undefined ? Cause.done() : Effect.succeed(line))),
  );

Stream.fromEffectRepeat повторяет эффект и отдаёт каждый его ответ как значение потока. Повторяет вечно, пока эффект не закончится особой ошибкой Cause.done(): она значит не «беда», а «всё, значений больше нет». Тип её из потока убирает сам fromEffectRepeat, поэтому ошибка у потока never.

Сколько раз позовут read? Ровно столько, сколько строк попросили:

const first = Stream.runCollect(Stream.take(fromWire(read), 2));

Stream.take(2) просит две строки и закрывает поток. Третий read не случится, потому что третью строку никто не попросил. Это и есть «тянуть»: работа идёт от потребителя к источнику, а не наоборот.

Линия, которая толкает

Линию Магистрали не попросишь подождать. Её можно только слушать:

export interface Feed {
  readonly listen: (onLine: (line: string) => void, onEnd: () => void) => () => void;
}

listen подписывает два обработчика и возвращает функцию отписки. Чтобы из такой линии вышел тянущий поток, между ними нужна очередь: линия толкает в неё, поток тянет из неё. Это делает Stream.callback:

import { Effect, Queue, Stream } from 'effect';

export const fromFeed = (feed: Feed): Stream.Stream<string> =>
  Stream.callback<string>((queue) =>
    Effect.acquireRelease(
      Effect.sync(() =>
        feed.listen(
          (line) => Queue.offerUnsafe(queue, line),
          () => Queue.endUnsafe(queue),
        ),
      ),
      (unsubscribe) => Effect.sync(unsubscribe),
    ),
  );

Разберём по строкам.

  • Stream.callback создаёт очередь и даёт её функции. То, что функция положит в очередь, станет значениями потока.
  • Queue.offerUnsafe(queue, line) кладёт строку синхронно, прямо из обработчика линии. «Unsafe» тут значит «вне эффекта»: обработчик это обычная функция, yield* в ней не напишешь.
  • Queue.endUnsafe(queue) говорит потоку, что значений больше не будет.
  • Effect.acquireRelease из урока 55: подписка это ресурс. Подписались при запуске потока, отписались, когда поток закончился, упал или его перестали читать. Stream.callback держит для этого свою область, Scope, и закрывает её вместе с потоком.

Последний пункт легко пропустить, и о нём «Где тут дыра» ниже.

Для настоящих событий в Effect есть готовый Stream.fromEventListener(target, 'message'): внутри он устроен ровно так же, callback, offerUnsafe и acquireRelease вокруг addEventListener и removeEventListener. Своя Feed нужна потому, что у линии Магистрали нет EventTarget.

Шаги потока

Над потоком работают те же глаголы, что над массивом и над эффектом. Stream.map меняет каждое значение, Stream.filter пропускает только подходящие, Stream.mapEffect делает с каждым значением эффект и пропускает дальше его ответ. Все три ленивые: они описывают новый поток, а не бегут по старому.

Из строк провода башне нужны кадры. Разбирает их схема Incoming из урока 84, это эффект, значит mapEffect:

import { Effect, Option, Schema, Stream } from 'effect';

export const frames = <E, R>(lines: Stream.Stream<string, E, R>): Stream.Stream<FrameV2, E, R> =>
  lines.pipe(
    Stream.mapEffect((line) => Effect.option(Schema.decodeUnknownEffect(Incoming)(line))),
    Stream.filter(Option.isSome),
    Stream.map((frame) => frame.value),
  );

Effect.option превращает отказ разбора в Option.none(), а кадр в Option.some(кадр): отказ больше не ошибка потока, а значение. Stream.filter(Option.isSome) пропускает только кадры, и TypeScript сужает тип по предикату, как у массива. Stream.map достаёт кадр из Some.

Обрати внимание на E и R в подписи. Если у строк была ошибка, например обрыв провода, она проходит сквозь frames нетронутой: frames не знает о ней и не должна знать. Мусор же выбрасывается молча. Запомни это слово, «молча»: Липвигу оно не понравится.

Под капотом

Где тут дыра

const lines = Stream.callback<string>((queue) =>
  Effect.sync(() => {
    magistral.listen(
      (line) => Queue.offerUnsafe(queue, line),
      () => Queue.endUnsafe(queue),
    );
  }),
);

На его примере всё работает: Ринцвальд собирает ночь через runCollect, линия утром зовёт onEnd, поток кончается. Теперь соседний пример: башне нужны только первые три кадра, чтобы проверить линию после ремонта, Stream.take(lines, 3). Поток закрывается после третьей строки, очередь закрывается вместе с ним. А подписка остаётся: линия до утра толкает строки в закрытую очередь, и каждая проверка добавляет ещё одного слушателя. К утру у линии десяток подписчиков, которые никого не слушают. Отписка нужна не тогда, когда линия кончилась, а тогда, когда поток перестали читать, и знает об этом только поток. Поэтому отписка живёт в acquireRelease внутри callback, а не в голове автора.

Магистраль, этап 10

Протокол Магистрали становится ретранслятором. Новый модуль src/relay.ts: поток из линии, которая толкает, поток из провода, из которого берут, кадры из строк и фильтр кадров по адресу. Тесты считают подписки и чтения: лишнее чтение или забытая отписка видны сразу.

Практика

Глубже в архиве

Потоки и асинхронные итераторы без Effect разобраны в уроке про стримы раздела про асинхронность.

Дальше

К полуночи башня уже читает линию потоком. Липвиг смотрит, как строки бегут по журналу, и качает головой.

Ответ Почтамта ждёт в следующем уроке.