Software Mansion•5d ago

RTSP push approach with Membrane.RTSP.Server

We are trying to add RTSP into media server, using Membrane. The main thing is that we need to implement push approach: Elixir TCP server starts listening for incoming RTSP connections from cameras, and then pushes the incoming RTSP video stream to the clients. We used Membrane.RTSP.Server with handler which handles announce, describe and record steps accordingly. On RECORD step, we are passing socket control to pipeline pid:

Enum.each(tracks, fn {_, track} ->
    options = [
    :binary,
    active: false,
    reuseaddr: true,
    debug: false
    ]

    :ok = :inet.setopts(track.rtp_socket, options)
    :ok = :gen_udp.controlling_process(track.rtp_socket, pipeline_pid)
end)

Enum.each(tracks, fn {_, track} ->
    options = [
    :binary,
    active: false,
    reuseaddr: true,
    debug: false
    ]

    :ok = :inet.setopts(track.rtp_socket, options)
    :ok = :gen_udp.controlling_process(track.rtp_socket, pipeline_pid)
end)

This is done in order to pass the socket control to the UDPReceiver module, which takes socket control from the pipeline and listens for incoming RTP packets. The UDPReceiver module is a Membrane.Source module, which is supposed to receive UDP packets and push them to the output pad.

8 Replies

oleg.okunevychOP•5d ago

UDPReceiver module:

UDPReceiver.ex

oleg.okunevychOP•5d ago

The issue that when we push the stream using FFmpeg, we are receiving warnings for some of video packets and errors for all audio packets. The warnings and errors are as follows:

