Documentation needed for MQTT.Server

[mynameisrufus]

trafficstars

I'm having trouble running the server. I'm running mix test with the following test in it:

defmodule MyMqttServerTest do
  use ExUnit.Case
  alias MQTT.Client

  test "greets the world" do
    {:ok, connection, false} = MQTT.Client.connect(%{
        transport: {:tcp, %{host: "localhost", port: 1883}},
        client_id: UUID.uuid1()
    })

    topic = "/hello"
    message = "hello"

    {:ok, [{^topic, 0}]} = Client.subscribe(connection, [{topic, 0}])
    :ok = Client.publish(connection, topic, message)

    assert_receive {:mqtt_client, ^connection, {:publish, ^topic, ^message, _}}

    :ok = Client.disconnect(connection)
  end
end

Child process:

%{
  id: MyMqttServer,
  type: :worker,
  start: {MyMqttServer, :start_link, [[:hello]]}
}

defmodule MyMqttServer do
  use MQTT.Server

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

  def init(stack) do
    {:ok, stack}
  end

  def init(stack, _) do
    {:ok, stack}
  end
  
  def handle_subscribe(_topic, _qos, state) do
    IO.inspect("I never get output")
    {:ok, state}
  end
end

It seems to be running but I can't seem to override the fun? I'm still relatively new to this so apologies if this is a simple question.

[asabil]

Hi @mynameisrufus, this library is useful if you want low level MQTT support, and thus requires a bit more setup in order to create a server using it. Among other things, you need to setup your listening socket yourself.

Here is some (untested) example code using ranch as the socket listener.

defmodule MyMqttServer.Application do
  @moduledoc false

  use Application

  def start(_type, _args) do
    opts = [
      strategy: :one_for_one,
      name: MyMqttServer.Supervisor
    ]

    children = [
      ranch_listeners(ip: {127, 0, 0, 1},  port: 8883)
    ]

    Supervisor.start_link(children, opts)
  end

  defp ranch_listeners(config) do
    :ranch.child_spec(:my_mqtt_server, 10, :ranch_tcp, options, MyMqttServer.Connection, [])
  end
end

defmodule MyMqttServer.Connection do
  use MQTT.Server

  def start_link(ref, ranch_socket, ranch_transport, options) do
    :proc_lib.start_link(__MODULE__, :init, [ref, ranch_socket, ranch_transport, options])
  end

  def stop(pid) do
    MQTT.Server.stop(pid, :normal, 500)
  end

  def init(ref, ranch_socket, ranch_transport, _options) do
    :ok = :proc_lib.init_ack({:ok, self()})
    :ok = :ranch.accept_ack(ref)
    transport = MQTT.Transport.new(ranch_transport.name(), ranch_socket)
    MQTT.Server.enter_loop(__MODULE__, [], transport)
  end

   def init(_args, %{protocol: "MY-PROTOCOL/1.0"}) do
    {:ok, :my_state}
  end
  def init(_args, _) do
    {:stop, :unacceptable_protocol}
  end
  
  def handle_publish(topic, message, _opts, s) do
    IO.inspect("publication requested")
    {:ok, state}
  end

  def handle_subscribe(topic, _qos, s) do
    IO.inspect("subscription requested")
    {:ok, :failed, s} # Reject all subscriptions
  end

  def handle_unsubscribe(topic, s) do
    IO.inspect("unsubscription requested")
    {:ok, s}
  end

  def handle_info(_info, s) do
    {:ok, s}
  end

  def terminate(_reason, _) do
    :ok
  end
end

Please note that this is a Low level library, it is useful if you want to use the mqtt wire protocol without necessarily implementing the entire protocol. Also, as of now QoS other than 0 are not supported.

Hope this helps.

[mynameisrufus]

Thanks @asabil,

I will see if I can write some tests and or documentation for this and make a PR. I love the approach of the low level lib, just needs some more doco, I will see what I can do.