readme.md

Example!

This example starts four tortoise clients, :a, :b, :c, and :d who subscribe to the same shared topic: bar. Notice that the client expect a broker running on localhost, on port 1883, and accept username/password less connections.

Start the example by typing MqttTest.start() in an iex session; now one can observe the name of the receiving client change when a message is posted to the bar-topic, using: Tortoise.publish(:a, "bar", "hello")

handler.ex

defmodule MqttTest.Handler do
  @moduledoc false

  require Logger

  defstruct [:name]
  alias __MODULE__, as: State

  @behaviour Tortoise.Handler

  @impl true
  def init(opts) do
    name = Keyword.get(opts, :name)
    {:ok, %State{name: name}}
  end

  @impl true
  def connection(:up, state) do
    Logger.info("Connection has been established")
    {:ok, state}
  end

  def connection(:down, state) do
    Logger.warn("Connection has been dropped")
    {:ok, state}
  end

  @impl true
  def subscription(:up, topic, state) do
    Logger.info("Subscribed to #{topic}")
    {:ok, state}
  end

  def subscription({:warn, [requested: req, accepted: qos]}, topic, state) do
    Logger.warn("Subscribed to #{topic}; requested #{req} but got accepted with QoS #{qos}")
    {:ok, state}
  end

  def subscription({:error, reason}, topic, state) do
    Logger.error("Error subscribing to #{topic}; #{inspect(reason)}")
    {:ok, state}
  end

  def subscription(:down, topic, state) do
    Logger.info("Unsubscribed from #{topic}")
    {:ok, state}
  end

  @impl true
  def handle_message(topic, publish, state) do
    Logger.info("#{state.name}: #{Enum.join(topic, "/")} #{inspect(publish)}")
    {:ok, state}
  end

  @impl true
  def terminate(reason, _state) do
    Logger.warn("Client has been terminated with reason: #{inspect(reason)}")
    :ok
  end

end

iex session

/tmp/mqtt_test » iex -S mix
Erlang/OTP 21 [erts-10.0.4] [source] [64-bit] [smp:8:8] [ds:8:8:10] [async-threads:1] [hipe] [dtrace]

=INFO REPORT==== 3-Aug-2018::16:18:26.661585 ===
    application: logger
    exited: stopped
    type: temporary
Interactive Elixir (1.6.5) - press Ctrl+C to exit (type h() ENTER for help)
iex(1)> MqttTest.start()
{:ok, #PID<0.183.0>}
iex(2)>
16:18:36.849 [info]  Connection has been established

16:18:36.849 [info]  Connection has been established

16:18:36.849 [info]  Connection has been established

16:18:36.849 [info]  Connection has been established

16:18:36.863 [info]  Subscribed to $share/share-group/bar

16:18:36.863 [info]  Subscribed to $share/share-group/bar

16:18:36.863 [info]  Subscribed to $share/share-group/bar

16:18:36.863 [info]  Subscribed to $share/share-group/bar
iex(2)> Tortoise.publish(:a, "bar", "hi")
:ok
iex(3)>
16:18:54.260 [info]  b: bar "hi"
iex(3)> Tortoise.publish(:a, "bar", "hi")
:ok
iex(4)>
16:18:55.439 [info]  a: bar "hi"
iex(4)> Tortoise.publish(:a, "bar", "hi")
:ok
iex(5)>
16:18:56.391 [info]  c: bar "hi"
iex(5)> Tortoise.publish(:a, "bar", "hi")
:ok
iex(6)>
16:18:57.431 [info]  a: bar "hi"
iex(6)> Tortoise.publish(:a, "bar", "hi")
:ok
iex(7)>
16:18:58.575 [info]  a: bar "hi"

mqtt_test.ex

defmodule MqttTest do
  def start do
    {:ok, _} =
      Tortoise.Supervisor.start_child(
        client_id: :a,
        handler: {MqttTest.Handler, [name: :a]},
        server: {Tortoise.Transport.Tcp, host: 'localhost', port: 1883},
        subscriptions: [{"$share/share-group/bar", 0}])

    {:ok, _} =
      Tortoise.Supervisor.start_child(
        client_id: :b,
        handler: {MqttTest.Handler, [name: :b]},
        server: {Tortoise.Transport.Tcp, host: 'localhost', port: 1883},
        subscriptions: [{"$share/share-group/bar", 0}])

    {:ok, _} =
      Tortoise.Supervisor.start_child(
        client_id: :c,
        handler: {MqttTest.Handler, [name: :c]},
        server: {Tortoise.Transport.Tcp, host: 'localhost', port: 1883},
        subscriptions: [{"$share/share-group/bar", 0}])

    {:ok, _} =
      Tortoise.Supervisor.start_child(
        client_id: :d,
        handler: {MqttTest.Handler, [name: :d]},
        server: {Tortoise.Transport.Tcp, host: 'localhost', port: 1883},
        subscriptions: [{"$share/share-group/bar", 0}])
  end
end