Stack state error in gstreamer rtspsrc "pad-added" signal callback

[oyclin]

My code is :

local lgi = require 'lgi'
local core = require 'lgi.core'
local GLib = lgi.require("GLib", "2.0")
local Gst = lgi.require("Gst", "1.0")
local Gio = lgi.require("Gio", "2.0")
local GObject = lgi.require("GObject", "2.0")

local log_p = function(str, ...)
    local log = string.format(str, ...)
    print(log)
end
local LLOG = log_p
local LERR = log_p
local LTRACE = log_p


local meta = {}

local codec_type_find_demux = {
    ["PCMA"] = "rtppcmadepay",
    ["PCMU"] = "rtppcmudepay",
    ["AC3"] = "rtpac3depay",
    ["MP4A-LATM"] = "rtpmp4adepay",
    ["MPEG4-GENERIC"] = "rtpmp4gdepay",
    ["AAC"] = "rtpmp4gdepay",
    ["MPA"] = "rtpmpadepay",
    ["MP2"] = "rtpmpadepay",
    ["MP3"] = "rtpmpadepay",
    ["OPUS"] = "rtpopusdepay",
    ["X-GST-OPUS-DRAFT-SPITTKA-00"] = "rtpopusdepay",
    ["MULTIOPUS"] = "rtpopusdepay",
    ["VORBIS"] = "rtpvorbisdepay",
}

local codec_type_parse_element = {
    ["AC3"] = "ac3parse",
    ["MP4A-LATM"] = "aacparse",
    ["MPEG4-GENERIC"] = "aacparse",
    ["AAC"] = "aacparse",
    ["MPA"] = "mpegaudioparse",
    ["MP2"] = "mpegaudioparse",
    ["MP3"] = "mpegaudioparse",
    ["OPUS"] = "opusparse",
    ["PCMA"] = "none",
    ["PCMU"] = "none",
    ["X-GST-OPUS-DRAFT-SPITTKA-00"] = "opusparse",
    ["MULTIOPUS"] = "opusparse",
    ["VORBIS"] = "vorbisparse",
}

function meta:pad_link_audio(src, pad, caps, structure)
    local codec = structure:get_string("encoding-name")
    local depay_name = codec_type_find_demux[codec]
    if not depay_name then
        LERR("Unsupport audio encoding mode : %s", codec)
        return
    end

    local parse_name = codec_type_parse_element[codec]

    local depay = Gst.ElementFactory.make(depay_name)
    local parse = Gst.ElementFactory.make(parse_name)

    local sink = Gst.ElementFactory.make("filesink")
    sink.location = "/tmp/test.aac"

    self.pipeline:add_many(depay, parse, sink)
    depay:link_many(parse, sink)
    pad:link(depay:get_static_pad("sink"))

    depay.state = "PLAYING"
    parse.state = "PLAYING"
    sink.state = "PLAYING"

    self.status.audio_caps_string = caps:to_string()
    LTRACE("====> rtsp link audio pad done ")
end

function meta:pad_link_video(src, pad, caps, structure)
    local codec = structure:get_string("encoding-name")
    local depay, parse, force_caps_string
    if codec == "H264" then
        depay = Gst.ElementFactory.make("rtph264depay")
        parse = Gst.ElementFactory.make("h264parse")
        force_caps_string = "video/x-h264, stream-format=byte-stream, alignment=au"
    elseif codec == "H265" then
        depay = Gst.ElementFactory.make("rtph265depay")
        parse = Gst.ElementFactory.make("h265parse")
        force_caps_string = "video/x-h265, stream-format=byte-stream, alignment=au"
    end

    LTRACE("Get depay = %s, parse = %s", depay, parse)
    if not depay or not parse then
        LERR("Unsupport rtsp video encoding mode : %s", codec)
        return
    end

    local capsfilter = Gst.ElementFactory.make("capsfilter")
    print("capsfilter = ", capsfilter)
    capsfilter.caps = Gst.Caps.from_string(force_caps_string)
    
    local sink = Gst.ElementFactory.make("filesink")
    sink.location = "/tmp/test.h264"

    Gst.Bin.add_many(self.pipeline, depay, parse, capsfilter, sink)
    Gst.Element.link_many(depay, parse, capsfilter, sink)
    local depay_sink_pad = Gst.Element.get_static_pad(depay, "sink")
    Gst.Pad.link(pad, depay_sink_pad)

    depay.state = "PLAYING"
    parse.state = "PLAYING"
    capsfilter.state = "PLAYING"
    sink.state = "PLAYING"

    self.status.video_caps_string = force_caps_string
    LTRACE("====> rtsp link video pad done ")
end


