Kombu.ConsumerServer behaviour (Kombu v0.1.0)

OTP GenServer behavior and base implementation mirroring Python Kombu's ConsumerMixin and ConsumerProducerMixin.

Provides:

Connection management with automatic reconnection and backoff
Lifecycle hooks (on_connection_error, on_connection_revived, on_consume_ready, on_consume_end)
Integration with Kombu.Consumer and Kombu.Producer
Safe message consumption in a supervised worker loop

Summary

Types

state()

Callbacks

get_consumers(channel, state)

handle_message(body, message, state)

init(args)

on_connection_error(error, interval, state)

on_connection_revived(state)

on_consume_end(state)

on_consume_ready(consumers, state)

on_decode_error(message, error, state)

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

publish(server, body, opts \\ [])

Publishes a message using the server's producer (if producer is enabled).

start_link(mod, args, gen_opts \\ [])

stop(server, reason \\ :normal, timeout \\ 5000)

Stops the consumer server gracefully.

Types

state()

@type state() :: term()

Callbacks

get_consumers(channel, state)

@callback get_consumers(channel :: term(), state()) ::
  {[Kombu.Consumer.t()], state()} | [Kombu.Consumer.t()]

handle_message(body, message, state)

(optional)

@callback handle_message(body :: term(), message :: Kombu.Message.t(), state()) ::
  {:noreply, state()} | {:stop, reason :: term(), state()}

init(args)

@callback init(args :: term()) :: {:ok, state()} | {:stop, reason :: term()}

on_connection_error(error, interval, state)

(optional)

@callback on_connection_error(error :: term(), interval :: float(), state()) :: state()

on_connection_revived(state)

(optional)

@callback on_connection_revived(state()) :: state()

on_consume_end(state)

(optional)

@callback on_consume_end(state()) :: state()

on_consume_ready(consumers, state)

(optional)

@callback on_consume_ready(consumers :: [Kombu.Consumer.t()], state()) :: state()

on_decode_error(message, error, state)

(optional)

@callback on_decode_error(message :: Kombu.Message.t(), error :: term(), state()) ::
  state()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

publish(server, body, opts \\ [])

@spec publish(GenServer.server(), term(), keyword()) :: term()

Publishes a message using the server's producer (if producer is enabled).

start_link(mod, args, gen_opts \\ [])

@spec start_link(module(), term(), keyword()) :: GenServer.on_start()

stop(server, reason \\ :normal, timeout \\ 5000)

@spec stop(GenServer.server(), term(), timeout()) :: :ok

Stops the consumer server gracefully.