Эффектное программирование. Часть 2: генераторы в полевых условиях

В прошлой статье я рассказал об итераторах и генераторах в javascript, но не привел никаких практичных примеров. Этой статьей я хочу заполнить этот пробел и, на примере простого веб-приложения, продемонстрировать паттерн с применением генераторов.

Эта статья также может быть полезна тем, кто хочет разобраться, как работает redux-saga.

Также я давно хотел познакомиться с Deno (альтернатива Node) и использовал его в качестве среды для запуска кода, так что примеры будут в этот раз на typescript.

TLDR - Код

Немного о Deno

Раз уж я начал говорить о Deno, немного расскажу о нём. Deno это среда исполнения кода на javascript и typescript, альтернатива Node. Основные отличия от Node:

• Поддержка typescript на нативном уровне (имеется в виду без ручной трансляции в javascript, этим занимается подсистема Deno)

• Система прав, построенная на флагах запуска

• Указание полных url вместо названий пакетов, соответственно, возможность использовать разные версия пакета в одной программе

Также у Deno свой набор стандартных утилит с более современным API, чем у node, например, любые асинхронные операции возвращают промисы, вместо передачи колбека.

Наше приложение

Итак, что будет представлять собой наше веб-приложение:

• Чат-бот, который будет уметь совсем немного:

  • Сообщать текущее время

  • Складывать числа

• Он будет уметь общаться сразу с несколькими пользователями

• Веб-интерфейс для этого дела

Выглядеть это всё будет вот так:

Для создания веб-сервера с поддержкой веб-сокетов в Deno не нужны специализированные библиотеки, для нашей задачи достаточно стандартных функций. Но сперва...

Ещё немного о генераторах

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

Асинхронные итераторы

При каждом вызове, итератор возвращает какое-то значение. Если для каждого шага возвращаемое значение это промис, такой итератор называют асинхронным. Для таких итераторов возможен обход с помощью специального варианта цикла: for await. Для этого объект, для которого мы хотим иметь поддержку обхода таким циклом, должен иметь специальный метод Symbol.asyncIterator, который возвращает итератор (генераторы, созданные с ключевым словом async, возвращают асинхронные итераторы).

Пример:

async function* timer() {
  let i = 0;
  while (true) {
    yield new Promise((resolve) => setTimeout(() => resolve(++i), 1000));
  }
}

for await (const tick of timer()) {
  console.log(tick);
}
// 1 2 3 ... 

yield*

Помимо оператора yield, для возврата текущего значения итератора, существует оператор yield*. Он принимает итератор в качестве параметра и последовательно возвращает все его значения. В генератор же он возвращает выходное значение итератора (первое значение, для которого done равно true, для генератора это значение, переданное в return).

Пример:

function* concat<T>(...iterables: Iterable<T>[]) {
  for (const iter of iterables) {
    yield* iter;
  }
}

for (const i of concat([1, 2], [3, 4])) {
  console.log(i);
}
// 1 2 3 4

Пишем код

Http сервер, слушающий на некотором порту, создается в deno очень просто:

import { serve } from "https://deno.land/std@0.74.0/http/server.ts";

const server = serve({ port: 8080 });
console.log("Listening on 8080");

for await (const req of server) {
  req.respond({ status: 200, body: 'Hello, world!' });
}

Github

Как можно видеть, сервер, результат вызова функции serve, является асинхронным итератором. Каждый вызов метода next, возвращает промис, который зарезолвится, когда поступит входящее соединение.

Далее от нашего сервера нам понадобятся две вещи: отдавать статичные файлы и принимать подключения websocket. Напишем простенький хелпер, который будет соединять обработчики запроса в цепочку по аналогии с мидлварами в express.

type MiddlewarePayload = {
  url: URL;
  req: ServerRequest;
};

type MiddlewareFn = (options: MiddlewarePayload) => Promise<true | undefined>;

