GenEvent поведение

Модуль поведения для реализации функциональности обработки событий.

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

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

Пример

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

В качестве примера рассмотрим GenEvent, который накапливает сообщения до тех пор, пока они не будут собраны явным вызовом.

# Define an Event Handler
defmodule LoggerHandler do
  use GenEvent

  # Callbacks

  def handle_event({:log, x}, messages) do
    {:ok, [x | messages]}
  end

  def handle_call(:messages, messages) do
    {:ok, Enum.reverse(messages), []}
  end
end

# Start a new event manager.
{:ok, pid} = GenEvent.start_link([])

# Attach an event handler to the event manager.
GenEvent.add_handler(pid, LoggerHandler, [])
#=> :ok

# Send some events to the event manager.
GenEvent.notify(pid, {:log, 1})
#=> :ok

GenEvent.notify(pid, {:log, 2})
#=> :ok

# Call functions on specific handlers in the manager.
GenEvent.call(pid, LoggerHandler, :messages)
#=> [1, 2]

GenEvent.call(pid, LoggerHandler, :messages)
#=> []

Мы начинаем новый менеджер событий, вызвав GenEvent.start_link/1. Уведомления могут быть отправлены менеджеру событий, который затем вызовет handle_event/2 для каждого зарегистрированного обработчика.

Мы можем добавить новых обработчиков с помощью add_handler/3 и add_mon_handler/3. Вызовы также могут быть сделаны для определенных обработчиков с помощью call/3.

Обратные вызовы

В модуле GenEvent требуется реализовать 6 обратных вызовов. Добавив use GenEvent в свой модуль, Elixir автоматически определит все 6 обратных вызовов, оставив вам возможность реализовать те, которые вы хотите настроить.

Регистрация имен

GenEvent связан с теми же правилами регистрации имен, что и GenServer. Подробнее об этом в документации GenServer.

Режимы

GenEvent поддерживает три разных уведомления.

При GenEvent.ack_notify/2 менеджер подтверждает каждое событие, обеспечивая обратную связь, но обработка сообщения происходит асинхронно.

При GenEvent.sync_notify/2 менеджер подтверждает событие сразу после обработки его всеми обработчиками событий.

При GenEvent.notify/2 все события обрабатываются асинхронно, и нет подтверждения (что означает отсутствие обратной связи).

Потоковая передача

GenEvent сообщения могут быть переданы в потоковом режиме с помощью stream/2. Вам потребуется запустить другой процесс для обработки потока:

Task.start_link fn ->
  stream = GenEvent.stream(pid)

  # Discard the next 3 events
  _ = Enum.take(stream, 3)

  # Print all remaining events
  for event <- stream do
    IO.inspect event
  end
end

Теперь вызовите GenEvent.notify/2 несколько раз. Вы увидите, что первые три события будут пропущены, а остальные будут непрерывно выводиться.

Дополнительная информация и совместимость

Если вы хотите узнать больше о GenEvent, документация и ссылки в Erlang предоставят дополнительную информацию.

Однако следует учитывать, что Elixir и Erlang gen события не на 100% совместимы. :gen_event.add_sup_handler/3 не поддерживается GenEvent Elixir, который, в свою очередь, поддерживает GenEvent.add_mon_handler/3.

Преимущества подхода мониторинга описаны в разделе «Не пейте слишком много сладкого напитка» в ссылке «Научитесь Erlang» выше. Из-за этих изменений GenEvent Elixir по умолчанию не ловит выходы.

Кроме того, Elixir также приводит к нормализации {:error, _} кортежей, возвращаемых многими функциями, для большей согласованности с собой и модулем GenServer.

Резюме

Типы

handler(): Поддерживаемые значения для новых обработчиков
manager(): Ссылка на менеджер событий
name(): Имя менеджера GenEvent
on_start(): Значения возвращаемые функциями start*
options(): Параметры, используемые функциями start*

Функции

