Membrane RTSP source

Hi @ConnorRigby, you’re totally right - source is the way to pass the stream into a Membrane pipeline. The docs you’re looking for are here: Membrane.Source. As described there, sources should implement Membrane.Element.Base and Membrane.Element.WithOutputPads behaviours. Various source implementations are available at Membrane GitHub, for example file source or portaudio source.

wow thanks. I’m not sure how i managed to miss those docs. Looks like exactly what i needed.

@mat-hek I’ve successfully wired up my first pipeline. It seems like it almost works, but i’ve ran into an issue. My stream only supports interleaving, however the membrane-element-rtp-h264 element does not support interleaving.

There is only one channel, so i tried to hack it together but just filtering out the interleaving packets with something along the lines of this:

  defp process(state, buffer)

  defp process(%{length: l} = state, buffer) when is_number(l) do
    buffer = state.buffer <> buffer
    if byte_size(buffer) >= l do

      <<chunk::binary-size(l), rest::binary>> = buffer
      actions = [buffer: {:output, %Buffer{payload: chunk}}, redemand: :output]
      {{:ok, actions}, %{length: nil, buffer: rest}}

    else

      {{:ok, demand: {:input, state.length}}, %{state | buffer: buffer}}

    end
  end

  defp process(state, <<36, channel::integer-8, length::integer-16, chunk::binary>>) do
    actions = [buffer: {:output, %Buffer{payload: chunk}}, redemand: :output]
    {{:ok, actions}, %{length: nil, buffer: rest}}
    process(%{state | length: length}, rest)
  end

but that didn’t seem to work. In the logs i can see:

[h264 @ 0x7fbee401ed00] non-existing PPS 0 referenced
[h264 @ 0x7fbee401ed00] decode_slice_header error
h264 @ 0x7fbee401ed00] no frame!

and then i get an error:

13:14:04.542 [error] GenServer #PID<0.281.0> terminating
** (Membrane.ActionError) Error while handling :split action:
Unknown error: :send_pkt
Callback: Membrane.Element.FFmpeg.H264.Decoder.handle_process_list
Action args: {:handle_process,
 [
   [
     :input,
     %Membrane.Buffer{
       metadata: %{},
       payload: <<0, 0, 0, 1, 101, 136, 128, 16, 0, 12, 255, 245, 154, 34, 103,
         162, 245, 12, 56, 225, 60, 222, 189, 150, 153, 78, 16, 77, 254, 201,
         165, 53, 240, 253, 3, 133, 170, 112, 0, 254, 178, 211, 19, ...>>
     }
   ]
 ]}

I think my issue is my deinterleaver is actually only outputting RTP frames, and the h264 frames are being dropped? Is there plans to adding “official” support for interleaving? I’d be happy to help contribute it, but i’m having a hard time determining where to add it. wireshark calls it an rtsp interleave frame so i assumed it could be added to the rtsp source, but for the deinterleaver to work properly, it needs to decode the RTP and maybe even the h264 frames. Any help would be appreciated.

What kind of interleaving do you mean? What exactly is interleaved? As far as I understand, h264 RTP interleaving mode works a bit differently than you assume: it’s not about sending multiple h264 channels in one stream, but sending NALUs in an order different than the one they are encoded. For example, if your stream is
A B C D E F G H
it can be sent as
A D G B E H C F

After receiving, the original order needs to be restored. This way aims to avoid loosing too many subsequent frames at once. It is not really widely supported feature, so it’s quite strange that your camera supports only that. This is described in RFC 6184.

To support this, you would need to adjust the depayloader. The depayloader gets RTP packets payload (not entire RTP packets - these are parsed by the RTP parser, one ‘step’ before) and outputs h264 stream. The stream can be payloaded in different modes. Currently, the two usually used ones are supported: Fu-a and Stap-a, however, it’s possible that your camera uses another. Depending on the mode, there are some ways to get Decoding Order Number, thanks to which you can order the NALUs properly. For sure you shouldn’t decode h264 there.

The error you ended up with basically means the decoder failed because of invalid input. The logs indicate that a PPS NALU is either missing or at an improper position in the stream. Make sure to enable membrane logger to have all the logs present/

