Packages
ex_esdb
0.8.0
0.11.0
0.10.0
0.9.0
0.8.0
0.7.8
0.7.7
0.7.6
0.7.5
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.1
0.6.0
0.5.1
0.5.0
0.4.8
0.4.7
0.4.6
0.4.5
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.3.3
0.3.2
0.3.1
0.3.0
0.2.5
0.2.4
0.2.3
0.2.2
0.2.1
0.2.0
0.1.7
0.1.6
0.1.5
0.1.4
0.1.3
0.1.2
0.1.1
0.1.0
0.0.20
0.0.19
0.0.18
0.0.17
0.0.16
0.0.15
0.0.14-alpha
0.0.13-alpha
0.0.12-alpha
0.0.11-alpha
0.0.10-alpha
0.0.9-alpha
0.0.8-alpha
0.0.6-alpha
0.0.5-alpha
0.0.4-alpha
0.0.3-alpha
0.0.2-alfa
0.0.1-alfa
ExESDB is a reincarnation of rabbitmq/khepri, specialized for use as a BEAM-native event store.
Current section
Files
Jump to
Current section
Files
lib/ex_esdb/emitter_pool_logging_worker.ex
defmodule ExESDB.EmitterPoolLoggingWorker do
@moduledoc """
Logging worker responsible for handling EmitterPool logging events.
This worker subscribes to emitter_pool logging events and processes them
according to configured logging policies.
"""
use GenServer
require Logger
alias Phoenix.PubSub
alias ExESDB.StoreNaming
alias ExESDB.Themes
defstruct [
:store_id,
:logging_level,
:terminal_output_enabled
]
def start_link(opts) do
store_id = StoreNaming.extract_store_id(opts)
name = StoreNaming.genserver_name(__MODULE__, store_id)
GenServer.start_link(__MODULE__, opts, name: name)
end
def child_spec(opts) do
store_id = StoreNaming.extract_store_id(opts)
%{
id: StoreNaming.child_spec_id(__MODULE__, store_id),
start: {__MODULE__, :start_link, [opts]},
type: :worker,
restart: :permanent,
shutdown: 5_000
}
end
@impl GenServer
def init(opts) do
store_id = StoreNaming.extract_store_id(opts)
logging_level = Keyword.get(opts, :emitter_pool_logging_level, :info)
terminal_output = Keyword.get(opts, :emitter_pool_terminal_output, false) # Default to false - pools can be chatty
# Subscribe to emitter_pool logging events
:ok = PubSub.subscribe(:ex_esdb_logging, "logging:emitter_pool")
:ok = PubSub.subscribe(:ex_esdb_logging, "logging:store:#{store_id}")
state = %__MODULE__{
store_id: store_id,
logging_level: logging_level,
terminal_output_enabled: terminal_output
}
Logger.debug("EmitterPoolLoggingWorker started for store #{store_id}")
{:ok, state}
end
@impl GenServer
def handle_info({:log_event, %{component: :emitter_pool} = event}, state) do
process_logging_event(event, state)
{:noreply, state}
end
@impl GenServer
def handle_info({:log_event, _event}, state) do
# Ignore events from other components
{:noreply, state}
end
@impl GenServer
def handle_info(_msg, state) do
{:noreply, state}
end
defp process_logging_event(event, state) do
try do
%{
event_type: event_type,
store_id: store_id,
pid: pid,
message: message,
metadata: metadata
} = event
# Only process events for our store
if store_id == state.store_id do
case event_type do
:startup ->
handle_startup_event(pid, message, metadata, state)
:shutdown ->
handle_shutdown_event(pid, message, metadata, state)
:action ->
handle_action_event(pid, message, metadata, state)
:error ->
handle_error_event(pid, message, metadata, state)
_ ->
handle_generic_event(event_type, pid, message, metadata, state)
end
end
rescue
error ->
Logger.warning("EmitterPoolLoggingWorker received malformed event: #{inspect(event)}, error: #{inspect(error)}")
end
end
defp handle_startup_event(pid, message, metadata, state) do
if state.terminal_output_enabled do
pool_name = Map.get(metadata, :pool_name, "unknown")
sub_topic = Map.get(metadata, :sub_topic, "unknown")
emitter_count = Map.get(metadata, :emitter_count, 0)
IO.puts("")
IO.puts("")
IO.puts("┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓")
IO.puts("#{Themes.emitter_pool_success_msg(pid, " 🚀 EMITTER POOL STARTUP 🚀 ")}")
IO.puts("┣━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┫")
IO.puts("#{Themes.emitter_pool_success_msg(pid, " Pool Name: #{pool_name}")}")
IO.puts("#{Themes.emitter_pool_success_msg(pid, " Store ID: #{state.store_id}")}")
IO.puts("#{Themes.emitter_pool_success_msg(pid, " Topic: #{sub_topic}")}")
IO.puts("#{Themes.emitter_pool_success_msg(pid, " Workers: #{emitter_count}")}")
IO.puts("#{Themes.emitter_pool_success_msg(pid, " PID: #{inspect(pid)}")}")
IO.puts("┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛")
IO.puts("")
IO.puts("")
end
Logger.info("[EmitterPool:#{state.store_id}] #{message}", metadata)
end
defp handle_shutdown_event(pid, message, metadata, state) do
if state.terminal_output_enabled do
pool_name = Map.get(metadata, :pool_name, "unknown")
sub_topic = Map.get(metadata, :sub_topic, "unknown")
reason = Map.get(metadata, :reason, "Manual Stop")
IO.puts("")
IO.puts("")
IO.puts("┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓")
IO.puts("#{Themes.emitter_pool_failure_msg(pid, " 🚨 EMITTER POOL SHUTDOWN 🚨 ")}")
IO.puts("┣━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┫")
IO.puts("#{Themes.emitter_pool_failure_msg(pid, " Pool Name: #{pool_name}")}")
IO.puts("#{Themes.emitter_pool_failure_msg(pid, " Store ID: #{state.store_id}")}")
IO.puts("#{Themes.emitter_pool_failure_msg(pid, " Topic: #{sub_topic}")}")
IO.puts("#{Themes.emitter_pool_failure_msg(pid, " Reason: #{reason}")}")
IO.puts("┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛")
IO.puts("")
IO.puts("")
end
Logger.warning("[EmitterPool:#{state.store_id}] #{message}", metadata)
end
defp handle_action_event(pid, message, metadata, state) do
if state.terminal_output_enabled and state.logging_level in [:debug, :info] do
IO.puts("#{Themes.emitter_pool_action_msg(pid, "[EmitterPool] #{message}")}")
end
Logger.info("[EmitterPool:#{state.store_id}] #{message}", metadata)
end
defp handle_error_event(pid, message, metadata, state) do
if state.terminal_output_enabled do
IO.puts("#{Themes.emitter_pool_failure_msg(pid, "[EmitterPool:ERROR] #{message}")}")
end
Logger.error("[EmitterPool:#{state.store_id}] #{message}", metadata)
end
defp handle_generic_event(event_type, pid, message, metadata, state) do
if state.terminal_output_enabled and state.logging_level == :debug do
IO.puts("#{Themes.emitter_pool_action_msg(pid, "[EmitterPool:#{event_type}] #{message}")}")
end
Logger.debug("[EmitterPool:#{state.store_id}:#{event_type}] #{message}", metadata)
end
end