В эрланге (и эликсире) мне всегда недоставало способа организовать «потоковый» обмен сообщениями, наподобие того, который обеспечивает какой-нибудь Message Broker. Нормальные разработчики смиряются с ограничениями, которые им задают их фреймворки: в Финиксе есть PubSub, в OTP — :gen_event, в эликсире — депрекейтнутый еще до рождения GenEvent.
Ни один из них меня не устраивал. PubSub — штука мощная, но асинхронность прибита там гвоздями, а иногда все-таки надо вызывать подписчиков синхронно (я понимаю, что можно высылать вместе с сообщением свой pid и дожидаться ответов, но этот ad-hoc, к сожалению, не поможет в ситуации, когда количество подписчиков неизвестно). :gen_event — почти то, что нужно, но из его дизайна прямо торчат уши сайд-эффектных обработчиков, из-за чего удобная фильтрация входящего потока сообщений превращается в спагетти. В общем, всё как всегда: сам не сделаешь — никто не сделает.
Так родилась библиотека Antenna, которая предоставляет все те возможности, которые обычно обеспечиваются посредством вкрячивания дополнительной зависимости от брокера сообщений. Требования, которые я к ней предъявлял, были следующими:
готовность к back pressure из коробки (миллион ивентов, отосланных одномоментно, не должны повалить обработчики, а добавление новых нод в кластер должно их прозрачно разгружать)
каналы (топики, теги), куда можно отослать ивент, который будет доставлен всем зарегистрированным обработчикам
обработчик (помимо подписок на каналы) должен иметь возможность прозрачно фильтровать входящий поток сообщений, используя всю мощь паттерн-матчинга эрланга
как обработчики, так и подписки, — должны успешно восстанавливаться после краша
отправитель сообщения может затребовать синхронное выполнение обработчиков, блокируя вызывающий процесс до получения всех ответов (аналог rpc)
библиотека должна предоставлять удобные примитивы для тестирования её использования
Те, кому доводилось использовать RabbitMQ, без труда заметят, откуда я воровал идеи.
Псевдокод
С давних пор я использую некое подобие TDD с запахом DDD, или наоборот, я не силен в аббревиатурной баззвордистике. В общем, я начинаю проектирование любого куска кода с вызова его (пока несуществующего) API. Первым делом добиваюсь того, чтобы вызывать мой код было легко и приятно, а с реализацией я уж как-нибудь разберусь потом. Поэтому я набросал что-то вот такое:
{:ok, pid} = match({:tag_answer, _}, self())
subscribe(:chan, pid)
event(:chan, {:tag_answer, 42})
receive do
{:event, :chan, {:tag_answer, 42}} -> :ok
after
1_000 -> :error
endПервые две строки — регистрация обработчика и подписка. Следующая — отсылка сообщения. Заключительный блок — проверка, что сообщение получено. Эти три куска обычно будут находиться в разных местах кода, никак не связанных между собой.
Первое, что я заметил, — первые две строки почти всегда ходят парой, поэтому имеет смысл принимать список каналов для подписки прямо в определении «матчера» (не отменяя, разумеется, функцию subscribe/2, если кому-то захочется присосаться к каналу позже).
Еще я заметил, что обработчиков может быть тоже много, почему нет.
Таким образом, определяющим уникальным идентификатором для матчера — остаётся собственно матч. Он может быть довольно заковыристым, поэтому в качестве id я выбрал его текстовое представление (строка "{:tag_answer, _}" для примера выше). Это решение мне не очень нравится, но лучшего у меня (пока) нет. Как минимум, инспектировать матчеры так гораздо проще, чем, например, если использовать что-то хешеподобное.
Итак, у нас получается отношение много ко многим: много обработчиков на один матч, которые могут получать сообщения из многих каналов. Например, мы можем подписаться на все сообщения вида {:error, _}, прилетевшие из всех каналов, и приспособить два обработчика: логгер сплюнет в консоль, телеметрия отошлет что-нибудь в графану, или куда там принято всё отсылать.
Основная архитектура
Back pressure в эликсире подразумевает использование библиотеки GenStage. Я уже реализовывал на ней свой Throttler, теперь настал черед брокера. Мы помним про горизонтальное масштабирование — а значит, — несколько консьюмеров, хотя бы по одному на ноде. Каждый консьюмер будет высылать сообщения в указанные каналы, матчеры — проверять, надо ли вообще суетиться (матчится ли сообщение), и если да — вызывать обработчики. Вроде звучит адекватно.
Консьюмеры я размазал по нодам при помощи безымянных процессов под управлением DistributedSupervisor, броадкастер — тоже (только этот процесс именован, а значит — один на кластер, и будет перезапущен на другой ноде, если текущая скроется в тумане).
Каждый матчер — тоже процесс, который в своём стейте хранит список обработчиков и, собственно, матч. Тут меня поджидала первая архитектурная дилемма: как хранить матч? {:foo, _} — просто так не сохранишь, такой код допустим только как LHO в собственно прямых вызовах паттерн-матчинга, а хранить его AST — не вариант, потому что его тогда не вставить в матч. В общем, я в результате решил генерировать функцию-матчер (всё равно реализация match/4 возможна только в виде макроса):
quote generated: true, location: :keep do
matcher = fn
unquote(match) -> true
_ -> false
end
…
endНу, отлично. Теперь в самом процессе матчера появляется колбэк наподобие вот такого:
@impl GenServer
@doc false
def handle_cast({:handle_event, channel, event}, state) do
if state.matcher.(event) do
Enum.each(state.handlers, fn
handler when is_function(handler, 1) -> handler.(event)
handler when is_function(handler, 2) -> handler.(channel, event)
process -> send(process, {:antenna_event, channel, event})
end)
end
{:noreply, state}
endИ мы готовы слать ему сообщения из консьюмеров (матчеры тоже находятся под управлением DistributedSupervisor, то есть, равномерно размазаны по кластеру.
В этот момент я сказал git init, потому что MVP уже вырисовывался.
Синхронный вызов
Да этого момента всё было довольно-таки тривиально. Но как организовать синхронный вызов, когда все сообщения проходят через семь кругов ада (броадкаст, консьюмеры, матчеры — и все на хрен пойми каких нодах в кластере)?
Вы когда-нибудь задумывались, почему сигнатура асинхронного колбэка GenServer’a — арности 2, а синхронного — 3? Кстати, это один из моих любимых вопросов на собеседовании: сразу становится понятно, мечтательный формошлёп перед тобой, или низкоуровневый фрик-социопат.
Второй аргумент в колбэке handle_call/3 — это идентификатор процесса (без потери общности), дожидающегося синхронного ответа. И вместо туплы {:reply, result, state} — из этого колбэка можно вернуть {:noreply, new_state}, а когда-нибудь потом уже выслать синхронный ответ напрямую вызывавшему процессу при помощи GenServer.reply/2. Если вы этого не знали, выпейте за меня ковшичек виски: это тот молоток, который всё вокруг вас сделает гвоздями.
GenStage, в свою очередь, тоже экспортирует reply/2. Поэтому теперь мне просто надо пробросить from через все консьюмеры и матчеры, а потом глубоко внутри написать что-то вроде:
Enum.each(results, fn
{nil, _} -> :ok
{from, results} -> GenStage.reply(from, results)
end)И если это был асинхронный вызов — никакого from нет, и мы ничего не делаем. А если он есть — мы высылаем ему назад аккумулированные результаты вызова всех хендлеров (и пущай он, гад, подавится).
Вот, кажется, и всё, что я хотел рассказать сегодня. Ссылка на библиотеку, её исходный код и тесты — выше.
Удачного брокеринга!