ack_notify(manager, event): Отправляет уведомление о подтверждении события менеджеру событий
add_handler(manager, handler, args): Добавляет новый обработчик событий к менеджеру событий
add_mon_handler(manager, handler, args): Добавляет контролируемый обработчик событий к менеджеру событий
call(manager, handler, request, timeout \\ 5000): Выполняет синхронный вызов обработчику событий, установленный в manager
notify(manager, event): Отправляет уведомление о событии менеджеру событий
remove_handler(manager, handler, args): Удаляет обработчик событий из менеджера событий
start(options \\ []): Запускает процесс менеджера событий без ссылок (вне дерева надзора)
start_link(options \\ []): Запускает менеджер событий, связанный с текущим процессом
stop(manager, reason \\ :normal, timeout \\ :infinity): Останавливает менеджер с заданным reason
stream(manager, options \\ []): Возвращает поток, который потребляет события из менеджера событий
swap_handler(manager, handler1, args1, handler2, args2): Заменяет старый обработчик событий новым в менеджере событий
swap_mon_handler(manager, handler1, args1, handler2, args2): Заменяет старый обработчик событий новым контролируемым обработчиком в менеджере событий
sync_notify(manager, event): Отправляет синхронное уведомление о событии менеджеру событий
which_handlers(manager): Возвращает список всех установленных обработчиков событий в менеджере событий

Обратные вызовы

code_change(old_vsn, state, extra): Вызывается для изменения состояния обработчика при загрузке другой версии модуля обработчика (горячая замена кода), и структура данных состояния должна быть изменена
handle_call(request, state): Вызывается для обработки синхронных call/4 сообщений определенному обработчику
handle_event(event, state): Вызывается для обработки notify/2, ack_notify/2 или sync_notify/2 сообщений
handle_info(msg, state): Вызывается для обработки всех остальных сообщений. Все обработчики запускаются в процессе GenEvent, поэтому сообщения, предназначенные для других обработчиков, должны игнорироваться с помощью универсального оператора catch
init(args): Вызывается, когда обработчик добавлен в процесс GenEvent. add_handler/3 (и add_mon_handler/3) заблокируются до возврата
terminate(reason, state): Вызывается, когда сервер собирается завершиться. Он должен выполнить необходимую очистку

Типы

handler()

handler() :: atom | {atom, term}

Поддерживаемые значения для новых обработчиков

manager()

manager() :: pid | name | {atom, node}

Ссылка на менеджер событий

name()

name() :: atom | {:global, term} | {:via, module, term}

Имя менеджера GenEvent

on_start()

on_start() :: {:ok, pid} | {:error, {:already_started, pid}}

Значения, возвращаемые функциями start*

options()

options() :: [{:name, name}]

Параметры, используемые функциями start*

Функции

ack_notify(manager, event)

ack_notify(manager, term) :: :ok

Отправляет уведомление о подтверждении события менеджеру событий.

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

См. notify/2 для получения дополнительной информации. Обратите внимание, что эта функция специфична для Elixir GenEvent и не работает с Erlang.

add_handler(manager, handler, args)

add_handler(manager, handler, term) :: :ok | {:error, term}

Добавляет новый обработчик событий в manager.

Менеджер событий вызовет init/1 обратный вызов с args для инициализации обработчика событий и его внутреннего состояния.

Если init/1 вернёт корректное значение, указывающее на успешное завершение, менеджер событий добавит обработчик событий, и эта функция вернёт :ok. Если обратный вызов завершится с reason или вернёт {:error, reason}, обработчик событий будет проигнорирован, и эта функция вернёт {:error, reason}.

Если указанный обработчик уже был установлен в менеджере, эта функция вернёт {:error, :already_present}.

Для установки нескольких экземпляров одного и того же обработчика необходимо использовать {Module, id} вместо Module. Обработчик можно будет затем использовать с {Module, id} вместо просто Module.

add_mon_handler(manager, handler, args)

add_mon_handler(manager, handler, term) ::
  :ok |
  {:error, term}

Добавляет отслеживаемый обработчик событий в manager.

Ожидает те же входные данные и возвращает те же значения, что и add_handler/3.

