Antenna is a mixture of Phoenix.PubSub and :gen_event functionality with some batteries included.

It implements back-pressure on top of GenStage, is fully conformant with OTP Design Principles. and is distributed out of the box.

Antenna supports both asynchronous and synchronous events. While the most preferrable way would be to stay fully async with Antenna.event/3, one still might propagate the event synchronously with Antenna.sync_event/3 and collect all the responses from all the handlers.

One can have as many isolated Antennas as necessary, distinguished by Antenna.t:id/0.

Getting Started

To start using Antenna, add it to your dependencies in mix.exs:

def deps do
  [
    {:antenna, "~> 0.3.0"}
  ]
end

Then add it to your supervision tree:

defmodule MyApp.Application do
  use Application

  def start(_type, _args) do
    children = [
      Antenna
    ]

    opts = [strategy: :one_for_one, name: MyApp.Supervisor]
    Supervisor.start_link(children, opts)
  end
end

Configuration

Antenna can be configured in your config.exs:

config :antenna,
  id: MyApp.Antenna,           # Optional custom identifier
  distributed: true,           # Enable distributed mode (default: true)
  sync_timeout: 5_000          # Default timeout for sync_event/4

Sequence Diagram

sequenceDiagram
  Consumer->>+Broadcaster: sync_event(channel, event)
  Consumer->>+Broadcaster: event(channel, event) 
  Broadcaster-->>+Consumer@Node1: event
  Broadcaster-->>+Consumer@Node2: event
  Broadcaster-->>+Consumer@NodeN: event
  Consumer@Node1-->>-NoOp: mine?
  Consumer@NodeN-->>-NoOp: mine?
  Consumer@Node2-->>+Matchers: event
  Matchers-->>+Handlers: handle_match/2
  Matchers-->>+Handlers: handle_match/2
  Handlers->>-Consumer: response(to: sync_event)

ASCII representation.

Common Use Cases

Basic Event Handling

# Define a matcher for specific events
Antenna.match(MyApp.Antenna, {:user_registered, user_id, _meta}, fn channel, event ->
  # Handle the event
  {:user_registered, user_id, meta} = event
  Logger.info("New user registered: #{user_id} on channel #{channel}")
  MyApp.UserNotifier.welcome(user_id)
end, channels: [:user_events])

# Send events
Antenna.event(MyApp.Antenna, [:user_events], {:user_registered, "user123", %{source: "web"}})

Error Handling

# Match on error events
Antenna.match(MyApp.Antenna, {:error, reason, _context} = event, fn channel, event ->
  # Log and handle errors
  Logger.error("Error on channel #{channel}: #{inspect(event)}")
  MyApp.ErrorTracker.track(event)
end, channels: [:errors])

# Send error events
Antenna.event(MyApp.Antenna, [:errors], {:error, :validation_failed, %{user_id: 123}})

Distributed Setup

Antenna automatically handles distribution when nodes are connected:

# On node1@host1
Node.connect(:"node2@host2")
Antenna.match(MyApp.Antenna, {:replicate, data}, fn _, {:replicate, data} ->
  MyApp.Storage.replicate(data)
end, channels: [:replication])

# On node2@host2
Antenna.event(MyApp.Antenna, [:replication], {:replicate, some_data})
# The event will be processed on both nodes

Custom Matchers

You can use pattern matching and guards in your matchers:

# Match on specific patterns with guards
Antenna.match(MyApp.Antenna, 
  {:temperature, celsius, _} when celsius > 30,
  fn _, {:temperature, c, location} ->
    Logger.warning("High temperature (#{c}°C) detected at #{location}")
  end,
  channels: [:sensors])

# Match on map patterns
Antenna.match(MyApp.Antenna,
  %{event: :payment, amount: amount} when amount > 1000,
  fn _, event ->
    MyApp.LargePaymentProcessor.handle(event)
  end,
  channels: [:payments])

Usage Example

The consumer of this library is supposed to declare one or more matchers, subscribing to one or more channels, and then call Antenna.event/2 to propagate the event.

assert {:ok, _pid, "{:tag_1, a, _} when is_nil(a)"} =
  Antenna.match(Antenna, {:tag_1, a, _} when is_nil(a), self(), channels: [:chan_1])

assert :ok = Antenna.event(Antenna, [:chan_1], {:tag_1, nil, 42})
assert_receive {:antenna_event, :chan_1, {:tag_1, nil, 42}}

Architecture

Antenna uses a distributed pub/sub architecture with the following components:

1. Broadcaster - Handles event distribution across nodes
2. Matchers - Pattern match on events and delegate to handlers
3. Guard - Manages channel subscriptions and handler registration
4. PubSub - Implements the core pub/sub functionality

Events flow through the system as follows:

1. Client calls event/3 or sync_event/4
2. Broadcaster distributes the event to all nodes
3. Matchers on each node pattern match against the event
4. Matching handlers process the event
5. For sync_event, responses are collected and returned

Best Practices

1. Channel Organization

   • Use atoms for channel names
   • Group related events under common channels
   • Consider using hierarchical channel names (e.g., :users_created, :users_updated)

2. Pattern Matching

   • Use specific patterns to avoid unnecessary pattern matches
   • Include guards for more precise matching
   • Consider the order of pattern matches when using multiple matchers

3. Handler Design

   • Keep handlers focused and single-purpose
   • Use sync_event/4 when you need responses
   • Consider timeouts for sync operations
   • Handle errors within handlers to prevent cascade failures

4. Performance

   • Use async events (event/3) when possible
   • Keep handler processing time minimal
   • Consider using separate processes for long-running operations
   • Monitor matcher and handler counts

5. Testing

   • Test matchers with various event patterns
   • Verify handler behavior
   • Test distributed scenarios
   • Use ExUnit's async: true when possible