Redis.PubSub.Sharded (Redis v0.8.0)

Copy Markdown View Source

Sharded Pub/Sub client for Redis Cluster (Redis 7+).

Uses SSUBSCRIBE / SUNSUBSCRIBE / SPUBLISH to route messages by hash slot. Unlike regular pub/sub which broadcasts to every node, sharded pub/sub only involves the node that owns the channel's hash slot.

Usage

{:ok, sharded} = Redis.PubSub.Sharded.start_link(
  nodes: [{"127.0.0.1", 7000}, {"127.0.0.1", 7001}, {"127.0.0.1", 7002}]
)

:ok = Redis.PubSub.Sharded.ssubscribe(sharded, "orders:123")

receive do
  {:redis_pubsub, :smessage, channel, payload} ->
    IO.puts("sharded message on " <> channel <> ": " <> payload)
end

:ok = Redis.PubSub.Sharded.sunsubscribe(sharded, "orders:123")

Options

  • :nodes - list of seed nodes as {host, port} tuples or "host:port" strings
  • :cluster - pid of an existing Redis.Cluster (used to discover topology)
  • :password - Redis password (applied to all nodes)
  • :timeout - connection timeout ms (default: 5_000)
  • :name - GenServer name registration

Message Format

{:redis_pubsub, :smessage, channel, payload}

Summary

Functions

Returns a specification to start this module under a supervisor.

Subscribe subscriber to a sharded channel.

Stops the sharded pub/sub client.

Returns current subscription state: channels with their subscriber counts.

Unsubscribe subscriber from a sharded channel.

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

ssubscribe(sharded, channel, subscriber \\ self())

@spec ssubscribe(GenServer.server(), String.t(), pid()) :: :ok | {:error, term()}

Subscribe subscriber to a sharded channel.

Messages arrive as {:redis_pubsub, :smessage, channel, payload}.

start_link(opts \\ [])

@spec start_link(keyword()) :: GenServer.on_start()

stop(sharded)

@spec stop(GenServer.server()) :: :ok

Stops the sharded pub/sub client.

subscriptions(sharded)

@spec subscriptions(GenServer.server()) :: map()

Returns current subscription state: channels with their subscriber counts.

sunsubscribe(sharded, channel, subscriber \\ self())

@spec sunsubscribe(GenServer.server(), String.t(), pid()) :: :ok | {:error, term()}

Unsubscribe subscriber from a sharded channel.