Отслеживаемые обработчики

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

Если вызывающий процесс позже завершится с reason, менеджер событий удалит обработчик событий, вызвав terminate/2 обратный вызов с {:stop, reason} в качестве аргумента. Если обработчик событий позже будет удалён, менеджер событий отправит сообщение {:gen_event_EXIT, handler, reason} вызывающему процессу. Причина может быть одной из следующих:

:normal — если обработчик событий был удалён из-за вызова remove_handler/3 или :remove_handler был возвращён функцией обратного вызова
:shutdown — если обработчик событий был удалён, потому что менеджер событий завершается
{:swapped, new_handler, pid} — если идентификатор процесса (PID) заменил обработчик событий другим
term — если обработчик событий удаляется из-за ошибки. Какой термин используется, зависит от ошибки

Помните, что сообщение {:gen_event_EXIT, handler, reason} не гарантируется, что будет доставлено в случае сбоя менеджера. Если вы хотите гарантировать доставку сообщения, у вас есть два варианта:

отслеживать менеджер событий
связать с менеджером событий, а затем установить Process.flag(:trap_exit, true) в вашем обратном вызове обработчика

Наконец, эта функциональность работает только с GenEvent, запущенным через этот модуль (она не обратной совместима с :gen_event Erlang).

call(manager, handler, request, timeout \\ 5000)

call(manager, handler, term, timeout) ::
  term |
  {:error, term}

Выполняет синхронный вызов обработчику событий handler, установленного в manager.

Указанный request отправляется, и вызывающий процесс ожидает получения ответа или истечения времени ожидания. Менеджер событий вызовет handle_call/2 для обработки запроса.

Значение возврата reply определяется в возвращаемом значении handle_call/2. Если указанный обработчик событий не установлен, функция возвращает {:error, :not_found}.

notify(manager, event)

notify(manager, term) :: :ok

Отправляет уведомление о событии в manager.

Менеджер событий вызовет handle_event/2 для каждого установленного обработчика событий.

notify является асинхронной и вернёт сразу после отправки уведомления. notify не завершится с ошибкой, даже если указанный менеджер событий не существует, если он не задан как атом.

remove_handler(manager, handler, args)

remove_handler(manager, handler, term) ::
  term |
  {:error, term}

Удаляет обработчик событий из manager.

Менеджер событий вызовет terminate/2 для завершения обработчика событий и возвратит значение обратного вызова. Если указанный обработчик событий не установлен, функция возвращает {:error, :not_found}.

start(options \\ [])

start(options) :: on_start

Запускает процесс менеджера событий без связей (вне дерева управления).

См. start_link/1 для получения дополнительной информации.

start_link(options \\ [])

start_link(options) :: on_start

Запускает менеджер событий, связанный с текущим процессом.

Это часто используется для запуска GenEvent в рамках дерева управления.

Он принимает опцию :name, которая описана в разделе Name Registration в документации модуля GenServer.

Если менеджер событий успешно создан и инициализирован, функция возвращает {:ok, pid}, где pid — PID сервера. Если процесс с заданным именем сервера уже существует, функция возвращает {:error, {:already_started, pid}} с PID этого процесса.

Обратите внимание, что GenEvent, запущенный с помощью start_link/1, связан с родительским процессом и завершится не только при сбоях, но и если родительский процесс завершится с :normal причиной.

stop(manager, reason \\ :normal, timeout \\ :infinity)

stop(manager, reason :: term, timeout) :: :ok

Останавливает менеджер с заданной reason.

Перед завершением менеджер событий вызовет terminate(:stop, ...) для каждого установленного обработчика событий. Он возвращает :ok если менеджер завершается с заданной причиной, если он завершается по другой причине, вызов завершится.

Эта функция сохраняет семантику OTP в отношении обработки ошибок. Если причина отличается от :normal, :shutdown или {:shutdown, _}, будет зарегистрирован отчёт об ошибке.

stream(manager, options \\ [])

stream(manager, Keyword.t) :: GenEvent.Stream.t

Возвращает поток, который потребляет события из manager.

