Fix handle closed connection not getting the right state (#57)

* Fix handle closed connection not getting the right state

* Bump version

* Add catchall mapping for response status

* Fix credo warning

* Fix code reviews

* Fix code reviews

Author: gBillal

README.md

@@ -18,7 +18,7 @@ of dependencies in `mix.exs`:
 ```elixir
 def deps do
   [
-    {:membrane_rtsp, "~> 0.11.0"}
+    {:membrane_rtsp, "~> 0.12.0"}
   ]
 end
 ```

lib/membrane_rtsp/request.ex

@@ -165,9 +165,8 @@ defmodule Membrane.RTSP.Request do
   algorithm
   """
   @spec process_uri(t(), URI.t()) :: binary()
-  def process_uri(request, uri) do
-    %URI{uri | userinfo: nil}
-    |> apply_path(request)
+  def process_uri(request, %URI{} = uri) do
+    apply_path(%URI{uri | userinfo: nil}, request)
   end
 
   defp apply_path(%URI{} = base_url, %__MODULE__{path: nil}), do: base_url

lib/membrane_rtsp/response.ex

@@ -59,7 +59,7 @@ defmodule Membrane.RTSP.Response do
   def parse(response) do
     [headers, body] = String.split(response, ["\r\n\r\n", "\n\n", "\r\r"], parts: 2)
 
-    with {:ok, {response, headers}} <- parse_start_line(headers),
+    with {:ok, {%__MODULE__{} = response, headers}} <- parse_start_line(headers),
          {:ok, headers} <- parse_headers(headers),
          {:ok, body} <- parse_body(body, headers) do
       {:ok, %__MODULE__{response | headers: headers, body: body}}
@@ -110,7 +110,7 @@ defmodule Membrane.RTSP.Response do
     headers = Enum.at(split_response, 0)
     body = Enum.at(split_response, 1)
 
-    with {:ok, {response, headers}} <- parse_start_line(headers),
+    with {:ok, {%__MODULE__{} = response, headers}} <- parse_start_line(headers),
          {:ok, headers} <- parse_headers(headers),
          false <- is_nil(body),
          body_size <- byte_size(body),
@@ -239,6 +239,8 @@ defmodule Membrane.RTSP.Response do
   defp render_status(404), do: "Not Found"
   defp render_status(405), do: "Method Not Allowed"
   defp render_status(455), do: "Method Not Valid In This State"
+  defp render_status(461), do: "Unsupported Transport"
   defp render_status(500), do: "Internal Server Error"
   defp render_status(501), do: "Not Implemented"
+  defp render_status(_status), do: "Generic Error"
 end

lib/membrane_rtsp/rtsp.ex

@@ -75,7 +75,7 @@ defmodule Membrane.RTSP do
 
   @doc """
   Transfer the control of the TCP socket the session was using to a new process. For more information see `:gen_tcp.controlling_process/2`.
-  From now on the session won't try to receive responses to requests from the socket, since now an other process is controlling it. 
+  From now on the session won't try to receive responses to requests from the socket, since now an other process is controlling it.
   Instead of this, the session will synchronously wait for the response to be supplied with `handle_response/2`.
   """
   @spec transfer_socket_control(t(), pid()) ::
@@ -267,7 +267,7 @@ defmodule Membrane.RTSP do
   defp parse_response(raw_response, state) do
     with {:ok, parsed_response} <- Response.parse(raw_response),
          {:ok, state} <- handle_session_id(parsed_response, state),
-         {:ok, state} <- detect_authentication_type(parsed_response, state) do
+         {:ok, %State{} = state} <- detect_authentication_type(parsed_response, state) do
       state = %State{state | cseq: state.cseq + 1}
       {:ok, parsed_response, state}
     end
@@ -338,7 +338,7 @@ defmodule Membrane.RTSP do
   # Some responses do not have to return the Session ID
   # If it does return one, it needs to match one stored in the state.
   @spec handle_session_id(Response.t(), State.t()) :: {:ok, State.t()} | {:error, reason :: any()}
-  defp handle_session_id(%Response{} = response, state) do
+  defp handle_session_id(%Response{} = response, %State{} = state) do
     with {:ok, session_value} <- Response.get_header(response, "Session") do
       [session_id | _rest] = String.split(session_value, ";")
 

lib/membrane_rtsp/server.ex

@@ -124,6 +124,9 @@ defmodule Membrane.RTSP.Server do
         reuseaddr: true
       ])
 
+    {:ok, port} = :inet.port(socket)
+    Logger.info("RTSP server listening on port #{port}")
+
     udp_rtp_port = config[:udp_rtp_port]
     udp_rtcp_port = config[:udp_rtcp_port]
 

lib/membrane_rtsp/server/conn.ex

@@ -33,7 +33,7 @@ defmodule Membrane.RTSP.Server.Conn do
       %Logic.State{recording_with_tcp?: true} = state ->
         {:noreply, state}
 
-      _other ->
+      {:error, _reason, state} ->
         state.request_handler.handle_closed_connection(state.request_handler_state)
         {:stop, :normal, state}
     end
@@ -56,17 +56,21 @@ defmodule Membrane.RTSP.Server.Conn do
   end
 
   defp do_process_client_requests(state, timeout) do
-    with {:ok, request} <- get_request(state.socket, timeout) do
-      case Logic.process_request(request, state) do
-        %Logic.State{recording_with_tcp?: true} = state ->
-          state
+    case get_request(state.socket, timeout) do
+      {:ok, request} ->
+        case Logic.process_request(request, state) do
+          %Logic.State{recording_with_tcp?: true} = state ->
+            state
 
-        %Logic.State{session_state: :recording} = state ->
-          do_process_client_requests(state, :infinity)
+          %Logic.State{session_state: :recording} = state ->
+            do_process_client_requests(state, :infinity)
 
-        state ->
-          do_process_client_requests(state, state.session_timeout)
-      end
+          state ->
+            do_process_client_requests(state, state.session_timeout)
+        end
+
+      {:error, reason} ->
+        {:error, reason, state}
     end
   end
 
@@ -90,7 +94,7 @@ defmodule Membrane.RTSP.Server.Conn do
 
   defp do_parse_request([raw_request, body]) do
     case Request.parse(raw_request <> "\r\n\r\n") do
-      {:ok, request} ->
+      {:ok, %Request{} = request} ->
         content_length =
           case Request.get_header(request, "Content-Length") do
             {:ok, value} -> String.to_integer(value)

lib/membrane_rtsp/server/handler.ex

@@ -201,13 +201,27 @@ defmodule Membrane.RTSP.Server.Handler do
       @impl true
       def init(config), do: config
 
+      @impl true
+      def handle_open_connection(_conn, state), do: state
+
       @impl true
       def handle_announce(_req, state), do: {Response.new(501), state}
 
       @impl true
       def handle_record(_req, state), do: {Response.new(501), state}
 
-      defoverridable init: 1, handle_announce: 2, handle_record: 2
+      @impl true
+      def handle_pause(state), do: {Response.new(501), state}
+
+      @impl true
+      def handle_closed_connection(_state), do: :ok
+
+      defoverridable init: 1,
+                     handle_open_connection: 2,
+                     handle_announce: 2,
+                     handle_record: 2,
+                     handle_pause: 1,
+                     handle_closed_connection: 1
     end
   end
 end

mix.exs

@@ -1,7 +1,7 @@
 defmodule Membrane.RTSP.MixProject do
   use Mix.Project
 
-  @version "0.11.0"
+  @version "0.12.0"
   @github_url "https://github.com/membraneframework/membrane_rtsp"
 
   def project do

mix.lock

@@ -1,18 +1,18 @@
 %{
   "bunch": {:hex, :bunch, "1.6.1", "5393d827a64d5f846092703441ea50e65bc09f37fd8e320878f13e63d410aec7", [:mix], [], "hexpm", "286cc3add551628b30605efbe2fca4e38cc1bea89bcd0a1a7226920b3364fe4a"},
   "bunt": {:hex, :bunt, "1.0.0", "081c2c665f086849e6d57900292b3a161727ab40431219529f13c4ddcf3e7a44", [:mix], [], "hexpm", "dc5f86aa08a5f6fa6b8096f0735c4e76d54ae5c9fa2c143e5a1fc7c1cd9bb6b5"},
-  "credo": {:hex, :credo, "1.7.13", "126a0697df6b7b71cd18c81bc92335297839a806b6f62b61d417500d1070ff4e", [:mix], [{:bunt, "~> 0.2.1 or ~> 1.0", [hex: :bunt, repo: "hexpm", optional: false]}, {:file_system, "~> 0.2 or ~> 1.0", [hex: :file_system, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "47641e6d2bbff1e241e87695b29f617f1a8f912adea34296fb10ecc3d7e9e84f"},
+  "credo": {:hex, :credo, "1.7.15", "283da72eeb2fd3ccf7248f4941a0527efb97afa224bcdef30b4b580bc8258e1c", [:mix], [{:bunt, "~> 0.2.1 or ~> 1.0", [hex: :bunt, repo: "hexpm", optional: false]}, {:file_system, "~> 0.2 or ~> 1.0", [hex: :file_system, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "291e8645ea3fea7481829f1e1eb0881b8395db212821338e577a90bf225c5607"},
   "dialyxir": {:hex, :dialyxir, "1.4.7", "dda948fcee52962e4b6c5b4b16b2d8fa7d50d8645bbae8b8685c3f9ecb7f5f4d", [:mix], [{:erlex, ">= 0.2.8", [hex: :erlex, repo: "hexpm", optional: false]}], "hexpm", "b34527202e6eb8cee198efec110996c25c5898f43a4094df157f8d28f27d9efe"},
   "earmark_parser": {:hex, :earmark_parser, "1.4.44", "f20830dd6b5c77afe2b063777ddbbff09f9759396500cdbe7523efd58d7a339c", [:mix], [], "hexpm", "4778ac752b4701a5599215f7030989c989ffdc4f6df457c5f36938cc2d2a2750"},
   "elixir_uuid": {:hex, :elixir_uuid, "1.2.1", "dce506597acb7e6b0daeaff52ff6a9043f5919a4c3315abb4143f0b00378c097", [:mix], [], "hexpm", "f7eba2ea6c3555cea09706492716b0d87397b88946e6380898c2889d68585752"},
   "erlex": {:hex, :erlex, "0.2.8", "cd8116f20f3c0afe376d1e8d1f0ae2452337729f68be016ea544a72f767d9c12", [:mix], [], "hexpm", "9d66ff9fedf69e49dc3fd12831e12a8a37b76f8651dd21cd45fcf5561a8a7590"},
-  "ex_doc": {:hex, :ex_doc, "0.39.1", "e19d356a1ba1e8f8cfc79ce1c3f83884b6abfcb79329d435d4bbb3e97ccc286e", [:mix], [{:earmark_parser, "~> 1.4.44", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_c, ">= 0.1.0", [hex: :makeup_c, repo: "hexpm", optional: true]}, {:makeup_elixir, "~> 0.14 or ~> 1.0", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1 or ~> 1.0", [hex: :makeup_erlang, repo: "hexpm", optional: false]}, {:makeup_html, ">= 0.1.0", [hex: :makeup_html, repo: "hexpm", optional: true]}], "hexpm", "8abf0ed3e3ca87c0847dfc4168ceab5bedfe881692f1b7c45f4a11b232806865"},
-  "ex_sdp": {:hex, :ex_sdp, "1.1.1", "1a7b049491e5ec02dad9251c53d960835dc5631321ae978ec331831f3e4f6d5f", [:mix], [{:bunch, "~> 1.3", [hex: :bunch, repo: "hexpm", optional: false]}, {:elixir_uuid, "~> 1.2", [hex: :elixir_uuid, repo: "hexpm", optional: false]}], "hexpm", "1b13a72ac9c5c695b8824dbdffc671be8cbb4c0d1ccb4ff76a04a6826759f233"},
+  "ex_doc": {:hex, :ex_doc, "0.39.3", "519c6bc7e84a2918b737aec7ef48b96aa4698342927d080437f61395d361dcee", [:mix], [{:earmark_parser, "~> 1.4.44", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_c, ">= 0.1.0", [hex: :makeup_c, repo: "hexpm", optional: true]}, {:makeup_elixir, "~> 0.14 or ~> 1.0", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1 or ~> 1.0", [hex: :makeup_erlang, repo: "hexpm", optional: false]}, {:makeup_html, ">= 0.1.0", [hex: :makeup_html, repo: "hexpm", optional: true]}], "hexpm", "0590955cf7ad3b625780ee1c1ea627c28a78948c6c0a9b0322bd976a079996e1"},
+  "ex_sdp": {:hex, :ex_sdp, "1.1.2", "7e7465cb13b557cc76ef3e854bad7626b73cc1d1f480d38b5fbcf539c7d8a45d", [:mix], [{:bunch, "~> 1.3", [hex: :bunch, repo: "hexpm", optional: false]}, {:elixir_uuid, "~> 1.2", [hex: :elixir_uuid, repo: "hexpm", optional: false]}], "hexpm", "50a27c2d745924679acca32b3d5499d0b35d135a180b83422df82c289afce564"},
   "file_system": {:hex, :file_system, "1.1.1", "31864f4685b0148f25bd3fbef2b1228457c0c89024ad67f7a81a3ffbc0bbad3a", [:mix], [], "hexpm", "7a15ff97dfe526aeefb090a7a9d3d03aa907e100e262a0f8f7746b78f8f87a5d"},
   "jason": {:hex, :jason, "1.4.4", "b9226785a9aa77b6857ca22832cffa5d5011a667207eb2a0ad56adb5db443b8a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "c5eb0cab91f094599f94d55bc63409236a8ec69a21a67814529e8d5f6cc90b3b"},
   "makeup": {:hex, :makeup, "1.2.1", "e90ac1c65589ef354378def3ba19d401e739ee7ee06fb47f94c687016e3713d1", [:mix], [{:nimble_parsec, "~> 1.4", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "d36484867b0bae0fea568d10131197a4c2e47056a6fbe84922bf6ba71c8d17ce"},
   "makeup_elixir": {:hex, :makeup_elixir, "1.0.1", "e928a4f984e795e41e3abd27bfc09f51db16ab8ba1aebdba2b3a575437efafc2", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.2.3 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "7284900d412a3e5cfd97fdaed4f5ed389b8f2b4cb49efc0eb3bd10e2febf9507"},
-  "makeup_erlang": {:hex, :makeup_erlang, "1.0.2", "03e1804074b3aa64d5fad7aa64601ed0fb395337b982d9bcf04029d68d51b6a7", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm", "af33ff7ef368d5893e4a267933e7744e46ce3cf1f61e2dccf53a111ed3aa3727"},
+  "makeup_erlang": {:hex, :makeup_erlang, "1.0.3", "4252d5d4098da7415c390e847c814bad3764c94a814a0b4245176215615e1035", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm", "953297c02582a33411ac6208f2c6e55f0e870df7f80da724ed613f10e6706afd"},
   "mix_test_watch": {:hex, :mix_test_watch, "1.4.0", "d88bcc4fbe3198871266e9d2f00cd8ae350938efbb11d3fa1da091586345adbb", [:mix], [{:file_system, "~> 0.2 or ~> 1.0", [hex: :file_system, repo: "hexpm", optional: false]}], "hexpm", "2b4693e17c8ead2ef56d4f48a0329891e8c2d0d73752c0f09272a2b17dc38d1b"},
   "mockery": {:hex, :mockery, "2.5.0", "a87acd74fd733aa3b9cb5663d6f690178b056608f2652f18e4ec423ddd5496ed", [:mix], [], "hexpm", "52492b2eba61055df1c626e894663b624b5e6fdfaaaba1d9a8596236fbf4da69"},
   "nimble_parsec": {:hex, :nimble_parsec, "1.4.2", "8efba0122db06df95bfaa78f791344a89352ba04baedd3849593bfce4d0dc1c6", [:mix], [], "hexpm", "4b21398942dda052b403bbe1da991ccd03a053668d147d53fb8c4e0efe09c973"},

test/membrane_rtsp/server/server_logic_test.exs

@@ -224,7 +224,7 @@ defmodule Membrane.RTSP.ServerLogicTest do
                |> Logic.process_request(state)
     end
 
-    test "not allowed when playing", %{state: state} do
+    test "not allowed when playing", %{state: %State{} = state} do
       state = %State{state | session_state: :playing}
 
       mock(:gen_tcp, [send: 2], fn %{}, response ->
@@ -238,7 +238,7 @@ defmodule Membrane.RTSP.ServerLogicTest do
   end
 
   describe "handle PLAY request" do
-    test "handle PLAY request", %{state: state} do
+    test "handle PLAY request", %{state: %State{} = state} do
       uri = %URI{@url | path: "/stream/trackId=0"} |> URI.to_string()
 
       configured_media = %{
@@ -279,7 +279,7 @@ defmodule Membrane.RTSP.ServerLogicTest do
   end
 
   describe "handle TEARDOWN request" do
-    test "Re-initialize the session if it's not playing", %{state: state} do
+    test "Re-initialize the session if it's not playing", %{state: %State{} = state} do
       state = %State{
         state
         | session_state: :ready,
@@ -293,7 +293,7 @@ defmodule Membrane.RTSP.ServerLogicTest do
                |> Logic.process_request(state)
     end
 
-    test "free resources", %{state: state} do
+    test "free resources", %{state: %State{} = state} do
       state = %State{state | session_state: :playing}
 
       mock(FakeHandler, [respond: 2], fn nil, state -> {Response.new(200), state} end)
@@ -304,7 +304,7 @@ defmodule Membrane.RTSP.ServerLogicTest do
                |> Logic.process_request(state)
     end
 
-    test "resets recording_with_tcp? flag", %{state: state} do
+    test "resets recording_with_tcp? flag", %{state: %State{} = state} do
       state = %State{state | recording_with_tcp?: true, session_state: :playing}
 
       mock(FakeHandler, [respond: 2], fn nil, state -> {Response.new(200), state} end)