const combineProcessors = (...fns: MiddlewareFn[]) => async (options: MiddlewarePayload) => {
  for (const fn of fns) {
    const result = await fn(options);
    if (result) {
      return result;
    }
  }
}

Github

Каждый обработчик, переданный в такой комбайнер должен вернуть промис, который должен зарезолвиться в true, если он обработал запрос, иначе он резолвит undefined.

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

const processors = combineProcessors(index, staticFiles, wsMiddleware);

const server = serve({ port: 8080 });

console.log("Listening on 8080");

// для нашей программы не важен host запроса,
// но он нужен для конструктора URL
const BASE = "http://localhost";
for await (const req of server) {
  const url = new URL(req.url, BASE);
  const result = await processors({ url, req });
  if (!result) {
    req.respond({ status: 404 });
  }
}

Github

Я не буду рассказывать, про функции index и staticFiles, они занимаются раздачей статичных файлов, их код можно почитать в github, если вдруг интересно. А вот на обработчике соединений websocket мы остановимся поподробнее.

Каналы

Описание модели каналов появилось задолго до рождения языка javascript. Эта модель описывает межпроцессное взаимодействие и коммуникации в асинхронной среде. Нативная реализация присутствует во многих современных языках: go, rust, kotlin, clojure, и т. д.

Если вы знакомы с такой структурой как stream, то переход к каналам будет довольно простым. Stream (поток), как и каналы, предоставляет асинхронный доступ к последовательным данным. Главное отличие - stream использует подписную модель доступа к данным (когда придет сообщение - вызови обработчик), а каналы блокирующую (давай следующее сообщение, и пока оно не придет, дальше не ходи). Вот пример использования:

/** Streams **/
const stream = new Stream();
stream.subscribe(callback);
// где-то далее по коду
stream.emit(data);

/** Channels **/
const ch = new Channel();
// где-то далее по коду
ch.put(data)
// где-то далее по коду
const data = await ch.take();

Вот так выглядит реализация каналов в нашем примере:

class Channel {
  private takers: Array<(payload: string) => void> = [];
  private buffer: string[] = [];  
  private callTakers() {
    while (this.takers.length > 0 && this.buffer.length > 0) {  
      const taker = this.takers.shift()!;
      const payload = this.buffer.shift()!;
      taker(payload);    
    }
  }
  take() {
    const p = new Promise<string>((resolve) => {
      this.takers.push(resolve);
    });
    this.callTakers(); 
    return p;
  }
  put(message: string) { 
    this.buffer.push(message);
    this.callTakers();
  }
  async listen(sock: WebSocket) {
    for await (const event of sock) {
      if (typeof event === "string") {
        this.put(event);
      }
    }
  }
}

Github

Поясню, что тут происходит:

• есть массив buffer, в который помещаются входящие сообщения

• есть массив takers, куда помещаются функции резолва промисов

• на каждый метод put (положить сообщение в канал) и take (дождаться и взять сообщение из канала) вызывается проверка, есть ли хотя бы одно сообщение в buffer и хотя бы один taker и, в этом случае, происходит резолв сообщения и удаление его из буфера и taker-а из takers

• также есть хелпер listen, который подписывается на все сообщения переданного socket-а и кладет их в канал

Зачем вообще нам понадобился канал? Чем не угодила подписная модель? Это будет видно далее, сейчас скажу только, что таким образом мы будем писать асинхронный код словно это синхронный код (собственно то, для чего создавались async/await).

А причем тут генераторы?

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

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

Назовем эффектом следующую структуру:

{
  type: string;
  [key: string]: any;
}

Если читатель знаком с архитектурой flux или redux, то сразу узнает эту форму - это же action! Да, в нашем случае эффект будет выполнять очень похожую задачу, только в redux работает такая формула:

const newState = reducer(state, action)

У нас же будет работать следующая схема:

while (true) {
  const { value: effect, done } = iter.next(current);
  // тут код обработки эффекта
  if (done) break;
}

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