15:10:28.596 [warning] pid=<0.1070.0> mfa=Membrane.RTP.H264.Depayloader.log_malformed_buffer/2  <0.1043.0>/:depayloader An error occurred while parsing H264 RTP payload.
Reason: malformed_data
Packet: %Membrane.Buffer{payload: <<128, 224, 7, 165, 204, 224, 151, 40, 149, 236, 50, 6, 65, 154, 100, 75, 225, 8, 67, 200, 124, 12, 194, 144, 51, 8, 0, 151, 255, 0, 0, 3, 0, 2, 217, 255, 200, 244, 102, 8, 135, 253, 99, 64, 0, 0, 3, 0, 0, 3, 0, 2, 85, 73, 220, 166, 97, 103, 240, 14, 0, 0, 3, 0, 0, 3, 0, 0, 27, 20, 103, 200, 2, 18, 22, 117, 188, 6, 116, 101, 152, 129, 128, 0, 0, 3, 0, 0, 27, 102, 93, 165, 175, 142, 103, 222, 251, 113, 169, 253, 135, 191, 234, 217, 191, 2, 116, 90, 61, 114, 75, 16, 2, 82, 163, 25, 5, 188, 15, 182, 7, 68, 230, 233, 142, 123, 4, 59, 236, 160, 19, 71, 79, 230, 62, 49, 11, 130, 129, 178, 131, 62, 235, 85, 241, 124, 32, 22, 182, 139, 157, 4, 17, 113, 115, 7, 206, 93, 32, 188, 11, 5, 131, 163, 234, 29, 203, 21, 226, 27, 33, 78, 89, 38, 15, 112, 129, 129, 132, 73, 59, 79, 217, 232, 152, 118, 235, 176, 25, 4, 62, 187, 66, 167, 252, 176, 82, 55, 150, 220, 54, 5, 2, 6, 209, 119, 139, 210, 39, 73, 216, 200, 171, 232, 192, 91, 124, 15, 212, 130, 121, 100, 2, 16, 86, 139, 173, 159, 220, 195, 48, 175, 112, 189, 0, 53, 234, 236, 197, 144, 84, 33, 70, 39, 130, 231, 89, 225, 149, 198, 201, 206, 125, 165, 67, 162, 113, 157, 168, 144, 123, 118, 127, 146, 167, 116, 184, 150, 139, 226, 194, 198, 209, 147, 211, 11, 135, 174, 236, 170, 107, 142, 17, 100, 104, 234, 25, 228, 17, 18, 182, 65, 199, 11, 184, 6, 130, 1, 215, 169, 163, 184, 173, 139, 223, 40, 26, 78, 226, 40, 8, 55, 81, 14, 147, 14, 89, 223, 103, 95, 98, 75, 10, 27, 11, 255, 11>>, pts: nil, dts: nil, metadata: %{name: :source_video, port: 17002, socket: #Port<0.52>, ip: {127, 0, 0, 1}}}

15:10:28.596 [warning] pid=<0.1070.0> mfa=Membrane.RTP.H264.Depayloader.log_malformed_buffer/2  <0.1043.0>/:depayloader An error occurred while parsing H264 RTP payload.
Reason: malformed_data
Packet: %Membrane.Buffer{payload: <<128, 224, 7, 165, 204, 224, 151, 40, 149, 236, 50, 6, 65, 154, 100, 75, 225, 8, 67, 200, 124, 12, 194, 144, 51, 8, 0, 151, 255, 0, 0, 3, 0, 2, 217, 255, 200, 244, 102, 8, 135, 253, 99, 64, 0, 0, 3, 0, 0, 3, 0, 2, 85, 73, 220, 166, 97, 103, 240, 14, 0, 0, 3, 0, 0, 3, 0, 0, 27, 20, 103, 200, 2, 18, 22, 117, 188, 6, 116, 101, 152, 129, 128, 0, 0, 3, 0, 0, 27, 102, 93, 165, 175, 142, 103, 222, 251, 113, 169, 253, 135, 191, 234, 217, 191, 2, 116, 90, 61, 114, 75, 16, 2, 82, 163, 25, 5, 188, 15, 182, 7, 68, 230, 233, 142, 123, 4, 59, 236, 160, 19, 71, 79, 230, 62, 49, 11, 130, 129, 178, 131, 62, 235, 85, 241, 124, 32, 22, 182, 139, 157, 4, 17, 113, 115, 7, 206, 93, 32, 188, 11, 5, 131, 163, 234, 29, 203, 21, 226, 27, 33, 78, 89, 38, 15, 112, 129, 129, 132, 73, 59, 79, 217, 232, 152, 118, 235, 176, 25, 4, 62, 187, 66, 167, 252, 176, 82, 55, 150, 220, 54, 5, 2, 6, 209, 119, 139, 210, 39, 73, 216, 200, 171, 232, 192, 91, 124, 15, 212, 130, 121, 100, 2, 16, 86, 139, 173, 159, 220, 195, 48, 175, 112, 189, 0, 53, 234, 236, 197, 144, 84, 33, 70, 39, 130, 231, 89, 225, 149, 198, 201, 206, 125, 165, 67, 162, 113, 157, 168, 144, 123, 118, 127, 146, 167, 116, 184, 150, 139, 226, 194, 198, 209, 147, 211, 11, 135, 174, 236, 170, 107, 142, 17, 100, 104, 234, 25, 228, 17, 18, 182, 65, 199, 11, 184, 6, 130, 1, 215, 169, 163, 184, 173, 139, 223, 40, 26, 78, 226, 40, 8, 55, 81, 14, 147, 14, 89, 223, 103, 95, 98, 75, 10, 27, 11, 255, 11>>, pts: nil, dts: nil, metadata: %{name: :source_video, port: 17002, socket: #Port<0.52>, ip: {127, 0, 0, 1}}}

15:10:28.603 [error] pid=<0.1086.0> mfa=Membrane.Core.Element.handle_info/2  <0.1043.0>/:aac_depayloader Error occured in Membrane Element:
** (MatchError) no match of right hand side value: <<128, 225, 11, 77, 99, 55, 181, 189, 165, 80, 172, 29, 0, 48, 0, 232, 7, 120, 32, 152, 222, 2, 0, 76, 97, 118, 99, 54, 49, 46, 49, 57, 46, 49, 48, 48, 0, 66, 43, 5, 0, 168, 132, 1, 64, 42, 33, 0, 224, 33, ...>>
    (membrane_rtp_aac_plugin 0.9.4) lib/membrane_element_rtp_aac/utils.ex:63: Membrane.RTP.AAC.Utils.parse_packet/2
    (membrane_rtp_aac_plugin 0.9.4) lib/membrane_element_rtp_aac/depayloader.ex:37: Membrane.RTP.AAC.Depayloader.handle_buffer/4
    (membrane_core 1.1.2) lib/membrane/core/callback_handler.ex:139: Membrane.Core.CallbackHandler.exec_callback/4
    (membrane_core 1.1.2) lib/membrane/core/callback_handler.ex:69: Membrane.Core.CallbackHandler.exec_and_handle_callback/5
    (elixir 1.18.1) lib/enum.ex:2546: Enum."-reduce/3-lists^foldl/2-0-"/3
    (membrane_core 1.1.2) lib/membrane/core/element/buffer_controller.ex:81: Membrane.Core.Element.BufferController.do_handle_incoming_buffers/4
    (membrane_core 1.1.2) lib/membrane/core/element.ex:231: Membrane.Core.Element.handle_info/2
    (stdlib 5.2) gen_server.erl:1095: :gen_server.try_handle_info/3
    (stdlib 5.2) gen_server.erl:1183: :gen_server.handle_msg/6
    (stdlib 5.2) proc_lib.erl:241: :proc_lib.init_p_do_apply/3

15:10:28.603 [error] pid=<0.1086.0> mfa=Membrane.Core.Element.handle_info/2  <0.1043.0>/:aac_depayloader Error occured in Membrane Element:
** (MatchError) no match of right hand side value: <<128, 225, 11, 77, 99, 55, 181, 189, 165, 80, 172, 29, 0, 48, 0, 232, 7, 120, 32, 152, 222, 2, 0, 76, 97, 118, 99, 54, 49, 46, 49, 57, 46, 49, 48, 48, 0, 66, 43, 5, 0, 168, 132, 1, 64, 42, 33, 0, 224, 33, ...>>
    (membrane_rtp_aac_plugin 0.9.4) lib/membrane_element_rtp_aac/utils.ex:63: Membrane.RTP.AAC.Utils.parse_packet/2
    (membrane_rtp_aac_plugin 0.9.4) lib/membrane_element_rtp_aac/depayloader.ex:37: Membrane.RTP.AAC.Depayloader.handle_buffer/4
    (membrane_core 1.1.2) lib/membrane/core/callback_handler.ex:139: Membrane.Core.CallbackHandler.exec_callback/4
    (membrane_core 1.1.2) lib/membrane/core/callback_handler.ex:69: Membrane.Core.CallbackHandler.exec_and_handle_callback/5
    (elixir 1.18.1) lib/enum.ex:2546: Enum."-reduce/3-lists^foldl/2-0-"/3
    (membrane_core 1.1.2) lib/membrane/core/element/buffer_controller.ex:81: Membrane.Core.Element.BufferController.do_handle_incoming_buffers/4
    (membrane_core 1.1.2) lib/membrane/core/element.ex:231: Membrane.Core.Element.handle_info/2
    (stdlib 5.2) gen_server.erl:1095: :gen_server.try_handle_info/3
    (stdlib 5.2) gen_server.erl:1183: :gen_server.handle_msg/6
    (stdlib 5.2) proc_lib.erl:241: :proc_lib.init_p_do_apply/3

FFmpeg command used to push the stream:

ffmpeg -re -stream_loop -1 -i ~/big_buck_bunny_1080p_h264.mov \
-c:v libx264 \
-b:v 2000k \
-maxrate 2000k \
-bufsize 4000k \
-c:a aac \
-b:a 128k \
-ar 44100 \
-ac 2 \
-f rtsp \
-rtsp_transport udp \
-timeout 5000000 \
rtsp://localhost:8554/5e1671fa-171a-4d9b-b2b6-2eaf13452ced

ffmpeg -re -stream_loop -1 -i ~/big_buck_bunny_1080p_h264.mov \
-c:v libx264 \
-b:v 2000k \
-maxrate 2000k \
-bufsize 4000k \
-c:a aac \
-b:a 128k \
-ar 44100 \
-ac 2 \
-f rtsp \
-rtsp_transport udp \
-timeout 5000000 \
rtsp://localhost:8554/5e1671fa-171a-4d9b-b2b6-2eaf13452ced

Would you mind to help us sort out, what's wrong with the implementation?

Billal•5d ago

Is your packets ordered, since you're using UDP, the packets may arrive out of order so you need first to re-order them before depayloading

oleg.okunevychOP•2d ago

@Billal good catch, we actually haven't done it. Implemented a Filter which reorders packets by sequence number and sends when buffer reaches a limit, the amount of warnings reduced, packets are coming to the Parser through the Depayloader, however it got stuck in the Parser (no demand at output pad according to Pipeline Dashboard). PS: We are receiving NACKs from RTCP Receiver, do we need to handle it somehow? Right now we aren't handling the RTCP Socket at all.

13:05:26.381 [debug] pid=<0.1097.0> mfa=Membrane.RTCP.Receiver.handle_event/4  <0.1016.0>/:video_rtp/{:stream_receive_bin, 2787058368}/:rtcp_receiver Sending NACK to 2787058368 with ids [6305, 6306, 6307, 6308, 6309, 6310, 6311, 6312, 6313, 6314, 6315, 6316, 6317, 6318, 6319, 6320, 6321, 6322]

13:05:26.381 [debug] pid=<0.1097.0> mfa=Membrane.RTCP.Receiver.handle_event/4  <0.1016.0>/:video_rtp/{:stream_receive_bin, 2787058368}/:rtcp_receiver Sending NACK to 2787058368 with ids [6305, 6306, 6307, 6308, 6309, 6310, 6311, 6312, 6313, 6314, 6315, 6316, 6317, 6318, 6319, 6320, 6321, 6322]

Billal•2d ago

If you fixed the first issue, now to the second one, most probably erlang is dropping packets because of the internal buffer. Set the buffer and recbuf to some bigger value: options = [ :binary, active: false, reuseaddr: true, debug: false, buffer: 500_000, recbuf: 500_000 ] increase the value if packets are still dropped. In the doc, it says that you should watch recbuf set by the kernel and ajust buffer to be >= that value, however from my own tests I'm still observing dropped packets, so i set both values.

oleg.okunevychOP•2d ago

thank you, it seems setting this option helped with warnings, but still experience issue with Parser: it receives payload, but no actual demand on output pad. What could cause it?

    |> child(:depayloader, Membrane.RTP.H264.Depayloader)
    |> child(:rtsp_video_parser, %Membrane.H264.Parser{
      generate_best_effort_timestamps: %{framerate: {25, 1}},
      output_stream_structure: :annexb
    })
    |> via_in(:input,
      options: [encoding: :H264, segment_duration: Membrane.Time.seconds(4)]
    )
    |> child(:hls, %Membrane.HTTPAdaptiveStream.SinkBin{
      target_window_duration: Membrane.Time.seconds(120),
      manifest_module: Membrane.HTTPAdaptiveStream.HLS,
      storage: %Membrane.HTTPAdaptiveStream.Storages.FileStorage{
        directory: "/tmp/"
      }
    })

    |> child(:depayloader, Membrane.RTP.H264.Depayloader)
    |> child(:rtsp_video_parser, %Membrane.H264.Parser{
      generate_best_effort_timestamps: %{framerate: {25, 1}},
      output_stream_structure: :annexb
    })
    |> via_in(:input,
      options: [encoding: :H264, segment_duration: Membrane.Time.seconds(4)]
    )
    |> child(:hls, %Membrane.HTTPAdaptiveStream.SinkBin{
      target_window_duration: Membrane.Time.seconds(120),
      manifest_module: Membrane.HTTPAdaptiveStream.HLS,
      storage: %Membrane.HTTPAdaptiveStream.Storages.FileStorage{
        directory: "/tmp/"
      }
    })

Billal•2d ago

Most probably the issue is that parameter sets are not in the stream. You should check the SDP sent by ffmpeg to the rtsp server, get the parameters sets and send them to the parser with repeat parameter sets set to true.

|> child(:depayloader, Membrane.RTP.H264.Depayloader)
    |> child(:rtsp_video_parser, %Membrane.H264.Parser{
      generate_best_effort_timestamps: %{framerate: {25, 1}},
      output_stream_structure: :annexb,
      spss: spss,
      ppss: ppss,
      repeat_parameter_sets: true
    })
    |> via_in(:input,
      options: [encoding: :H264, segment_duration: Membrane.Time.seconds(4)]
    )
    |> child(:hls, %Membrane.HTTPAdaptiveStream.SinkBin{
      target_window_duration: Membrane.Time.seconds(120),
      manifest_module: Membrane.HTTPAdaptiveStream.HLS,
      storage: %Membrane.HTTPAdaptiveStream.Storages.FileStorage{
        directory: "/tmp/"
      }
    })

|> child(:depayloader, Membrane.RTP.H264.Depayloader)
    |> child(:rtsp_video_parser, %Membrane.H264.Parser{
      generate_best_effort_timestamps: %{framerate: {25, 1}},
      output_stream_structure: :annexb,
      spss: spss,
      ppss: ppss,
      repeat_parameter_sets: true
    })
    |> via_in(:input,
      options: [encoding: :H264, segment_duration: Membrane.Time.seconds(4)]
    )
    |> child(:hls, %Membrane.HTTPAdaptiveStream.SinkBin{
      target_window_duration: Membrane.Time.seconds(120),
      manifest_module: Membrane.HTTPAdaptiveStream.HLS,
      storage: %Membrane.HTTPAdaptiveStream.Storages.FileStorage{
        directory: "/tmp/"
      }
    })

Again fetch the sps and pps from the SDP

oleg.okunevychOP•2d ago

I've found sps and pps in handle_announce SDP payload and it worked once, will continue debugging, thank you!

Gaming

Programming

RTSP push approach with Membrane.RTSP.Server

Did you find this page helpful?