diff --git a/internal/capture/broadcast.go b/internal/capture/broadcast.go index f256818b..2ded2cd0 100644 --- a/internal/capture/broadcast.go +++ b/internal/capture/broadcast.go @@ -16,7 +16,7 @@ type BroacastManagerCtx struct { logger zerolog.Logger mu sync.Mutex - pipeline *gst.Pipeline + pipeline gst.Pipeline pipelineMu sync.Mutex pipelineFn func(url string) (string, error) diff --git a/internal/capture/screencast.go b/internal/capture/screencast.go index 0383c7f9..272b3964 100644 --- a/internal/capture/screencast.go +++ b/internal/capture/screencast.go @@ -23,7 +23,7 @@ type ScreencastManagerCtx struct { mu sync.Mutex wg sync.WaitGroup - pipeline *gst.Pipeline + pipeline gst.Pipeline pipelineStr string pipelineMu sync.Mutex @@ -202,7 +202,7 @@ func (manager *ScreencastManagerCtx) createPipeline() error { // get first image select { - case image, ok := <-manager.pipeline.Sample: + case image, ok := <-manager.pipeline.Sample(): if !ok { return errors.New("unable to get first image") } else { @@ -220,7 +220,7 @@ func (manager *ScreencastManagerCtx) createPipeline() error { defer manager.wg.Done() for { - image, ok := <-pipeline.Sample + image, ok := <-pipeline.Sample() if !ok { manager.logger.Debug().Msg("stopped receiving images") return diff --git a/internal/capture/streamsink.go b/internal/capture/streamsink.go index a6ab78d8..25df5259 100644 --- a/internal/capture/streamsink.go +++ b/internal/capture/streamsink.go @@ -24,7 +24,7 @@ type StreamSinkManagerCtx struct { wg sync.WaitGroup codec codec.RTPCodec - pipeline *gst.Pipeline + pipeline gst.Pipeline pipelineMu sync.Mutex pipelineFn func() (string, error) @@ -275,7 +275,7 @@ func (manager *StreamSinkManagerCtx) createPipeline() error { defer manager.wg.Done() for { - sample, ok := <-pipeline.Sample + sample, ok := <-pipeline.Sample() if !ok { manager.logger.Debug().Msg("stopped emitting samples") return diff --git a/internal/capture/streamsrc.go b/internal/capture/streamsrc.go index 89abca08..a3bec254 100644 --- a/internal/capture/streamsrc.go +++ b/internal/capture/streamsrc.go @@ -20,7 +20,7 @@ type StreamSrcManagerCtx struct { codecPipeline map[string]string // codec -> pipeline codec codec.RTPCodec - pipeline *gst.Pipeline + pipeline gst.Pipeline pipelineMu sync.Mutex pipelineStr string diff --git a/pkg/gst/gst.go b/pkg/gst/gst.go index e21afda1..3a9e686a 100644 --- a/pkg/gst/gst.go +++ b/pkg/gst/gst.go @@ -19,25 +19,44 @@ import ( "github.com/demodesk/neko/pkg/types" ) -type Pipeline struct { - id int - logger zerolog.Logger - Src string - Ctx *C.GstPipelineCtx - Sample chan types.Sample -} - -var pSerial int32 -var pipelines = make(map[int]*Pipeline) -var pipelinesLock sync.Mutex -var registry *C.GstRegistry +var ( + pSerial int32 + pipelines = make(map[int]*pipeline) + pipelinesLock sync.Mutex + registry *C.GstRegistry +) func init() { C.gst_init(nil, nil) registry = C.gst_registry_get() } -func CreatePipeline(pipelineStr string) (*Pipeline, error) { +type Pipeline interface { + Src() string + Sample() chan types.Sample + // attach sink or src to pipeline + AttachAppsink(sinkName string) + AttachAppsrc(srcName string) + // control pipeline lifecycle + Play() + Pause() + Destroy() + Push(buffer []byte) + // modify the property of a bin + SetPropInt(binName string, prop string, value int) bool + SetCapsFramerate(binName string, numerator, denominator int) bool + SetCapsResolution(binName string, width, height int) bool +} + +type pipeline struct { + id int + logger zerolog.Logger + src string + ctx *C.GstPipelineCtx + sample chan types.Sample +} + +func CreatePipeline(pipelineStr string) (Pipeline, error) { id := atomic.AddInt32(&pSerial, 1) pipelineStrUnsafe := C.CString(pipelineStr) @@ -54,63 +73,70 @@ func CreatePipeline(pipelineStr string) (*Pipeline, error) { return nil, fmt.Errorf("(pipeline error) %s", C.GoString(gstError.message)) } - p := &Pipeline{ + p := &pipeline{ id: int(id), logger: log.With(). Str("module", "capture"). Str("submodule", "gstreamer"). Int("pipeline_id", int(id)).Logger(), - Src: pipelineStr, - Ctx: ctx, - Sample: make(chan types.Sample), + src: pipelineStr, + ctx: ctx, + sample: make(chan types.Sample), } pipelines[p.id] = p return p, nil } -func (p *Pipeline) AttachAppsink(sinkName string) { +func (p *pipeline) Src() string { + return p.src +} + +func (p *pipeline) Sample() chan types.Sample { + return p.sample +} + +func (p *pipeline) AttachAppsink(sinkName string) { sinkNameUnsafe := C.CString(sinkName) defer C.free(unsafe.Pointer(sinkNameUnsafe)) - C.gstreamer_pipeline_attach_appsink(p.Ctx, sinkNameUnsafe) + C.gstreamer_pipeline_attach_appsink(p.ctx, sinkNameUnsafe) } -func (p *Pipeline) AttachAppsrc(srcName string) { +func (p *pipeline) AttachAppsrc(srcName string) { srcNameUnsafe := C.CString(srcName) defer C.free(unsafe.Pointer(srcNameUnsafe)) - C.gstreamer_pipeline_attach_appsrc(p.Ctx, srcNameUnsafe) + C.gstreamer_pipeline_attach_appsrc(p.ctx, srcNameUnsafe) } -func (p *Pipeline) Play() { - C.gstreamer_pipeline_play(p.Ctx) +func (p *pipeline) Play() { + C.gstreamer_pipeline_play(p.ctx) } -func (p *Pipeline) Pause() { - C.gstreamer_pipeline_pause(p.Ctx) +func (p *pipeline) Pause() { + C.gstreamer_pipeline_pause(p.ctx) } -func (p *Pipeline) Destroy() { - C.gstreamer_pipeline_destory(p.Ctx) +func (p *pipeline) Destroy() { + C.gstreamer_pipeline_destory(p.ctx) pipelinesLock.Lock() delete(pipelines, p.id) pipelinesLock.Unlock() - close(p.Sample) - C.free(unsafe.Pointer(p.Ctx)) - p = nil + close(p.sample) + C.free(unsafe.Pointer(p.ctx)) } -func (p *Pipeline) Push(buffer []byte) { +func (p *pipeline) Push(buffer []byte) { bytes := C.CBytes(buffer) defer C.free(bytes) - C.gstreamer_pipeline_push(p.Ctx, bytes, C.int(len(buffer))) + C.gstreamer_pipeline_push(p.ctx, bytes, C.int(len(buffer))) } -func (p *Pipeline) SetPropInt(binName string, prop string, value int) bool { +func (p *pipeline) SetPropInt(binName string, prop string, value int) bool { cBinName := C.CString(binName) defer C.free(unsafe.Pointer(cBinName)) @@ -121,11 +147,11 @@ func (p *Pipeline) SetPropInt(binName string, prop string, value int) bool { p.logger.Debug().Msgf("setting prop %s of %s to %d", prop, binName, value) - ok := C.gstreamer_pipeline_set_prop_int(p.Ctx, cBinName, cProp, cValue) + ok := C.gstreamer_pipeline_set_prop_int(p.ctx, cBinName, cProp, cValue) return ok == C.TRUE } -func (p *Pipeline) SetCapsFramerate(binName string, numerator, denominator int) bool { +func (p *pipeline) SetCapsFramerate(binName string, numerator, denominator int) bool { cBinName := C.CString(binName) cNumerator := C.int(numerator) cDenominator := C.int(denominator) @@ -134,11 +160,11 @@ func (p *Pipeline) SetCapsFramerate(binName string, numerator, denominator int) p.logger.Debug().Msgf("setting caps framerate of %s to %d/%d", binName, numerator, denominator) - ok := C.gstreamer_pipeline_set_caps_framerate(p.Ctx, cBinName, cNumerator, cDenominator) + ok := C.gstreamer_pipeline_set_caps_framerate(p.ctx, cBinName, cNumerator, cDenominator) return ok == C.TRUE } -func (p *Pipeline) SetCapsResolution(binName string, width, height int) bool { +func (p *pipeline) SetCapsResolution(binName string, width, height int) bool { cBinName := C.CString(binName) cWidth := C.int(width) cHeight := C.int(height) @@ -147,7 +173,7 @@ func (p *Pipeline) SetCapsResolution(binName string, width, height int) bool { p.logger.Debug().Msgf("setting caps resolution of %s to %dx%d", binName, width, height) - ok := C.gstreamer_pipeline_set_caps_resolution(p.Ctx, cBinName, cWidth, cHeight) + ok := C.gstreamer_pipeline_set_caps_resolution(p.ctx, cBinName, cWidth, cHeight) return ok == C.TRUE } @@ -175,7 +201,7 @@ func goHandlePipelineBuffer(buffer unsafe.Pointer, bufferLen C.int, duration C.i pipelinesLock.Unlock() if ok { - pipeline.Sample <- types.Sample{ + pipeline.sample <- types.Sample{ Data: C.GoBytes(buffer, bufferLen), Duration: time.Duration(duration), }