Handle socket closed by client before start of stream arrives in RTMP…

… Source Bin (#87)

* Handle unexpected socket closed by client before start of stream arrives

* Bump version to 0.23.0

README.md

@@ -14,7 +14,7 @@ The package can be installed by adding `membrane_rtmp_plugin` to your list of de
 ```elixir
 def deps do
   [
-	  {:membrane_rtmp_plugin, "~> 0.22.1"}
+	  {:membrane_rtmp_plugin, "~> 0.23.0"}
   ]
 end
 ```

lib/membrane_rtmp_plugin/rtmp/source/bin.ex

@@ -105,6 +105,10 @@ defmodule Membrane.RTMP.SourceBin do
     {[notify_parent: notification], state}
   end
 
+  def handle_child_notification(:unexpected_socket_closed, :src, _ctx, state) do
+    {[notify_parent: :unexpected_socket_close], state}
+  end
+
   @doc """
   Passes the control of the socket to the `source`.
 

mix.exs

@@ -1,7 +1,7 @@
 defmodule Membrane.RTMP.Mixfile do
   use Mix.Project
 
-  @version "0.22.1"
+  @version "0.23.0"
   @github_url "https://github.com/membraneframework/membrane_rtmp_plugin"
 
   def project do

test/membrane_rtmp_plugin/rtmp_source_bin_test.exs

@@ -184,6 +184,23 @@ defmodule Membrane.RTMP.SourceBin.IntegrationTest do
     assert :error = Task.await(ffmpeg_task)
   end
 
+  test "Handles client socket close before start of stream" do
+    {:ok, port} = start_tcp_server()
+
+    Task.async(fn ->
+      host = @local_ip |> String.to_charlist() |> :inet.parse_address() |> elem(1)
+      {:ok, socket} = :gen_tcp.connect(host, port, [], :infinity)
+      :gen_tcp.close(socket)
+    end)
+
+    pipeline = await_pipeline_started()
+
+    assert_end_of_stream(pipeline, :audio_sink, :input)
+    assert_end_of_stream(pipeline, :video_sink, :input)
+
+    assert_pipeline_notified(pipeline, :src, :unexpected_socket_close)
+  end
+
   defp start_tcp_server(validator \\ %Membrane.RTMP.MessageValidator.Default{}) do
     test_process = self()