Thanks for the reply.
I’ve started reading the RFC for deinterleaving. You are correct in that i will not need to decode h264. The reason the camera requires interleaving is because it streams audio and video on the same RTSP session. The DESCRIBE method lists two available channels. I had assumed the packets were in order because WireShark automatically reorders them based on the interleaving data. In reality, they are coming out of order. It looks like i will need to make my own depayloader to output my mpeg-4 audio data along with h264 data. After reading the spec, it doesn’t look that hard to implement deinterleaving.

Oh, that’s another interleaving I haven’t noticed you mentioned audio. In case of RTSP+RTP, there are two ways of having two (e.g. audio and video) streams in one RTSP session:

• having two RTP sessions
• having them interleaved in one RTP session

To my knowledge, in either option, both streams are payloaded and depayloaded separately - the RTP parser distinguishes the stream by ssrc and passes to the proper depayloader. That case is fully supported and even implemented as our RTP demo

oh interesting. Thanks for the heads up. Maybe i can get rid of my filter and just try using the rtp bin module. I haven’t given it a try yet as it looked like the bin feature was still a little new. Will report back

update: i tried this out:

  def handle_init(_) do
    spec = %ParentSpec{
      children: [
        rtsp: %RTSP{
          location: "rtsp://localhost:554/axis-media/media.amp/"
        },
        rtp: %RTP.Receiver{fmt_mapping: %{96 => "H264", 97 => "MPA"}}
      ],
      links: [
        link(:rtsp) 
        |> to(:rtp)
      ]
    }

    {{:ok, spec: spec}, %{mpa: nil, h264: nil}}
  end

and it failed with:

09:34:15.033 [error] GenServer #PID<0.287.0> terminating
** (Membrane.ActionError) Error while handling :split action:
Unknown error: :wrong_version
Callback: Membrane.Element.RTP.Parser.handle_process_list
Action args: {:handle_process,
 [
   [
     :input,
     %Membrane.Buffer{
       metadata: %{},
       payload: <<36, 0, 5, 120, 128, 96, 54, 103, 86, 115, 59, 168, 242, 53,
         181, 184, 124, 133, 136, 128, 24, 0, 12, 255, 245, 154, 34, 103, 162,
         245, 12, 56, 225, 60, 222, 189, 150, 153, 78, 16, 77, 254, 201, ...>>
     }
   ]
 ]}

in that payload the <<36, 0, 5, 120>> is the RTP interleave packet: <<36, channel_id::integer-8, rtp_packet_length::integer-16>> so it doesn’t look like it’s currently supported, unless i’m missing an option somewhere.

Ok, that’s an interleaving I haven’t heard about yet - not only audio and video are interleaved, but also RTSP packets. I’ve done short research and it seems I finally understand your first post So basing on RFC 7826 it looks like we have the following types of packets:

• RTP packets (prepended with this four-byte header)
• RTSP packets (plain text)
• possibly RTCP packets (prepended by the same header as RTP, but with the channel id incremented by one, as far as I understood)

So I think the parsing should be something like:

def dissect(<<36, rtp_channel::integer-8, length::integer-16, chunk::binary-size(length), rest::binary>>, rtp_channel) do
  # chunk is an RTP packet, forward to RTP parser
end

def dissect(<<36, rtcp_channel::integer-8, length::integer-16, chunk::binary-size(length), rest::binary>>, rtp_channel) when rtcp_channel == rtp_channel+1 do
  # chunk is an RTCP packet - not yet supported in Membrane - ignore or extract some information if needed
end

def dissect(<<36, _::binary>>, _rtp_channel) do
  # request more data
end

def dissect(rtsp_packet, _rtp_channel) do
  # got RTSP packet - even if not needed, has to be parsed to identify the beginning of the next packet,
  # that may be RTP. https://github.com/membraneframework/membrane-protocol-rtsp/blob/master/lib/rtsp/response.ex
  # may be helpful. You can also try skipping until the next 36 byte, but not sure it will work.
end

This should be done before the RTP parser (or before the RTP receiver bin). I don’t know RTSP that thoroughly, but hopefully this will finally be helpful.

Thanks for the input. Should i implement this as a Membrane Filter or something else? I tried a filter before, but i second guessed myself.