So startet Pipeline mit pendings Eingängen

Question

Ich bin für ein paar Tage jetzt an einer Pipeline mit folgenden Konfiguration arbeiten: - 2 live Eingangsströme (RTMP) - einen Setzer geht in - outputing zu einem anderen RTMP-Stream

Mit einigen Konverter, Warteschlange, etc. dazwischen, es funktioniert ziemlich gut.

Aber mein Problem ist, dass einer des RTMP-Eingangs nicht zur Startzeit zur Verfügung steht, so dass die Pipeline kann nicht gestartet werden, mit den Fehlern folgenden Absturz: - Fehler: Fehler alle Daten aus dem Stream lesen - Fehler : Interner Datenfluss Fehler

Was wäre der richtige Weg, um dies zu arbeiten, das heißt, den Stream mit der ersten Eingabe zu starten, auch wenn die zweite noch nicht bereit ist?

Ich habe mehrere Möglichkeiten ausprobiert: dynamisch die Pipeline ändern, mit Pad-Sonden spielen, Fehlermeldung anhören, .. aber bis jetzt kann ich es nicht funktionieren lassen.

Danke, PL

Answer 1

Wie Sie haben keinen Code schreiben ich Ihnen konzeptionelle Antwort ok sind erraten ..

Es gibt nur wenige Möglichkeiten für rtspsrc, mit denen Sie steuern können, wenn es ausfällt - reagarding Timeout Überschreitung oder Anzahl der Versuche überschritten Maximum. Das sind (nicht sicher, ob alle):

retry - this may be not very useful if it deals only with ports .. 
timeout - if you want to try with UDP some longer time you can enlarge this one 
tcp-timeout - this is important, try to play with it - make it much larger 
connection-speed - maybe it will help to make smaller this one 
protocols - I have experience that for bad streams TCP was much better for me 

Das eigentliche Konzept (Ich bin kein Experte, nehmen Sie es als eine andere Sicht auf das Problem):

Sie können zwei Bins erstellen - eine für jeden Stream . Ich würde rtspsrc und decodebin verwenden und die Ausgabeplötze von decoderbin blockieren, bis ich alle Blöcke habe, dann würde ich mit dem Compositor verbinden.

Wenn Sie einen Fehler erhalten (es sollte in der Phase des Wartens auf alle Pads sein), dann würden Sie den Behälter in den NULL-Zustand setzen (ich meine GStreamer Zustand NULL genannt) und wieder PLAYING/PAUSED .. Nun, Sie haben die pads richtig zu benutzen (keine ahnung was das ist: D) ​​.. kannst du deinen code dazu posten? Vielleicht versuchen, die Fehlermeldung zu verwerfen, um das Rohr nicht zu zerlegen.

Haben Sie auch nur Videoeingänge? Ich denke, nein, Sie können Audiomixer für Audio verwenden .. auch der Compositor hat schöne OpenGL-Version, die viel schneller ist genannt glvideomixer .. aber es kann eine andere OpenGL-Probleme einführen .. Wenn Sie Intel GPUs haben, dann sind Sie wahrscheinlich sicher.

Answer 2

Danke @otopolsky für die Eingabe. Ich bin mir nicht sicher über Ihre Kommentare zu Eigenschaften von rtspsrc, da ich rtmpsrc verwende.

Wie auch immer, ich habe versucht, Ihrer Idee zu folgen, Eingabestreams in Bins zu setzen. Um meines Demo-Codes willen, habe ich es nur für einen Input-Stream gemacht, mit all den Audio-Sachen beiseite gelegt.

Die Pipeline ist wie folgt:

rtmpsrc => decodebin => videoscale => videorate => capsfilter 
=> compositor.sink_0 => capsfilter => x264enc => flvmux => rtmpsink 

ich bin bauen, sobald die Pipeline im Spielzustand befindet, also der Behälter, um sicherzustellen, dynamisch an die Pipeline hinzugefügt wird. Der Behälter hat alle Elemente des zweiten Eingangs:

rtmpsrc => decodebin => videoscale => videorate => capsfilter 

ich Setup auch ein Geist-Pad für die bin, an dem ich eine Sonde verwenden, die die eigentliche Verbindung zwischen dem Behälter tun wird und die Pipeline:

def _prob_block(self, pad, info, data): 
    pad.remove_probe(info.id) 
    src_pad = self.bin.get_static_pad('src') 
    sink_pad = self.compositor.get_request_pad('sink_1') 
    src_pad.link(sink_pad) 
    return Gst.PadProbeReturn.OK 

ghost_pad = Gst.GhostPad.new('src', self.capsfilter_video_ben.get_static_pad('src')) 
self.bin.add_pad(ghost_pad) 
ghost_pad.add_probe(Gst.PadProbeType.BLOCK_DOWNSTREAM, self._prob_block, None) 

Dann habe ich diese Pseudo-Code für meine Anwendung:

def _activate_bin(): 
    # ... this is where I create bin along with bin elements 
    # ... and create the ghost pad (see above) 
    # Then add the bin to the pipeline 
    pipe._pipeline.add(bin) 
    bin.set_state(Gst.State.PLAYING) 

pipe = Pipeline() 
GObject.timeout_add_seconds(10, _activate_bin) 
pipe._pipeline.set_state(Gst.State.PLAYING) 

Diese fast fehlerlos arbeitet, beginnt die Pipeline mit einem Eingang (sink_0), die durch die Zusammenarbeit geht mpositor und dann in rtmpsink (ich kann tatsächlich den Ausgangsstrom spielen). Dann 10 Sekunden später, ich füge die Bin der Pipeline (Pad 'Senke_1' des Compositor). Auch hier funktioniert das einwandfrei, bis ich beide Eingaben in den Compositor bekommen habe. Hier fängt es an zu frieren.

Von dem, was ich lesen konnte, könnte es von Latenzen und/oder Uhren kommen, aber ich konnte das Problem nicht beheben. Die interessante Sache zu beachten ist, dass ich das Problem nicht habe, wenn ich die Pipeline mit beiden Eingabeströmen starte.

Hätten Sie eine Idee, warum es ein anderes Verhalten hat?