Поток является структурой GenEvent, которая реализует протокол Enumerable. Потребление событий начинается только при запуске перечисления.

Обратите внимание, что потоковая передача — это функция, специфичная для Elixir’s GenEvent, и не работает с Erlang.

Опции

:timeout — вызывает исключение, если событие не поступает в течение X миллисекунд (по умолчанию :infinity)

swap_handler(manager, handler1, args1, handler2, args2)

swap_handler(manager, handler, term, handler, term) ::
  :ok |
  {:error, term}

Заменяет старый обработчик событий новым в manager.

Сначала старый обработчик событий удаляется с помощью вызова terminate/2 с указанными args1 и собирает возвращаемое значение. Затем новый обработчик событий добавляется и инициализируется с помощью вызова init({args2, term}), где term является возвращаемым значением вызова terminate/2 в старом обработчике. Это позволяет передавать информацию от одного обработчика к другому.

Новый обработчик будет добавлен, даже если указанный старый обработчик событий не установлен или если обработчик не завершился с заданной причиной, в этом случае state = {:error, term}.

Если init/1 во втором обработчике возвращает корректное значение, эта функция возвращает :ok.

swap_mon_handler(manager, handler1, args1, handler2, args2)

swap_mon_handler(manager, handler, term, handler, term) ::
  :ok |
  {:error, term}

Заменяет старый обработчик событий новым отслеживаемым в manager.

Подробнее см. документацию к add_mon_handler/3 и swap_handler/5.

sync_notify(manager, event)

sync_notify(manager, term) :: :ok

Отправляет синхронное уведомление о событии в manager.

Другими словами, эта функция возвращает :ok только после того, как менеджер событий вызовет handle_event/2 обратный вызов для каждого установленного обработчика событий.

См. notify/2 для получения дополнительной информации.

which_handlers(manager)

which_handlers(manager) :: [handler]

Возвращает список всех установленных обработчиков событий в manager.

Обратные вызовы

code_change(old_vsn, state, extra)

code_change(old_vsn, state :: term, extra :: term) :: {:ok, new_state :: term} when old_vsn: term | {:down, term}

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

old_vsn — это предыдущая версия модуля (определённая атрибутом @vsn), при обновлении. При понижении версии предыдущая версия обернута в 2-кортеж с первым элементом :down. state — текущее состояние обработчика, а extra — любые дополнительные данные, необходимые для изменения состояния.

Возвращение {:ok, new_state} изменяет состояние на new_state и замена кода выполняется успешно.

Если code_change/3 вызывает исключение, замена кода завершается с ошибкой, и обработчик продолжит работать со своим предыдущим состоянием. Поэтому этот обратный вызов обычно не содержит побочных эффектов.

handle_call(request, state)

handle_call(request :: term, state :: term) ::
  {:ok, reply, new_state} |
  {:ok, reply, new_state, :hibernate} |
  {:remove_handler, reply} when reply: term, new_state: term

Вызывается для обработки синхронных сообщений call/4 для конкретного обработчика.

request — сообщение запроса, отправленное с помощью call/4, а state — текущее состояние обработчика.

Возвращение {:ok, reply, new_state} отправляет reply в качестве ответа на вызов и устанавливает состояние обработчика в new_state.

Возвращение {:ok, reply, new_state, :hibernate} аналогично {:ok, reply, new_state} за исключением того, что процесс приостанавливается. Подробнее см. handle_event/2 для получения дополнительной информации о приостановке.

Возвращение {:remove_handler, reply} отправляет reply в качестве ответа на вызов, удаляет обработчик из цикла GenEvent и вызывает terminate/2 с причиной :remove_handler и состоянием state.

handle_event(event, state)

handle_event(event :: term, state :: term) ::
  {:ok, new_state} |
  {:ok, new_state, :hibernate} |
  :remove_handler when new_state: term

Вызывается для обработки сообщений notify/2, ack_notify/2 или sync_notify/2.

event — сообщение события, а state — текущее состояние обработчика.

