catie/internal/ndi/device.go

package ndi

import (
	"fmt"
	"github.com/go-gst/go-gst/gst"
	"github.com/google/uuid"
	"github.com/rs/zerolog/log"
	"sync"
)

type Device struct {
	sync.RWMutex

	name  string
	uri   string
	props map[string]interface{}

	available bool

	gstDevice *gst.Device

	pipeline   *gst.Pipeline
	videoTee   *gst.Element
	videoProxy map[string]*gst.Element
	audioTee   *gst.Element
	audioProxy map[string]*gst.Element
}

func newDevice(d *gst.Device) *Device {
	name := d.GetDisplayName()
	uri := d.GetProperties().Values()["url-address"].(string)
	props := d.GetProperties().Values()

	device := &Device{
		name:      name,
		uri:       uri,
		props:     props,
		available: true,

		gstDevice: d,

		videoProxy: make(map[string]*gst.Element),
		audioProxy: make(map[string]*gst.Element),
	}

	err := device.createPipeline()
	if err != nil {
		log.Error().Err(err).Msg("Failed to create pipeline for device")
		return nil
	}

	return device
}

func (d *Device) updateProperties(dgst *gst.Device) {
	d.Lock()
	defer d.Unlock()

	d.name = dgst.GetDisplayName()
	d.uri = dgst.GetProperties().Values()["url-address"].(string)
	d.props = dgst.GetProperties().Values()
	d.available = true
}

func (d *Device) setUnavailable() {
	d.Lock()
	defer d.Unlock()

	d.available = false
}

func (d *Device) GetDisplayName() string {
	d.RLock()
	defer d.RUnlock()

	return d.name
}

func (d *Device) GetDeviceType() string {
	return deviceType
}

func (d *Device) GetProperties() map[string]interface{} {
	d.RLock()
	defer d.RUnlock()

	return d.props
}

// TODO handle case where ndidemux src pads are not present
func (d *Device) createPipeline() error {
	d.Lock()
	defer d.Unlock()

	pipeline, err := gst.NewPipeline("ndi-" + d.name)
	if err != nil {
		return fmt.Errorf("creating pipeline: %w", err)
	}
	pipeline.ForceClock(gst.ObtainSystemClock().Clock)

	input := d.gstDevice.CreateElement("")
	inputQueue, err := gst.NewElementWithProperties("queue2",
		map[string]interface{}{
			"name":             "input-queue",
			"max-size-buffers": 0,
			"max-size-bytes":   0,
			"max-size-time":    40000000, // 40ms
		})
	if err := input.Link(inputQueue); err != nil {
		return fmt.Errorf("linking input to input input-queue: %w", err)
	}

	ndiDemuxer, err := gst.NewElement("ndisrcdemux")
	if err != nil {
		return fmt.Errorf("creating ndisrcdemux: %w", err)
	}

	if err := inputQueue.Link(ndiDemuxer); err != nil {
		return fmt.Errorf("linking input-queue to ndisrcdemux: %w", err)
	}

	videoTee, err := gst.NewElementWithName("tee", "video-tee")
	if err != nil {
		return fmt.Errorf("creating video-tee: %w", err)
	}

	audioTee, err := gst.NewElementWithName("tee", "audio-tee")
	if err != nil {
		return fmt.Errorf("creating audio-tee: %w", err)
	}

	if err := pipeline.AddMany(input, ndiDemuxer, videoTee, audioTee); err != nil {
		return fmt.Errorf("adding elements to pipeline: %w", err)
	}

	d.pipeline = pipeline
	d.videoTee = videoTee
	d.audioTee = audioTee

	if _, err := ndiDemuxer.Connect("pad-added", d.onNdiSrcDemuxerPadAdded); err != nil {
		return fmt.Errorf("connecting pad-added signal: %w", err)
	}

	return nil
}

func (d *Device) onNdiSrcDemuxerPadAdded(element *gst.Element, pad *gst.Pad) {
	switch pad.GetName() {
	case "video":
		if plr := d.videoTee.GetStaticPad("sink").Link(pad); plr != gst.PadLinkOK {
			log.Error().Str("pad", pad.GetName()).Msgf("linking video pad to video tee: %s", plr)
		}
	case "audio":
		if plr := d.audioTee.GetStaticPad("sink").Link(pad); plr != gst.PadLinkOK {
			log.Error().Str("pad", pad.GetName()).Msgf("linking audio pad to audio tee: %s", plr)
		}
	}

}

func (d *Device) GetOutput() (string, *gst.Element, *gst.Element, error) {
	id, videoElement, audioElement, err := func() (string, *gst.Element, *gst.Element, error) {
		d.Lock()
		defer d.Unlock()
		id := uuid.NewString()

		videoProxySink, err := gst.NewElementWithName("proxysink", "video-proxy")
		if err != nil {
			return "", nil, nil, fmt.Errorf("creating video-proxy sink: %w", err)
		}
		if err := d.pipeline.Add(videoProxySink); err != nil {
			return "", nil, nil, fmt.Errorf("adding video-proxy sink to pipeline: %w", err)
		}

		if plr := d.videoTee.GetRequestPad("src_%u").Link(videoProxySink.GetStaticPad("sink")); plr != gst.PadLinkOK {
			return "", nil, nil, fmt.Errorf("linking video-tee to video-proxy sink: %s", plr)
		}

		if !videoProxySink.SyncStateWithParent() {
			log.Warn().Msg("video-proxy sink sync state not syncable")
		}

		audioProxySink, err := gst.NewElementWithName("proxysink", "audio-proxy")
		if err != nil {
			return "", nil, nil, fmt.Errorf("creating audio-proxy sink: %w", err)
		}

		if err := d.pipeline.Add(audioProxySink); err != nil {
			return "", nil, nil, fmt.Errorf("adding audio-proxy sink to pipeline: %w", err)
		}

		if plr := d.audioTee.GetRequestPad("src_%u").Link(audioProxySink.GetStaticPad("sink")); plr != gst.PadLinkOK {
			return "", nil, nil, fmt.Errorf("linking audio-tee to audio-proxy sink: %s", plr)
		}

		if !audioProxySink.SyncStateWithParent() {
			log.Warn().Msg("audio-proxy sink sync state not syncable")
		}

		d.videoProxy[id] = videoProxySink
		d.audioProxy[id] = audioProxySink

		return id, videoProxySink, audioProxySink, nil
	}()
	if err != nil {
		return "", nil, nil, fmt.Errorf("creating output: %w", err)
	}
	if err := d.setPipelineState(gst.StatePlaying); err != nil {
		return "", nil, nil, fmt.Errorf("setting pipeline state: %w", err)
	}

	return id, videoElement, audioElement, nil
}

func (d *Device) setPipelineState(state gst.State) error {
	d.Lock()
	defer d.Unlock()

	if d.pipeline == nil {
		return fmt.Errorf("pipeline is nil")
	}

	if state == d.pipeline.GetCurrentState() {
		return nil
	}

	if err := d.pipeline.SetState(state); err != nil {
		return fmt.Errorf("setting pipeline state: %w", err)
	}

	return nil
}