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 могут предоставить дополнительную информацию.

• :gen_event документация модуля
• Обработчики событий — Узнайте Erlang!

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

Преимущества подхода мониторинга описаны в разделе «Не пейте слишком много Kool-Aid» в ссылке «Learn you some Erlang» выше. Из-за этих изменений Elixir’s GenEvent по умолчанию не перехватывает завершения.

Кроме того, 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 \\ [])

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

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

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

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

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

sync_notify(manager, event)

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

which_handlers(manager)

Возвращает список всех обработчиков событий, установленных в 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, поэтому сообщения, предназначенные для других обработчиков, должны быть проигнорированы с помощью общего обработчика ошибок

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 для получения дополнительной информации. Обратите внимание, что эта функция специфична для GenEvent в Elixir и не работает с 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, запущенным через этот модуль (она несовместима с Erlang’s :gen_event).

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, поэтому сообщения, предназначенные для других обработчиков, следует игнорировать с помощью универсального обработчика.

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/0, его необходимо закрыть в terminate/2, так как процесс не завершается, поэтому он не будет очищен автоматически.