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"
	"os"
	"path/filepath"
	"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()

	log.Debug().Str("name", d.name).Msg("Creating pipeline for device")

	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("")
	if err := pipeline.Add(input); err != nil {
		return fmt.Errorf("adding input to pipeline: %w", err)
	}

	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 != nil {
		return fmt.Errorf("creating input-queue: %w", err)
	}
	if err := pipeline.Add(inputQueue); err != nil {
		return fmt.Errorf("adding input-queue to pipeline: %w", err)
	}

	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 := pipeline.Add(ndiDemuxer); err != nil {
		return fmt.Errorf("adding ndisrcdemux to pipeline: %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)
	}
	if err := pipeline.Add(videoTee); err != nil {
		return fmt.Errorf("adding video-tee to pipeline: %w", err)
	}

	audioTee, err := gst.NewElementWithName("tee", "audio-tee")
	if err != nil {
		return fmt.Errorf("creating audio-tee: %w", err)
	}
	if err := pipeline.Add(audioTee); err != nil {
		return fmt.Errorf("adding audio-tee 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)
	}
	cwd, err := os.Getwd()
	if err != nil {
		return fmt.Errorf("getting current working directory: %w", err)
	}
	log.Debug().Str("cwd", cwd).Msg("dumping pipeline to dot file")
	filename := filepath.Join(cwd, "dumpDot", "ndi-pipeline.dot")
	dot := pipeline.DebugBinToDotData(gst.DebugGraphShowAll)
	if err := os.WriteFile(filename, []byte(dot), 0644); err != nil {
		return fmt.Errorf("writing dot file: %w", err)
	}

	log.Debug().Str("name", d.name).Msg("Pipeline created")

	return nil
}

func (d *Device) onNdiSrcDemuxerPadAdded(element *gst.Element, pad *gst.Pad) {
	// This is called when ndisrcdemux adds a source pad (e.g., "video", "audio")
	// We need to link this source pad to the sink pad of the corresponding tee.
	log.Debug().Str("pad", pad.GetName()).Msg("NDI demuxer pad added, linking to tee sink")

	var teeSinkPad *gst.Pad
	var teeName string
	switch pad.GetName() {
	case "video":
		teeSinkPad = d.videoTee.GetStaticPad("sink")
		teeName = d.videoTee.GetName()
	case "audio":
		teeSinkPad = d.audioTee.GetStaticPad("sink")
		teeName = d.audioTee.GetName()
	default:
		log.Warn().Str("pad", pad.GetName()).Msg("Ignoring unknown pad from ndisrcdemux")
		return
	}

	if teeSinkPad == nil {
		log.Error().Str("pad", pad.GetName()).Str("tee", teeName).Msg("Failed to get static sink pad from tee")
		return
	}

	// Link the demuxer's source pad to the tee's sink pad
	log.Info().Str("src_pad", pad.GetName()).Str("sink_pad", teeSinkPad.GetName()).Str("tee", teeName).Msg("Attempting to link NDI demuxer pad to tee")
	if plr := pad.Link(teeSinkPad); plr != gst.PadLinkOK {
		log.Error().Str("src_pad", pad.GetName()).Str("sink_pad", teeSinkPad.GetName()).Str("tee", teeName).Msgf("Linking NDI demuxer pad to tee failed: %s", plr)
	} else {
		log.Info().Str("src_pad", pad.GetName()).Str("sink_pad", teeSinkPad.GetName()).Str("tee", teeName).Msg("Successfully linked NDI demuxer pad to tee")
	}
}

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
	}

	log.Info().Str("pipeline", d.pipeline.GetName()).Str("target_state", state.String()).Msg("Setting pipeline state")
	if err := d.pipeline.SetState(state); err != nil {
		log.Error().Str("pipeline", d.pipeline.GetName()).Str("target_state", state.String()).Err(err).Msg("SetState failed")
		return fmt.Errorf("setting pipeline state to %s: %w", state.String(), err)
	}

	return nil
}