GStreamer中用Tee实现RTSP多客户端流推送失败求助

作为GStreamer新手，想要创建一个RTSP服务器，实现每个输出流仅消费一次源。最初在工厂的do_create_element()中直接返回source_pipeline时，单客户端可正常连接消费流，但多客户端连接时会因src pad已被占用报错崩溃。

尝试改用Tee元素，在其后添加Queue并包装为Pipeline返回，但客户端连接时出现**"SDP contains no streams"**错误，手动链接Tee和Queue也无效，怀疑Queue缺少消费端，寻求解决方案。

原始示例代码 #

import gi

gi.require_version("Gst", "1.0")
gi.require_version("GstRtspServer", "1.0")

# noinspection PyUnresolvedReferences
from gi.repository import Gst, GstRtspServer, GObject


def create_source():
    s_src = "videotestsrc ! video/x-raw,rate=30,width=320,height=240,format=I420"
    s_h264 = "x264enc tune=zerolatency"
    pipeline_str = "( {s_src} ! queue max-size-buffers=1000 name=q_enc ! {s_h264} ! rtph264pay name=pay0 pt=96 )".format(
        **locals()
    )
    return Gst.parse_launch(pipeline_str)


def _main():
    GObject.threads_init()
    Gst.init(None)

    source = create_source()
    tee = Gst.ElementFactory.make("tee", "tee")

    source_pipeline = Gst.Pipeline()
    source_pipeline.add(source)
    source_pipeline.add(tee)

    source_pipeline.set_state(Gst.State.PLAYING)

    class Factory(GstRtspServer.RTSPMediaFactoryURI):
        def __init__(self):
            super(Factory, self).__init__()
            self._n = 0

        def do_create_element(self, url):  # -> GstBin (Pipeline)
            print("in [Factory] do_create_element")

            source_pipeline.set_state(Gst.State.PAUSED)

            stream_pipeline = Gst.Pipeline()

            # Create a new queue to buffer the tee's output
            q = Gst.ElementFactory.make("queue")
            stream_pipeline.add(q)

            source_pipeline.add(tee)
            source_pipeline.add(stream_pipeline)
            # tee.link(q)

            stream_pipeline.set_state(Gst.State.PLAYING)
            source_pipeline.set_state(Gst.State.PLAYING)

            return stream_pipeline

    factory = Factory()
    factory.set_shared(True)

    server = GstRtspServer.RTSPServer()
    server.get_mount_points().add_factory("/endpoint", factory)
    server.attach(None)

    loop = GObject.MainLoop()
    loop.run()


if __name__ == "__main__":
    _main()

客户端连接错误输出 #

gst-launch-1.0 rtspsrc location=rtsp://127.0.0.1:8554/endpoint ! rtph264depay ! h264parse ! mp4mux ! filesink location=file.mp4

Setting pipeline to PAUSED ...
Pipeline is live and does not need PREROLL ...
Progress: (open) Opening Stream
Pipeline is PREROLLED ...
Prerolled, waiting for progress to finish...
Progress: (connect) Connecting to rtsp://127.0.0.1:8554/endpoint
Progress: (open) Retrieving server options
Progress: (open) Retrieving media info
ERROR: from element /GstPipeline:pipeline0/GstRTSPSrc:rtspsrc0: Could not get/set settings from/on resource.
Additional debug info:
../gst/rtsp/gstrtspsrc.c(7637): gst_rtspsrc_setup_streams_start (): /GstPipeline:pipeline0/GstRTSPSrc:rtspsrc0:
SDP contains no streams
ERROR: pipeline doesn't want to preroll.
Setting pipeline to NULL ...
Freeing pipeline ...

核心问题分析 #

1. 错误使用工厂类：RTSPMediaFactoryURI用于处理外部URI源，自定义共享流应使用RTSPMediaFactory
2. 管道结构错误：返回的客户端管道中缺少RTP支付器（pay0），导致RTSP服务器无法生成包含流信息的SDP
3. Tee分支链接逻辑错误：没有正确为每个客户端创建独立的分支管道并链接到Tee的请求pad

修复后的完整代码 #

import gi

gi.require_version("Gst", "1.0")
gi.require_version("GstRtspServer", "1.0")

from gi.repository import Gst, GstRtspServer, GObject


def create_shared_source_pipeline():
    # 创建共享的源管道，最后输出到Tee
    s_src = "videotestsrc ! video/x-raw,rate=30,width=320,height=240,format=I420"
    s_h264 = "x264enc tune=zerolatency"
    # 把RTSP支付器移到客户端分支，源管道只负责编码到H264
    pipeline_str = f"{s_src} ! queue max-size-buffers=1000 ! {s_h264} ! tee name=tee"
    pipeline = Gst.parse_launch(pipeline_str)
    return pipeline


def _main():
    GObject.threads_init()
    Gst.init(None)

    # 初始化共享源管道
    shared_pipeline = create_shared_source_pipeline()
    tee = shared_pipeline.get_by_name("tee")
    shared_pipeline.set_state(Gst.State.PLAYING)

    class SharedStreamFactory(GstRtspServer.RTSPMediaFactory):
        def __init__(self):
            super().__init__()
            self.set_shared(True)  # 开启共享模式，多个客户端复用源
            self.shared_pipeline = shared_pipeline
            self.tee = tee

        def do_create_pipeline(self, url):
            # 为每个客户端创建独立的分支管道：queue + rtph264pay
            client_pipeline_str = "queue ! rtph264pay name=pay0 pt=96"
            client_pipeline = Gst.parse_launch(client_pipeline_str)

            # 获取Tee的请求pad，链接到客户端管道的sink pad
            tee_pad = self.tee.get_request_pad("src_%u")
            client_sink_pad = client_pipeline.get_static_pad("sink")
            tee_pad.link(client_sink_pad)

            # 创建媒体管道，把客户端分支加入其中
            media_pipeline = Gst.Pipeline.new(f"client-pipeline-{self.get_next_id()}")
            media_pipeline.add(client_pipeline)

            return media_pipeline

    # 初始化服务器
    server = GstRtspServer.RTSPServer()
    mount_points = server.get_mount_points()
    factory = SharedStreamFactory()
    mount_points.add_factory("/endpoint", factory)
    server.attach(None)

    print("RTSP server running at rtsp://127.0.0.1:8554/endpoint")
    loop = GObject.MainLoop()
    loop.run()


if __name__ == "__main__":
    _main()

关键修复点说明 #

1. 改用RTSPMediaFactory：重写do_create_pipeline方法，更适合自定义共享媒体流的场景
2. 分离共享源与客户端分支：共享源只负责视频采集和编码，每个客户端独立拥有queue + rtph264pay的分支，避免pad冲突
3. 动态链接Tee分支：通过get_request_pad为每个客户端获取Tee的独立输出pad，链接到客户端管道的sink，确保多客户端无冲突
4. 保留pay0元素：客户端分支包含rtph264pay name=pay0，RTSP服务器会自动识别该元素并生成正确的SDP流信息

测试验证 #

使用原客户端命令测试：

gst-launch-1.0 rtspsrc location=rtsp://127.0.0.1:8554/endpoint ! rtph264depay ! h264parse ! mp4mux ! filesink location=file.mp4

此时多客户端可同时连接，不会出现pad冲突或SDP错误。

内容的提问来源于stack exchange，提问作者Samuel Yvon