Implementing the MQTT 5 specific parts of the protocol

.envrc

@@ -0,0 +1 @@
+eval "$(lorri direnv)"

.formatter.exs

@@ -1,4 +1,5 @@
 # Used by "mix format"
 [
-  inputs: ["mix.exs", "{config,lib,test}/**/*.{ex,exs}"]
+  inputs: ["mix.exs", "{config,lib,test}/**/*.{ex,exs}"],
+  import_deps: [:stream_data]
 ]

.gitignore

@@ -25,3 +25,5 @@ tortoise-*.tar
 # Files generated by Erlang/Elixir QuickCheck
 /current_counterexample.eqc
 /.eqc-info
+
+/todo.org

README.md

@@ -80,7 +80,7 @@ defmodule Tortoise.Handler.Example do
     {:ok, args}
   end
 
-  def connection(status, state) do
+  def status_change(status, state) do
     # `status` will be either `:up` or `:down`; you can use this to
     # inform the rest of your system if the connection is currently
     # open or closed; tortoise should be busy reconnecting if you get

lib/tortoise.ex

@@ -62,7 +62,6 @@ defmodule Tortoise do
 
   alias Tortoise.Package
   alias Tortoise.Connection
-  alias Tortoise.Connection.Inflight
 
   @typedoc """
   An identifier used to identify the client on the server.
@@ -147,6 +146,12 @@ defmodule Tortoise do
   """
   @type topic() :: String.t()
 
+  # todo, documentation
+  @type topic_alias :: 0x0001..0xFFFF
+
+  # todo, documentation
+  @type topic_or_topic_alias() :: topic() | topic_alias()
+
   @typedoc """
   A topic filter for a subscription.
 
@@ -246,35 +251,59 @@ defmodule Tortoise do
   with `Tortoise` so it is easy to see where the message originated
   from.
   """
-  @spec publish(client_id(), topic(), payload, [options]) ::
-          :ok | {:ok, reference()} | {:error, :unknown_connection}
+  @spec publish(Process.dest(), topic_or_topic_alias(), payload, [options]) ::
+          :ok | {:ok, reference()} | {:error, reason}
         when payload: binary() | nil,
              options:
                {:qos, qos()}
                | {:retain, boolean()}
-               | {:identifier, package_identifier()}
-  def publish(client_id, topic, payload \\ nil, opts \\ []) do
+               | {:identifier, package_identifier()},
+             reason: :unknown_connection | :topic_alias_specified_twice
+  def publish(name_or_pid, topic, payload \\ nil, opts \\ [])
+
+  def publish(name_or_pid, topic, payload, opts) when is_binary(topic) do
+    {opts, properties} = Keyword.split(opts, [:retain, :qos, :transforms])
     qos = Keyword.get(opts, :qos, 0)
 
     publish = %Package.Publish{
       topic: topic,
       qos: qos,
       payload: payload,
-      retain: Keyword.get(opts, :retain, false)
+      retain: Keyword.get(opts, :retain, false),
+      properties: properties
     }
 
-    with {:ok, {transport, socket}} <- Connection.connection(client_id) do
-      case publish do
-        %Package.Publish{qos: 0} ->
+    case publish do
+      %Package.Publish{qos: 0} ->
+        with {:ok, {transport, socket}} <- Connection.connection(name_or_pid) do
           encoded_publish = Package.encode(publish)
           apply(transport, :send, [socket, encoded_publish])
+        else
+          {:error, :unknown_connection} ->
+            {:error, :unknown_connection}
+        end
+
+      %Package.Publish{qos: qos} when qos in [1, 2] ->
+        # transforms = Keyword.get(opts, :transforms, {[], nil})
+        Connection.publish(name_or_pid, publish)
+    end
+  end
 
-        %Package.Publish{qos: qos} when qos in [1, 2] ->
-          Inflight.track(client_id, {:outgoing, publish})
-      end
-    else
-      {:error, :unknown_connection} ->
-        {:error, :unknown_connection}
+  # Support passing in a topic alias instead of a proper topic, in
+  # this case we will lift the topic alias into a topic_alias value in
+  # the property list, set the topic to an empty string and pass it on
+  # to the regular `publish/4`
+  def publish(client_id, topic_alias, payload, opts)
+      when is_integer(topic_alias) and topic_alias > 0 do
+    case Keyword.get(opts, :topic_alias) do
+      nil ->
+        publish(client_id, "", payload, Keyword.put(opts, :topic_alias, topic_alias))
+
+      ^topic_alias ->
+        publish(client_id, "", payload, opts)
+
+      _otherwise ->
+        {:error, :topic_alias_specified_twice}
     end
   end
 
