Stream reset on connection error

pkg/pipeline/output/bin.go

@@ -178,6 +178,15 @@ func (b *Bin) AddStream(url string) error {
 	return o.(*StreamOutput).AddSink(b.bin, url)
 }
 
+func (b *Bin) ResetStream(name string, streamErr error) (bool, error) {
+	o := b.outputs[types.EgressTypeStream]
+	if o == nil {
+		return false, errors.ErrStreamNotFound(name)
+	}
+
+	return o.(*StreamOutput).Reset(name, streamErr)
+}
+
 func (b *Bin) GetStreamUrl(name string) (string, error) {
 	o := b.outputs[types.EgressTypeStream]
 	if o == nil {

pkg/pipeline/output/stream.go

@@ -211,6 +211,41 @@ func (o *StreamOutput) AddSink(bin *gst.Bin, url string) error {
 	return nil
 }
 
+func (o *StreamOutput) Reset(name string, streamErr error) (bool, error) {
+	o.RLock()
+	defer o.RUnlock()
+
+	var url string
+	var sink *streamSink
+	for u, s := range o.sinks {
+		if s.sink.GetName() == name {
+			url = u
+			sink = s
+			break
+		}
+	}
+	if sink == nil {
+		return false, errors.ErrStreamNotFound(name)
+	}
+
+	sink.reconnections++
+	if sink.reconnections > 3 {
+		return false, nil
+	}
+
+	redacted, _ := utils.RedactStreamKey(url)
+	logger.Warnw("resetting stream", streamErr, "url", redacted)
+
+	if err := sink.sink.BlockSetState(gst.StateNull); err != nil {
+		return false, err
+	}
+	if err := sink.sink.BlockSetState(gst.StatePlaying); err != nil {
+		return false, err
+	}
+
+	return true, nil
+}
+
 func (o *StreamOutput) RemoveSink(bin *gst.Bin, url string) error {
 	o.Lock()
 	defer o.Unlock()
@@ -253,9 +288,10 @@ func (o *StreamOutput) RemoveSink(bin *gst.Bin, url string) error {
 }
 
 type streamSink struct {
-	pad   string
-	queue *gst.Element
-	sink  *gst.Element
+	pad           string
+	queue         *gst.Element
+	sink          *gst.Element
+	reconnections int
 }
 
 func (o *streamSink) link(tee *gst.Element, live bool) error {

pkg/pipeline/watch.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"fmt"
 	"regexp"
+	"strings"
 	"time"
 
 	"github.com/tinyzimmer/go-glib/glib"
@@ -163,7 +164,15 @@ func (p *Pipeline) handleMessageError(gErr *gst.GError) error {
 
 	switch {
 	case element == elementGstRtmp2Sink:
-		// bad URI or could not connect. Remove rtmp output
+		if strings.HasPrefix(gErr.Error(), "Connection error") {
+			// try reconnecting
+			ok, err := p.out.ResetStream(name, gErr)
+			if ok || err != nil {
+				return err
+			}
+		}
+
+		// remove sink
 		url, err := p.out.GetStreamUrl(name)
 		if err != nil {
 			logger.Warnw("rtmp output not found", err, "url", url)