Roulette

Scalable PubSub client library which uses HashRing-ed gnatsd-cluster

Installation

If available in Hex, the package can be installed by adding roulette to your list of dependencies in mix.exs:

def deps do
  [
    {:roulette, "~> 1.0.4"}
  ]
end

Documentation can be generated with ExDoc and published on HexDocs. Once published, the docs can be found at https://hexdocs.pm/roulette.

Prepare your own PubSub module

defmodule MyApp.PubSub do
  use Roulette, otp_app: :my_app
end

Configuration

Setup configuration like following

config :my_app, MyApp.PubSub,
    servers: [
      [host: "gnatsd1.example.org", port: 4222],
      [host: "gnatsd2.example.org", port: 4222],
      [host: "gnatsd3.example.org", port: 4222]
    ]
    # ...

Application

Append your PubSub module onto your application's supervisor

defmodule MyApp.Application do
  use Application

  def start(_type, _args) do
    children = [
      {MyApp.Pubsub, []}
      # ... other children
    ]
    opts = [strategy: :one_for_one, name: MyApp.Supervisor]
    Supervisor..start_link(children, opts)
  end

end

Simple Usage

defmodule MyApp.Session do
  use GenServer

  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  def init(opts) do
    username = Keyword.fetch!(opts, :username)
    MyApp.PubSub.sub!(username)
    {:ok, %{username: username}}
  end

  def handle_info({:pubsub_message, topic, msg, pid}, state) do
    # handle msg
    {:noreply, state}
  end

  def terminate(reason, state) do
    :ok
  end

end

Anywhere else you want to publish message in your app.

MyApp.PubSub.pub!("foobar", data)

Premised gnatsd Network Architecture

gnatsd supports cluster-mode. This works with full-mesh and one-hop messaging system to sync events.

gnatsd's full-mesh architecture

Roulette assumes that you put a load-balancer like AWS-NBL in front of each gnatsd-clusters.

Roulette doesn't have a responsiblity for health-check and load-balancing between gnatsd-servers exists in a single gnatsd-cluster. Roulette assumes that It's a load-balancers' responsibility.

Roulette connects to each gnatsd-server through load-balancers, and doesn't mind which endpoint it connects to.

However if your application servers send PUBLISH so much, it'll cause troubles eventuallly.

Roulette resolves this problem with Consistent Hashing.

Setup multiple gnatsd-cluster beforehand, and when your app sends PUBLISH or SUBSCRIBE message, "Which cluster your app sends message to" is decided by the topic.

Full Configuration Description

Here is a minimum configuration example, You must setup servers list. Put your load-balancers' hostname into it.

config :my_app, MyApp.PubSub,
  servers: [
    "gnatsd-cluster1.example.org",
    "gnatsd-cluster2.example.org"
  ]

Or else, you can use keyword list for each host.

config :my_app, MyApp.PubSub,
  servers: [
    [host: "gnatsd-cluster1.example.org", port: 4222],
    [host: "gnatsd-cluster2.example.org", port: 4222]
  ]

If there is no port setting, 4222 is set by defaut.

role

• :both (default) - setup both Publisher and Subscriber connections
• :subscriber - setup Subscriber connections only
• :publisher - setup Publisher connections only

subscription_restart

:temporary

subscription-process sends EXIT message to consumer process when gnatsd-connection is disconnected.

:permanent

subscription-process try to keep subscription. when gnatsd-connection is disconnected, retry to sends SUBSCRIBE message through other connections.

Detailed Usage

Publish

ok/error style.

topic = "foobar"

case MyApp.PubSub.pub(topic, data) do

  :ok -> :ok

  :error -> :error

end

If you don't mind error handling(not recommended on production), you can use pub!/2 instead

topic = "foobar"

MyApp.PubSub.pub!(topic, data)

Subscribe

ok/error style.

sub/1 returns Supervisor.on_start()

topic = "foobar"

case MyApp.PubSub.sub("foobar") do

  {:ok, _pid} -> :ok

  other ->
    Logger.warn "failed to sub: #{inspect other}"
    :error

end

If you don't mind error handling(not recommended on production), you can use sub!/1 instead

MyApp.PubSub.sub!(topic)

Unsubscribe

ok/error style.

sub/1 returns Supervisor.on_start()

topic = "foobar"

case MyApp.PubSub.unsub("foobar") do

  :ok -> :ok

  {:error, :not_found} -> :ok

end

If you don't mind error handling(not recommended on production), you can use unsub!/1 instead

MyApp.PubSub.unsub!(topic)

In following example, you don't need to call unsub/1 on terminate/2. Because unsub is automatically handled, the process which calls sub terminates.

defmodule MyApp.Session do
  use GenServer

  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  def init(opts) do
    username = Keyword.fetch!(opts, :username)
    MyApp.PubSub.sub!(username)
    {:ok, %{username: username}}
  end

  def handle_info({:pubsub_message, topic, msg, pid}, state) do
    # handle msg
    {:noreply, state}
  end

  def terminate(reason, state) do
    # You don't need this line
    # MyApp.PubSub.unsub(state.username)
    :ok
  end

end

LICENSE

MIT-LICENSE

Author

Lyo Kaot <lyo.kato at gmail.com>