OffBroadway.EMQTT.Connection (off_broadway_emqtt v0.3.0)

Copy Markdown View Source

Manages emqtt connection lifecycle.

Each producer instance gets its own connection with:

  • auto_ack: false for delayed acknowledgements
  • max_inflight for protocol-level backpressure
  • Unique client ID per producer index

Summary

Types

qos()

Functions

disconnect(conn)
get_client_id(config, producer_index)

Returns the client ID the connection will advertise to the broker, given a producer config and producer index. Pure — callers can use this to correlate telemetry or logs without waiting for the connection to be established.

pause(conn)
puback(conn, packet_id)
pubcomp(conn, packet_id)
start_link(config, producer_index)
subscribe(conn, shared_group, topics)

Types

qos()

@type qos() :: 0 | 1 | 2 | :at_most_once | :at_least_once | :exactly_once

Functions

disconnect(conn)

@spec disconnect(pid()) :: :ok

get_client_id(config, producer_index)

@spec get_client_id(
  keyword(),
  non_neg_integer()
) :: String.t()

Returns the client ID the connection will advertise to the broker, given a producer config and producer index. Pure — callers can use this to correlate telemetry or logs without waiting for the connection to be established.

pause(conn)

@spec pause(pid()) :: :ok

puback(conn, packet_id)

@spec puback(pid(), non_neg_integer()) :: :ok

pubcomp(conn, packet_id)

@spec pubcomp(pid(), non_neg_integer()) :: :ok

start_link(config, producer_index)

@spec start_link(
  keyword(),
  non_neg_integer()
) :: {:ok, pid()} | {:error, term()}

subscribe(conn, shared_group, topics)

@spec subscribe(pid(), String.t() | nil, [{String.t(), qos()}]) ::
  {:ok, [{String.t(), 0..2}]} | {:error, term()}