ExMCP.HttpPlug.SSEHandler (ex_mcp v0.9.2)

View Source

Server-Sent Events handler with backpressure control.

This module implements a robust SSE handler that prevents memory leaks through demand-based flow control. It ensures that event producers cannot overwhelm the SSE connection with unbounded message queues.

Features

  • Demand-based backpressure control
  • Event buffering with size limits
  • Last-Event-ID support for resumption
  • Structured error propagation
  • Connection health monitoring
  • Graceful shutdown

Architecture

The handler uses a GenServer that manages the SSE connection lifecycle. Event producers must request permission before sending events, preventing unbounded mailbox growth.

Summary

Types

t()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

close(handler)

Closes the SSE connection gracefully.

request_send(handler)

Requests permission to send an event. This implements backpressure.

send_error(handler, error)

Sends an error event and closes the connection gracefully.

send_event(handler, event_type, data, opts \\ [])

Sends an event after permission has been granted.

start_link(conn, session_id, opts)

Starts an SSE handler for the given connection.

Types

t()

@type t() :: %ExMCP.HttpPlug.SSEHandler{
  conn: Plug.Conn.t(),
  event_buffer: :queue.queue(),
  event_counter: non_neg_integer(),
  heartbeat_ref: reference() | nil,
  last_event_id: String.t() | nil,
  mailbox_monitor: reference() | nil,
  opts: map(),
  producers: MapSet.t(pid()),
  session_id: String.t()
}

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

close(handler)

@spec close(pid()) :: :ok

Closes the SSE connection gracefully.

request_send(handler)

@spec request_send(pid()) :: :ok | {:error, any()}

Requests permission to send an event. This implements backpressure.

The caller will block until the handler is ready to accept more events. Returns :ok when it's safe to send, or {:error, reason} if the connection is closed or errored.

send_error(handler, error)

@spec send_error(pid(), any()) :: :ok

Sends an error event and closes the connection gracefully.

send_event(handler, event_type, data, opts \\ [])

@spec send_event(pid(), String.t(), any(), keyword()) :: :ok

Sends an event after permission has been granted.

This should only be called after request_send/1 returns :ok.

start_link(conn, session_id, opts)

@spec start_link(Plug.Conn.t(), String.t(), map()) :: {:ok, pid()} | {:error, any()}

Starts an SSE handler for the given connection.