RTMP plugin delivers end_of_stream on TCP disconnection

[dmorn]

This issue is also related to

• RTMP plugin does not deliver end_of_stream on DeleteMessage, as it is not parsing it #789
• end_of_stream is sent when the TCP connection is dropped membrane_rtmp_plugin#52

The RTMP stream is delivering end of stream when the TCP connection is dropped, see here. For us this is plain wrong logic, as one thing is an error such as a connection drop, one thing is an actual end of stream, in amf terms a deleteStream message.

We're turning output HLS streams into VOD when the transmission is ended, and this is an event you cannot withdraw, once player see that tag they stop reloading the playlist.

Can we add an option tunes this behaviour? Or even better (this is how we deal it in our work)

--- a/lib/membrane_rtmp_plugin/rtmp/source/source.ex
+++ b/lib/membrane_rtmp_plugin/rtmp/source/source.ex
@@ -224,8 +224,9 @@ defmodule Membrane.RTMP.Source do
   def handle_info({:socket_closed, _socket}, ctx, state) do
     cond do
       ctx.pads.output.end_of_stream? -> {[], state}
-      ctx.pads.output.start_of_stream? -> {[end_of_stream: :output], state}
-      true -> {[notify_parent: :unexpected_socket_closed, end_of_stream: :output], state}
+      # This might be a connection error. Only deleteStream message signals that
+      # the transmission is finished.
+      true -> {[notify_parent: :unexpected_socket_closed], state}
     end
   end

Basically we just notify the parent of the event, then the pipeline can decide how the rest of the children should react.

[mat-hek]

Since dropping the connection means we won't send anything anymore from the source, the end of stream is quite expected. I assume that the logic is to remove the element on the connection drop and spawn a new one - in that case, some further element should take care of intercepting the end_of_stream and be linked to the newly spawned source.

[dmorn]

Since dropping the connection means we won't send anything anymore from the source

A connection drop is something related to a problem in the transport layer. It signals a network fault, which in my opinion is unrelated to the end of stream event, which means that everything is done (whereas is not in this case).

I assume that the logic is to remove the element on the connection drop and spawn a new one

Thanks to crash groups & dynamic pads this is very easy to accomplish and it is actually how we're handling it.

element should take care of intercepting the end_of_stream

This element the has to know wether that was a false positive or not. I really think this is very confusing. This is an RTMP plugin, RTMP has a message that signals the real end_of_stream, let's just stick with it @mat-hek!

[mat-hek]

I assume that the logic is to remove the element on the connection drop and spawn a new one

Thanks to crash groups & dynamic pads this is very easy to accomplish and it is actually how we're handling it.

Hmm, let's assume we don't send the end_of_stream. You get the unexpected_socket_closed notification and remove the RTMP source and possibly other elements. When you do that, the first thing that happens is that end_of_stream is generated on the input pad just after the removed elements, so it seems the behaviour is the same anyways 🤔

[dmorn]

Not in case the removed elements and the leftovers are connected with a dynamic pad right?

[mat-hek]

I’d have to check, but I think in this case as well

[mat-hek]

Hi @dmorn, any update on this? BTW I checked and the end of stream is always generated:

Mix.install([:membrane_core])

defmodule Source do
  use Membrane.Source
  def_output_pad :output, flow_control: :manual, accepted_format: _any

  @impl true
  def handle_demand(:output, _size, _unit, _ctx, state) do
    {[], state}
  end
end

defmodule Sink do
  use Membrane.Sink
  def_input_pad :input, availability: :on_request, accepted_format: _any

  @impl true
  def handle_end_of_stream(pad, _ctx, state) do
    IO.puts("End of stream on pad #{inspect(pad)}")
    {[], state}
  end
end

defmodule Run do
  import Membrane.ChildrenSpec
  require Membrane.Pad, as: Pad

  def run() do
    p = Membrane.RCPipeline.start_link!()

    Membrane.RCPipeline.exec_actions(p,
      spec: [
        child(:sink, Sink),
        child(:src1, Source) |> via_in(Pad.ref(:input, 1)) |> get_child(:sink),
        child(:src2, Source) |> via_in(Pad.ref(:input, 2)) |> get_child(:sink)
      ]
    )

    Process.sleep(1000)
    Membrane.RCPipeline.exec_actions(p, remove_children: [:src1])
    Process.sleep(1000)
  end
end

Run.run()

[mat-hek]

Closing this for now, feel free to reopen when you tackle this again

[dmorn]

Hi @dmorn, any update on this? BTW I checked and the end of stream is always generated:

@mat-hek we have a bug then 😂 I think you should discuss about this thing with your team. In an HLS stream case, end_of_stream means that the playlist turns VOD. This is irreversible . In our opinion, end_of_stream should be explicit, as in this case with the proper RTMP message, in case of errors or others, maybe the best option would be to terminate the children.

Please let us know what you think as we really have to get it right in our use cases.

[mat-hek]

Well, so I see the following options:

• Add an option to RTMP to crash when the connection is not properly closed. Not sure it helps though.
• Before sending end of stream, send a custom event from RTMP when the connection is not properly closed. This event could then be intercepted by a further element, which wouldn't forward the end of stream to HLS
• Add an option to HLS not to make the playlist VOD until explicitly requested (via notification?)

The second option seems the best to me, let me know what you think ;)

[dmorn]

Hi @mat-hek! I thought about it and maybe I have a third option: custom event for the RTMP unpublish message?

[mat-hek]

👋 can you elaborate on how this would work?

[dmorn]

I'll validate and publish a PR as we need this again :D Basically I'll just record that the unpublish message is received; when sending the end_of_stream action out, I'll prepend it with a custom event action. This way behaviour will be kept unchanged, if someone wants instead to ensure the stream is actually terminated (us), it can do so monitoring the presence of that event. Basically its just option 2 above reversed

[mat-hek]

Got it, seems like a good idea ;)