@@ -304,37 +333,61 @@ defmodule Tortoise do
 
   See the documentation for `Tortoise.publish/4` for configuration.
   """
-  @spec publish_sync(client_id(), topic(), payload, [options]) ::
-          :ok | {:error, :unknown_connection}
+  @spec publish_sync(Process.dest(), topic_or_topic_alias(), payload, [options]) ::
+          :ok | {:error, reason}
         when payload: binary() | nil,
              options:
                {:qos, qos()}
                | {:retain, boolean()}
                | {:identifier, package_identifier()}
-               | {:timeout, timeout()}
-  def publish_sync(client_id, topic, payload \\ nil, opts \\ []) do
-    timeout = Keyword.get(opts, :timeout, :infinity)
+               | {:timeout, timeout()},
+             reason: :unknown_connection | :topic_alias_specified_twice
+  def publish_sync(pid_or_name, topic, payload \\ nil, opts \\ [])
+
+  def publish_sync(pid_or_name, topic, payload, opts) when is_binary(topic) do
+    {opts, properties} = Keyword.split(opts, [:retain, :qos, :transforms, :timeout])
     qos = Keyword.get(opts, :qos, 0)
 
     publish = %Package.Publish{
       topic: topic,
       qos: qos,
       payload: payload,
-      retain: Keyword.get(opts, :retain, false)
+      retain: Keyword.get(opts, :retain, false),
+      properties: properties
     }
 
-    with {:ok, {transport, socket}} <- Connection.connection(client_id) do
-      case publish do
-        %Package.Publish{qos: 0} ->
-          encoded_publish = Package.encode(publish)
-          apply(transport, :send, [socket, encoded_publish])
+    case publish do
+      # %Package.Publish{qos: 0} ->
+      # with {:ok, {transport, socket}} <- Connection.connection(client_id) do
+      #   encoded_publish = Package.encode(publish)
+      #   apply(transport, :send, [socket, encoded_publish])
+      # else
+      #   {:error, :unknown_connection} ->
+      #     {:error, :unknown_connection}
+      # end
+
+      %Package.Publish{qos: qos} when qos in [1, 2] ->
+        # transforms = Keyword.get(opts, :transforms, {[], nil})
+        timeout = Keyword.get(opts, :timeout, :infinity)
+        Tortoise.Connection.publish_sync(pid_or_name, publish, timeout)
+    end
+  end
 
-        %Package.Publish{qos: qos} when qos in [1, 2] ->
-          Inflight.track_sync(client_id, {:outgoing, publish}, timeout)
-      end
-    else
-      {:error, :unknown_connection} ->
-        {:error, :unknown_connection}
+  # Support passing in a topic alias instead of a proper topic, in
+  # this case we will lift the topic alias into a topic_alias value in
+  # the property list, set the topic to an empty string and pass it on
+  # to the regular `publish_sync/4`
+  def publish_sync(client_id, topic_alias, payload, opts)
+      when is_integer(topic_alias) and topic_alias > 0 do
+    case Keyword.get(opts, :topic_alias) do
+      nil ->
+        publish(client_id, "", payload, Keyword.put(opts, :topic_alias, topic_alias))
+
+      ^topic_alias ->
+        publish(client_id, "", payload, opts)
+
+      _otherwise ->
+        {:error, :topic_alias_specified_twice}
     end
   end
 end

lib/tortoise/app.ex → lib/tortoise/application.ex

@@ -1,4 +1,4 @@
-defmodule Tortoise.App do
+defmodule Tortoise.Application do
   @moduledoc false
 
   use Application
@@ -10,7 +10,8 @@ defmodule Tortoise.App do
 
     children = [
       {Registry, [keys: :unique, name: Tortoise.Registry]},
-      {Registry, [keys: :duplicate, name: Tortoise.Events]},
+      {Tortoise.TransmitterSupervisor, []},
+      {Tortoise.Session, [backend: Tortoise.Session.Ets]},
       {Tortoise.Supervisor, [strategy: :one_for_one]}
     ]