function meta:create_pad_added_callback()
    return function(src, pad)
        local self = self
        LTRACE("src = %s, pad = %s", src, pad)
        local caps = pad.caps
        LTRACE(" caps string = %s", caps:to_string())

        local structure = caps:get_structure(0)
        local name = structure:get_name()
        if name == "application/x-rtp" then
            local media = structure:get_string("media")
            if media == "audio" then
                self:pad_link_audio(src, pad, caps, structure)    
            elseif media == "video" then
                self:pad_link_video(src, pad, caps, structure)
            end
        else
            LERR("Unsupport rtsp mux type, name: %s", name)
        end
    end
end

function meta:init()
    self.status = {}
    self.loop = GLib.MainLoop(nil, false)
    local src = Gst.ElementFactory.make("rtspsrc")
    self.pipeline = Gst.Pipeline.new("testrtspsrc")
    src.location = "rtsp://192.168.0.130:554/130"
    
    GObject.signal_connect_closure(src, "pad-added", GObject.Closure(self:create_pad_added_callback()), 1)
    
    self.pipeline:add_many(src)
    self.pipeline.state = "PLAYING"
    
    self.loop:run()
    LTRACE("========> quit rtsp loop")
    self:stop()
end

meta.__index = meta

local one = {}
setmetatable(one, meta)

one:init()

Run this code, when rtsp source have audio and video pad, it will run twice callback. But second callback will run when first callback not done, and then the stack is in a mess.

[oyclin]

Log like this:

[New Thread 0x7ffff6fc2700 (LWP 3195)]
[New Thread 0x7ffff65a9700 (LWP 3196)]
[New Thread 0x7ffff5da8700 (LWP 3197)]
[New Thread 0x7ffff55a1700 (LWP 3198)]
[New Thread 0x7ffff4c64700 (LWP 3199)]
[New Thread 0x7fffdffff700 (LWP 3200)]
[New Thread 0x7fffdf7fe700 (LWP 3201)]
[New Thread 0x7fffdeffd700 (LWP 3202)]
src = lgi.obj 0x555555845430:Gst.Bin(GstRTSPSrc), pad = lgi.obj 0x7ffff0063740:Gst.GhostPad(GstGhostPad)
 caps string = application/x-rtp, media=(string)video, payload=(int)96, clock-rate=(int)90000, encoding-name=(string)H264, packetization-mode=(string)1, sprop-parameter-sets=(string)"Z0LAKI2NUDwBEvLCAAADAAIAAAMAeR4RCNQ\=\,aM4xsg\=\=", profile-level-id=(string)42c028, level-asymmetry-allowed=(string)1, a-tool=(string)GStreamer, a-type=(string)broadcast, a-framerate=(string)30, a-ts-refclk=(string)local, a-mediaclk=(string)sender, a-ssrc=(string)"301092090\ cname:user1320956193\@host-4fb6fac2", ssrc=(uint)301092090, clock-base=(uint)4245212816, seqnum-base=(uint)6484, npt-start=(guint64)0, play-speed=(double)1, play-scale=(double)1
[New Thread 0x7fffde7fc700 (LWP 3203)]
[New Thread 0x7fffddffb700 (LWP 3204)]
src = lgi.obj 0x555555845430:Gst.Bin(GstRTSPSrc), pad = lgi.obj 0x7fffcc008310:Gst.GhostPad(GstGhostPad)
Get depay = lgi.obj 0x7fffd4010180:Gst.Element(GstRtpH264Depay), parse = attempt to index a nil value
stack traceback:
	test.lua: in function 'target'
	/usr/local/share/lua/5.1/lgi/override/GObject-Closure.lua:286: in function </usr/local/share/lua/5.1/lgi/override/GObject-Closure.lua:283>
	[C]: in function 'make'
	test.lua:84: in function 'pad_link_video'
	test.lua:134: in function 'target'
	/usr/local/share/lua/5.1/lgi/override/GObject-Closure.lua:286: in function </usr/local/share/lua/5.1/lgi/override/GObject-Closure.lua:283>
	[C]: in function 'run'
	test.lua:154: in function 'init'
	test.lua:164: in main chunk
	[C]: at 0x55555555d240
capsfilter = 	lgi.obj 0x7ffff0052ff0:Gst.Element(GstCapsFilter)
/usr/local/bin/luajit: /usr/local/share/lua/5.1/lgi/override/Gst.lua:62: bad argument #2 to 'add' (Gst.Element expected, got string)
[Thread 0x7fffddffb700 (LWP 3204) exited]
[Thread 0x7fffde7fc700 (LWP 3203) exited]
[Thread 0x7fffdf7fe700 (LWP 3201) exited]
[Thread 0x7fffdffff700 (LWP 3200) exited]
[Thread 0x7ffff4c64700 (LWP 3199) exited]
[Thread 0x7ffff55a1700 (LWP 3198) exited]
[Thread 0x7ffff5da8700 (LWP 3197) exited]
[Thread 0x7ffff65a9700 (LWP 3196) exited]
[Thread 0x7ffff6fc2700 (LWP 3195) exited]
[Thread 0x7ffff7c4fb80 (LWP 3194) exited]
[Inferior 1 (process 3194) exited with code 01]