export async function handleWs(sock: WebSocket) {
  const incoming = new Channel();
  incoming.listen(sock);

  let current: string = "";
  const iter = dialog();
  while (true) {
    const { value: effect, done } = iter.next(current);
    if (!effect) {
      break;
    }
    switch (effect.type) {
      case "say": {
        sock.send(effect.text);
        break;
      }
      case "listen": {
        current = await incoming.take();
       	break;
      }
    }
    if (done) {
      break;
    }
  }
}

Github

Эта функция вызывается при установке нового веб-сокет соединения.

У нас используется два вида эффектов (вообще их может быть сколько угодно):

• say - указывает на то, что нужно отправить ответ пользователю

• listen - нужно дождаться сообщения от пользователя

Хоть цикл и бесконечный, не произойдет блокировки процесса, так как внутри цикла есть await, который прервет его выполнение до прихода сообщения в канал. Благодаря использованию каналов, мы можем построить такой код.

Посмотрим, как выглядит сам диалог:

const say = (text: string) => ({ type: "say", text } as const);
const listen = () => ({ type: "listen" } as const)

function* dialog() {
  yield say('Welcome to "Do what I say BOT"');
  while (true) {
    const message: string = yield listen();
    if (message.toLowerCase().includes("time")) {
      yield say(`It is ${format(new Date(), "HH:mm:ss")}`);
    } else if (message.toLowerCase().includes("sum")) {
      yield* sumSubDialog();
    } else {
      yield say(`I don't know what to say!`);
    }
  }
}

function* sumSubDialog() {
  yield say("Okay, what numbers should we sum?");
  let result = 0;
  let message = yield listen();
  while (true) {
    const num = Number(message);
    if (isNaN(num)) {
      break;
    } else {
      result += num;
    }
    yield say("Got it!");
    message = yield listen();
  }
  yield say(`The result is: ${result}`);
}

Github

Диалог является генератором, который на каждом шаге возвращает какой-то эффект. Эффект say выполняется сразу, отправляя сообщение в сокет, и код генератора продолжает работу, не ожидая никаких данных от внешней среды. Эффект listen приостанавливает выполнение генератора до получения сообщения, которое сразу же передается в генератор.

Какие же плюсы у такого подхода, зачем вообще так заморачиваться?

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

• Удобство декомпозиции. С помощью оператора yield* можно разбить логику работы основного генератора на множество подпрограмм.

• Простота эффектов. Сами эффекты представляют собой довольно примитивные структуры, они хорошо типизируются, легко конструируются, их можно сериализовать и передавать по сети (не придумал зачем, но ведь можно).

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

Из главных минусов подхода могу отметить следующее: в генераторах не инферятся типы возврата оператора yield, что в целом логично, так как по сути код генератора является сигнатурой, из него уже инферится тип того, что можно передать в next созданного итератора. Но в нашем подходе генератор используется задом-наперед, поэтому типы для возвращаемых yield значений приходится писать вручную.

Хочу отдельно отметить важный аспект описанного подхода. Код внутри генератора (описание диалога) очень абстрактен. Он не занимается передачей сообщений по каналам, не обрабатывает ошибки передачи, и даже ничего не знает о среде, в которой исполняется. Это чистая бизнес-логика, максимально точное и бесшумное описание бизнес-процессов.

Также из вышесказанного следует, что в идеальном случае генератор должен быть чистой функцией. Для генератора это значит, что для одних и тех же параметров вызова генератора возвращаемые итераторы должны быть идентичными (одинаковая последовательность входных данных генерирует одинаковые последовательности возвращаемых значений). Чистота генераторов гарантирует инкапсуляцию бизнес-логики, то есть в нём описан один юзкейс. Также это позволяет легко написать тесты для вашей бизнес-логики: достаточно проверить, что последовательность эффектов, которые вернет генератор будет ожидаемой.

Что ещё почитать на тему

1. js-csp. Реализация csp в js с помощью генераторов.

2. redux-saga. Самая популярная реализация эффектов в js.

3. Сопрограммы.