Add server (support publish only) (#1)

* Add basic server

* Add server handler behaviour

* Build messages timestamp

* Refactor message completion check

* Add amf0 tests

* test chunk stream

Author: gBillal

lib/ex_rtmp/amf0.ex

@@ -0,0 +1,79 @@
+defmodule ExRTMP.AMF0 do
+  @moduledoc """
+  AMF0 parser.
+  """
+
+  @spec parse(binary()) :: list()
+  def parse(data), do: do_parse(data, [])
+
+  @spec serialize(any()) :: iodata()
+  def serialize(true), do: <<0x01, 1>>
+  def serialize(false), do: <<0x01, 0>>
+  def serialize(nil), do: <<0x05>>
+  def serialize(value) when is_number(value), do: <<0x00::8, value::float-64>>
+  def serialize(atom) when is_atom(atom), do: serialize(to_string(atom))
+  def serialize(str) when is_binary(str), do: <<0x02::8, byte_size(str)::16, str::binary>>
+
+  def serialize(map) when is_map(map) do
+    pairs =
+      Enum.map(map, fn {key, value} ->
+        [<<byte_size(key)::16, key::binary>>, serialize(value)]
+      end)
+
+    [0x03, pairs, 0, 0, 9]
+  end
+
+  defp do_parse(<<>>, acc), do: Enum.reverse(acc)
+
+  defp do_parse(data, acc) do
+    case parse_value(data) do
+      {:object_start, rest} ->
+        {obj, rest} = parse_object(rest, %{})
+        do_parse(rest, [obj | acc])
+
+      {value, rest} ->
+        do_parse(rest, [value | acc])
+    end
+  end
+
+  defp parse_value(<<0x00::8, number::float-64, rest::binary>>), do: {number, rest}
+  defp parse_value(<<0x01::8, boolean::8, rest::binary>>), do: {boolean == 1, rest}
+  defp parse_value(<<0x03::8, rest::binary>>), do: {:object_start, rest}
+  defp parse_value(<<0x05::8, rest::binary>>), do: {nil, rest}
+
+  defp parse_value(<<0x02::8, str_len::16, str::binary-size(str_len), rest::binary>>),
+    do: {str, rest}
+
+  defp parse_value(<<0x08::8, count::32, rest::binary>>) do
+    {list, rest} =
+      Enum.map_reduce(1..count, rest, fn _idx, rest ->
+        {key, rest} = parse_object_key(rest)
+        {value, rest} = parse_value(rest)
+        {{key, value}, rest}
+      end)
+
+    <<0, 0, 9, rest::binary>> = rest
+    {list, rest}
+  end
+
+  defp parse_value(<<0x0A::8, count::32, rest::binary>>) do
+    Enum.map_reduce(1..count, rest, fn _idx, rest ->
+      parse_value(rest)
+    end)
+  end
+
+  defp parse_object(rest, obj) do
+    case parse_object_key(rest) do
+      {:object_end, rest} ->
+        {obj, rest}
+
+      {key, rest} ->
+        {value, rest} = parse_value(rest)
+        parse_object(rest, Map.put(obj, key, value))
+    end
+  end
+
+  defp parse_object_key(<<0::16, 0x09, rest::binary>>), do: {:object_end, rest}
+  defp parse_object_key(<<str_len::16, str::binary-size(str_len), rest::binary>>), do: {str, rest}
+  defp parse_object_key(_data), do: :error
+end

lib/ex_rtmp/chunk.ex

@@ -0,0 +1,112 @@
+defmodule ExRTMP.Chunk do
+  @moduledoc """
+  RTMP Chunk structure
+  """
+
+  @type payload_size :: non_neg_integer()
+  @type stream_id :: non_neg_integer()
+
+  @type t :: %__MODULE__{
+          fmt: 0..3,
+          stream_id: stream_id(),
+          timestamp: non_neg_integer() | nil,
+          message_length: non_neg_integer() | nil,
+          message_type_id: non_neg_integer() | nil,
+          message_stream_id: non_neg_integer() | nil,
+          payload: binary() | nil
+        }
+
+  defstruct [
+    :fmt,
+    :stream_id,
+    :timestamp,
+    :message_length,
+    :message_type_id,
+    :message_stream_id,
+    :payload
+  ]
+
+  @spec parse_header(binary()) :: {:ok, t(), binary()} | :more
+  def parse_header(data) do
+    with {:ok, chunk, rest} <- parse_stream_id(data),
+         {:ok, chunk, rest} <- parse_message_header(chunk, rest),
+         {:ok, chunk, rest} <- parse_extended_timestamp(chunk, rest) do
+      {:ok, chunk, rest}
+    end
+  end
+
+  @spec serialize(t()) :: iodata()
+  def serialize(%__MODULE__{} = chunk) do
+    timestamp = if chunk.timestamp, do: <<chunk.timestamp::24>>, else: <<>>
+    length = if chunk.message_length, do: <<chunk.message_length::24>>, else: <<>>
+    type_id = if chunk.message_type_id, do: <<chunk.message_type_id::8>>, else: <<>>
+
+    msg_stream_id =
+      if chunk.message_stream_id, do: <<chunk.message_stream_id::32-little>>, else: <<>>
+
+    [
+      <<chunk.fmt::2, encode_stream_id(chunk.stream_id)::bitstring>>,
+      timestamp,
+      length,
+      type_id,
+      msg_stream_id,
+      chunk.payload
+    ]
+  end
+
+  defp parse_stream_id(data) do
+    case data do
+      <<fmt::2, 0::6, cs_id::8, rest::binary>> ->
+        {:ok, %__MODULE__{fmt: fmt, stream_id: cs_id + 64}, rest}
+
+      <<fmt::2, 1::6, cs_id::16, rest::binary>> ->
+        {:ok, %__MODULE__{fmt: fmt, stream_id: cs_id + 64}, rest}
+
+      <<fmt::2, cs_id::6, rest::binary>> ->
+        {:ok, %__MODULE__{fmt: fmt, stream_id: cs_id}, rest}
+
+      _ ->
+        :more
+    end
+  end
+
+  defp parse_message_header(chunk, data) do
+    case {chunk.fmt, data} do
+      {0,
+       <<timestamp::24, length::24, type_id::8, msg_stream_id::integer-32-little, rest::binary>>} ->
+        chunk = %{
+          chunk
+          | timestamp: timestamp,
+            message_length: length,
+            message_type_id: type_id,
+            message_stream_id: msg_stream_id
+        }
+
+        {:ok, chunk, rest}
+
+      {1, <<timestamp::24, length::24, type_id::8, rest::binary>>} ->
+        chunk = %{chunk | timestamp: timestamp, message_length: length, message_type_id: type_id}
+        {:ok, chunk, rest}
+
+      {2, <<timestamp::24, rest::binary>>} ->
+        {:ok, %{chunk | timestamp: timestamp}, rest}
+
+      {3, rest} ->
+        {:ok, chunk, rest}
+
+      _ ->
+        :more
+    end
+  end
+
+  defp parse_extended_timestamp(%{timestamp: 0xFFFFFF} = chunk, <<ts::32, rest::binary>>) do
+    {:ok, %{chunk | timestamp: ts}, rest}
+  end
+
+  defp parse_extended_timestamp(%{timestamp: 0xFFFFFF}, _rest), do: :more
+  defp parse_extended_timestamp(chunk, data), do: {:ok, chunk, data}
+
+  defp encode_stream_id(stream_id) when stream_id in 2..63, do: <<stream_id::6>>
+  defp encode_stream_id(stream_id) when stream_id in 64..319, do: <<0::6, stream_id - 64::8>>
+  defp encode_stream_id(stream_id) when stream_id >= 320, do: <<1::6, stream_id - 64::16>>
+end

lib/ex_rtmp/chunk_parser.ex

@@ -0,0 +1,99 @@
+defmodule ExRTMP.ChunkParser do
+  @moduledoc false
+
+  alias ExRTMP.{Chunk, Message}
+
+  @type t :: %__MODULE__{
+          unprocessed_data: binary(),
+          messages: %{Chunk.stream_id() => Message.t()},
+          chunk_size: non_neg_integer(),
+          message_first_chunk: %{Chunk.stream_id() => tuple()}
+        }
+
+  defstruct [:unprocessed_data, :messages, :message_first_chunk, chunk_size: 128]
+
+  @spec new() :: t()
+  def new() do
+    %__MODULE__{
+      unprocessed_data: <<>>,
+      messages: %{},
+      message_first_chunk: %{}
+    }
+  end
+
+  @spec process(binary(), t()) :: {[Message.t()], t()}
+  def process(data, parser) do
+    do_process(parser.unprocessed_data <> data, parser)
+  end
+
+  defp do_process(data, parser, acc \\ []) do
+    with {:ok, chunk, rest} <- Chunk.parse_header(data),
+         {chunk, parser} <- set_missing_fields(chunk, parser),
+         payload_size <- get_chunk_payload_size(parser, chunk),
+         <<payload::binary-size(payload_size), rest::binary>> <- rest do
+      message = Map.get(parser.messages, chunk.stream_id, Message.new(chunk))
+
+      {parser, acc} =
+        case Message.append(message, payload) do
+          {:ok, message} ->
+            parser = %{parser | messages: Map.delete(parser.messages, chunk.stream_id)}
+            {parser, [message | acc]}
+
+          {:more, message} ->
+            parser = %{parser | messages: Map.put(parser.messages, chunk.stream_id, message)}
+            {parser, acc}
+        end
+
+      do_process(rest, parser, acc)
+    else
+      _ ->
+        {Enum.reverse(acc), %{parser | unprocessed_data: data}}
+    end
+  end
+
+  defp set_missing_fields(chunk, parser) when is_map_key(parser.messages, chunk.stream_id) do
+    {chunk, parser}
+  end
+
+  defp set_missing_fields(%{fmt: 0} = chunk, parser) do
+    tuple =
+      {chunk.timestamp, 0, chunk.message_length, chunk.message_type_id, chunk.message_stream_id}
+
+    message_first_chunk = Map.put(parser.message_first_chunk, chunk.stream_id, tuple)
+    {chunk, %{parser | message_first_chunk: message_first_chunk}}
+  end
+
+  defp set_missing_fields(chunk, parser) do
+    {timestamp, timestamp_delta, message_size, message_type, message_stream_id} =
+      Map.fetch!(parser.message_first_chunk, chunk.stream_id)
+
+    timestamp_delta = chunk.timestamp || timestamp_delta
+
+    chunk = %{
+      chunk
+      | timestamp: timestamp_delta + timestamp,
+        message_length: chunk.message_length || message_size,
+        message_type_id: chunk.message_type_id || message_type,
+        message_stream_id: chunk.message_stream_id || message_stream_id
+    }
+
+    message_first_chunk =
+      Map.put(
+        parser.message_first_chunk,
+        chunk.stream_id,
+        {chunk.timestamp, timestamp_delta, chunk.message_length, chunk.message_type_id,
+         chunk.message_stream_id}
+      )
+
+    {chunk, %{parser | message_first_chunk: message_first_chunk}}
+  end
+
+  defp get_chunk_payload_size(parser, chunk) do
+    remaining_bytes =
+      if message = Map.get(parser.messages, chunk.stream_id),
+        do: message.size - message.current_size,
+        else: chunk.message_length
+
+    min(parser.chunk_size, remaining_bytes)
+  end
+end