Roulette

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

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.

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

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]
    ]
    # ...

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

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)

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.

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.

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

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

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

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)

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)

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

MIT-LICENSE

Lyo Kaot <lyo.kato at gmail.com>