Возвращение {:ok, new_state} устанавливает состояние обработчика в new_state и цикл GenEvent продолжается.

Возвращение {:ok, new_state, :hibernate} аналогично {:ok, new_state}, за исключением того, что процесс приостанавливается, как только все обработчики обработают события. Процесс GenEvent продолжит цикл, как только сообщение окажется в его очереди сообщений. Если сообщение уже находится в очереди сообщений, это произойдет немедленно. Приостановка GenEvent вызывает сборку мусора и оставляет непрерывную кучу, что минимизирует использование памяти процессом.

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

Возвращение :remove_handler удаляет обработчик из цикла GenEvent и вызывает terminate/2 с причиной :remove_handler и состоянием state.

handle_info(msg, state)

handle_info(msg :: term, state :: term) ::
  {:ok, new_state} |
  {:ok, new_state, :hibernate} |
  :remove_handler when new_state: term

Вызывается для обработки всех остальных сообщений. Все обработчики выполняются в процессе GenEvent, поэтому сообщения, предназначенные для других обработчиков, следует игнорировать с помощью catch-all-блока.

msg — это сообщение, а state — текущее состояние обработчика.

Значения возврата такие же, как у handle_event/2.

init(args)

init(args :: term) ::
  {:ok, state} |
  {:ok, state, :hibernate} |
  {:error, reason :: any} when state: any

Вызывается, когда обработчик добавляется в процесс GenEvent. add_handler/3 (и add_mon_handler/3) будут блокированы до возврата.

args — это термин аргумента (третий аргумент), переданный в add_handler/3.

Возвращение {:ok, state} заставит add_handler/3 вернуть :ok и обработчик станет частью цикла GenEvent со состоянием state.

Возвращение {:ok, state, :hibernate} аналогично {:ok, state}, за исключением того, что процесс GenEvent приостанавливается перед продолжением цикла. См. handle_event/2 для получения дополнительной информации о приостановке.

Возвращение {:error, reason} заставит add_handler/3 вернуть {:error, reason} и обработчик не будет добавлен в цикл GenEvent.

terminate(reason, state)

terminate(reason, state :: term) :: term when reason: :stop | {:stop, term} | :remove_handler | {:error, term} | term

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

reason — причина удаления, а state — текущее состояние обработчика. Значение возврата возвращается в GenEvent.remove_handler/3 или игнорируется, если обработчик удаляется по другой причине.

reason является одним из:

:stop — менеджер завершает работу
{:stop, term} — завершен контролируемый процесс (для контролируемых обработчиков)
:remove_handler — обработчик удаляется
{:error, term} — обработчик упал или вернул неверное значение, и об ошибке ведётся запись в журнал
term — любой термин, переданный в функции, такие как GenEvent.remove_handler/3

Если обработчик является частью дерева надзора, GenEvent’s Supervisor отправит сигнал выхода при завершении. Сигнал выхода зависит от стратегии завершения в спецификации дочернего элемента. Если это :brutal_kill, GenEvent убивается, и поэтому terminate/2 для его обработчиков не вызывается. Однако, если это таймаут, Supervisor отправит сигнал выхода :shutdown, и GenEvent будет иметь время таймаута для вызова terminate/2 для всех своих обработчиков — если процесс всё ещё жив после таймаута, он убивается.

Если GenEvent получает сигнал выхода (который не является :normal) от любого процесса, когда он не ловит выходы, он завершается внезапно с той же причиной и, следовательно, не вызывает terminate/2 обработчиков. Обратите внимание, что процесс НЕ ловит выходы по умолчанию, и сигнал выхода отправляется при завершении связанного процесса или отключении его узла. Поэтому нет гарантии, что terminate/2 будет вызван при завершении GenEvent.

Следует соблюдать осторожность при очистке, так как GenEvent может продолжать цикл после удаления обработчика. Это отличается от большинства других поведений OTP. Например, если обработчик контролирует port (например, :gen_tcp.socket) или File.io_device, он должен быть закрыт в terminate/2, поскольку процесс не завершается, поэтому он не будет автоматически очищен.