在不停止管道的情況下發生故障時重新啓動接收器

Question

今天，我決定將基於gst-launch的小腳本轉換爲真正的Python/GStreamer應用程序，以添加一些功能。

由於tee，我開發了一個小程序，可以將我的麥克風的音頻發送到Icecast（shout2send）和本地存儲（filesink）。

由於網絡問題，有時shout2send可能會停止。我想每N秒重新啓動一次這個元素，直到連接回來，而不停止流水線，因爲本地音頻文件不應該受到網絡條件的影響。

這裏是我的嘗試：

1. 停止/啓動管道網絡錯誤（結果：流媒體作品，本地文件被截斷）一秒後
2. 從tee斷開鏈接，設置shout2send狀態NULL和（結果：GStreamer嚴重錯誤，如Trying to dispose element ... but it is in PLAYING instead of the NULL state）
3. 試圖瞭解如何在這種情況下使用焊盤（結果：與上面相同，但涉及更多的代碼）

我該怎麼辦？

這裏是我的代碼看起來像：

import gi 
gi.require_version("Gst", "1.0") 
from gi.repository import GLib 
from gi.repository import Gst 
# [...] 

def message_handler(bus, message): 
    if message.type == Gst.MessageType.ERROR: 
     if message.src == shout2send: 
      pass # TODO: restart the element 
     else: 
      print(message.parse_error()) 
      pipeline.set_state(Gst.State.NULL) 
      exit(1) 
    else: 
     print(message.type) 

pipeline = Gst.Pipeline() 
message_bus = pipeline.get_bus() 
message_bus.add_signal_watch() 
message_bus.connect('message', message_handler) 

# [...] 
tee.link(queue0) 
queue0.link(filesink) 
tee.link(queue1) 
queue1.link(shout2send) 

更新（15年9月12日）：非工作的代碼添加+登錄

我試圖按照"Dynamically changing the pipeline" fro GStreamer doc，但我的代碼沒有按沒有工作。

def event_probe(pad, info, *args): 
    Gst.Pad.remove_probe(pad, info) 
    queue1.unlink(shout2send) 
    tee.unlink(queue1) 
    pipeline.remove(shout2send) 
    pipeline.remove(queue1) 
    return Gst.PadProbeReturn.OK 

def message_handler(bus, message): 
    if message.type == Gst.MessageType.ERROR: 
     if message.src == shout2send: 
      pad = queue1.get_static_pad('src') 
      pad.add_probe(Gst.PadProbeType.BLOCK_DOWNSTREAM, event_probe, None) 
     else: 
      print(message.parse_error()) 
      pipeline.set_state(Gst.State.NULL) 
      exit(1) 
    else: 
     print(message.type) 

這裏是我所看到的，如果我跑我的腳本GST_DEBUG=3和我重新開始的Icecast而流：

[...] 
0:00:02.142033258 5462 0x55e414d900a0 WARN     shout2 gstshout2.c:674:gst_shout2send_render:<shout2send> error: shout_send() failed: Socket error 
0:00:02.658137998 5462 0x55e414d90140 WARN     basesrc gstbasesrc.c:2943:gst_base_src_loop:<pulsesrc> error: Internal data flow error. 
0:00:02.658169752 5462 0x55e414d90140 WARN     basesrc gstbasesrc.c:2943:gst_base_src_loop:<pulsesrc> error: streaming task paused, reason error (-5) 
(GLib.Error('Internal data flow error.', 'gst-stream-error-quark', 1), 'gstbasesrc.c(2943): gst_base_src_loop(): /GstPipeline:pipeline0/GstPulseSrc:pulsesrc:\nstreaming task paused, reason error (-5)') 
0:00:02.658628129 5462 0x7f6ba8002a30 WARN    audiosrc gstaudiosrc.c:244:audioringbuffer_thread_func:<pulsesrc> error reading data -1 (reason: Success), skipping segment 

Answer 1

由於otopolsky的意見，我做到了:)

我做了什麼錯誤的：

1. 元素必須設置爲NULL：這非常重要
2. oggmux必須在tee之後，在兩條子管道上：否則Icecast將列出流，但無法爲其服務。用同樣的方法opusenc

建議：

1. 沒有必要取消鏈接每次你不需要的元素：剛剛突破需要的地方
2. 沒有必要從管道的每一個元素刪除不需要：讓他們，如果你想重用它們

最終代碼（重新連接工作正常且獨立於本地編碼/錄音）：

def event_probe2(pad, info, *args): 
    Gst.Pad.remove_probe(pad, info.id) 
    tee.link(opusenc1) 
    opusenc1.set_state(Gst.State.PLAYING) 
    oggmux1.set_state(Gst.State.PLAYING) 
    queue1.set_state(Gst.State.PLAYING) 
    shout2send.set_state(Gst.State.PLAYING) 
    return Gst.PadProbeReturn.OK 

def reconnect(): 
    pad = tee.get_static_pad('src_1') 
    pad.add_probe(Gst.PadProbeType.BLOCK_DOWNSTREAM, event_probe2, None) 

def event_probe(pad, info, *args): 
    Gst.Pad.remove_probe(pad, info.id) 
    tee.unlink(opusenc1) 
    opusenc1.set_state(Gst.State.NULL) 
    oggmux1.set_state(Gst.State.NULL) 
    queue1.set_state(Gst.State.NULL) 
    shout2send.set_state(Gst.State.NULL) 
    GLib.timeout_add_seconds(interval, reconnect) 
    return Gst.PadProbeReturn.OK 

def message_handler(bus, message): 
    if message.type == Gst.MessageType.ERROR: 
     if message.src == shout2send: 
      pad = tee.get_static_pad('src_1') 
      pad.add_probe(Gst.PadProbeType.BLOCK_DOWNSTREAM, event_probe, None) 
     else: 
      print(message.parse_error()) 
      pipeline.set_state(Gst.State.NULL) 
      exit(1) 
    else: 
     print(message.type) 

小問題：

1. 我用tee.get_static_pad('src_1')，但我想我能得到SRC ID的地方，而不是使用固定的值
2. 也許整個事情會在一個更好的形式被寫入（不過這是我使用Python +的Gstreamer第一個程序和它的作品，所以我與它的罰款）
3. 爲了避免數據丟失，我稱之爲第二後pipeline.set_state(Gst.State.NULL)一個pipeline.send_event(Gst.Event.new_eos())，但我仍然得到像WARN audiosrc gstaudiosrc.c:244:audioringbuffer_thread_func:<pulsesrc> error reading data -1 (reason: Success), skipping segment
消息

代碼：https://github.com/ViGLug/libre-streaming