neko/server/internal/capture/gst/gst.go

222 lines
5.0 KiB
Go
Raw Permalink Normal View History

2020-01-13 21:05:38 +13:00
package gst
/*
#cgo pkg-config: gstreamer-1.0 gstreamer-app-1.0
#include "gst.h"
*/
import "C"
import (
"fmt"
"sync"
2022-09-17 22:43:17 +12:00
"sync/atomic"
2021-02-15 05:30:24 +13:00
"time"
2020-01-13 21:05:38 +13:00
"unsafe"
2022-09-17 22:43:17 +12:00
"github.com/rs/zerolog"
"github.com/rs/zerolog/log"
2021-10-06 09:38:24 +13:00
"m1k1o/neko/internal/types"
2020-01-13 21:05:38 +13:00
)
type Pipeline struct {
2022-09-17 22:43:17 +12:00
id int
logger zerolog.Logger
Src string
Ctx *C.GstPipelineCtx
Sample chan types.Sample
2020-01-13 21:05:38 +13:00
}
2022-09-17 22:43:17 +12:00
var pSerial int32
2020-01-13 21:05:38 +13:00
var pipelines = make(map[int]*Pipeline)
var pipelinesLock sync.Mutex
2020-01-19 12:30:09 +13:00
var registry *C.GstRegistry
var gMainLoop *C.GMainLoop
2020-01-13 21:05:38 +13:00
2020-01-19 12:30:09 +13:00
func init() {
2022-09-17 22:43:17 +12:00
C.gst_init(nil, nil)
2020-01-19 12:30:09 +13:00
registry = C.gst_registry_get()
}
func RunMainLoop() {
if gMainLoop != nil {
return
}
gMainLoop = C.g_main_loop_new(nil, C.int(0))
C.g_main_loop_run(gMainLoop)
}
func QuitMainLoop() {
if gMainLoop == nil {
return
}
C.g_main_loop_quit(gMainLoop)
gMainLoop = nil
}
func CreatePipeline(pipelineStr string) (*Pipeline, error) {
2022-09-17 22:43:17 +12:00
id := atomic.AddInt32(&pSerial, 1)
2020-01-13 21:05:38 +13:00
pipelineStrUnsafe := C.CString(pipelineStr)
defer C.free(unsafe.Pointer(pipelineStrUnsafe))
pipelinesLock.Lock()
defer pipelinesLock.Unlock()
2022-09-17 22:43:17 +12:00
var gstError *C.GError
ctx := C.gstreamer_pipeline_create(pipelineStrUnsafe, C.int(id), &gstError)
if gstError != nil {
defer C.g_error_free(gstError)
2023-01-22 11:43:04 +13:00
fmt.Printf("(pipeline error) %s", C.GoString(gstError.message))
2022-09-17 22:43:17 +12:00
return nil, fmt.Errorf("(pipeline error) %s", C.GoString(gstError.message))
2021-08-16 01:37:27 +12:00
}
2020-02-11 18:15:59 +13:00
p := &Pipeline{
2022-09-17 22:43:17 +12:00
id: int(id),
logger: log.With().
Str("module", "capture").
Str("submodule", "gstreamer").
Int("pipeline_id", int(id)).Logger(),
Src: pipelineStr,
Ctx: ctx,
2020-01-13 21:05:38 +13:00
}
2020-02-11 18:15:59 +13:00
pipelines[p.id] = p
return p, nil
2020-01-13 21:05:38 +13:00
}
func (p *Pipeline) AttachAppsink(sinkName string, sampleChannel chan types.Sample) {
2022-09-17 22:43:17 +12:00
sinkNameUnsafe := C.CString(sinkName)
defer C.free(unsafe.Pointer(sinkNameUnsafe))
p.Sample = sampleChannel
2022-09-17 22:43:17 +12:00
C.gstreamer_pipeline_attach_appsink(p.Ctx, sinkNameUnsafe)
}
func (p *Pipeline) AttachAppsrc(srcName string) {
srcNameUnsafe := C.CString(srcName)
defer C.free(unsafe.Pointer(srcNameUnsafe))
C.gstreamer_pipeline_attach_appsrc(p.Ctx, srcNameUnsafe)
2020-01-13 21:05:38 +13:00
}
2020-09-24 18:09:02 +12:00
func (p *Pipeline) Play() {
2022-09-17 22:43:17 +12:00
C.gstreamer_pipeline_play(p.Ctx)
}
func (p *Pipeline) Pause() {
C.gstreamer_pipeline_pause(p.Ctx)
}
func (p *Pipeline) Destroy() {
C.gstreamer_pipeline_destory(p.Ctx)
pipelinesLock.Lock()
delete(pipelines, p.id)
pipelinesLock.Unlock()
C.free(unsafe.Pointer(p.Ctx))
p = nil
}
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)))
2020-09-24 18:09:02 +12:00
}
2022-09-17 22:43:17 +12:00
func (p *Pipeline) SetPropInt(binName string, prop string, value int) bool {
cBinName := C.CString(binName)
defer C.free(unsafe.Pointer(cBinName))
cProp := C.CString(prop)
defer C.free(unsafe.Pointer(cProp))
cValue := C.int(value)
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)
return ok == C.TRUE
}
func (p *Pipeline) SetCapsFramerate(binName string, numerator, denominator int) bool {
cBinName := C.CString(binName)
cNumerator := C.int(numerator)
cDenominator := C.int(denominator)
defer C.free(unsafe.Pointer(cBinName))
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)
return ok == C.TRUE
}
func (p *Pipeline) SetCapsResolution(binName string, width, height int) bool {
cBinName := C.CString(binName)
cWidth := C.int(width)
cHeight := C.int(height)
defer C.free(unsafe.Pointer(cBinName))
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)
return ok == C.TRUE
2020-01-13 21:05:38 +13:00
}
2020-01-27 14:28:39 +13:00
// gst-inspect-1.0
2020-01-19 12:30:09 +13:00
func CheckPlugins(plugins []string) error {
var plugin *C.GstPlugin
for _, pluginstr := range plugins {
plugincstr := C.CString(pluginstr)
plugin = C.gst_registry_find_plugin(registry, plugincstr)
C.free(unsafe.Pointer(plugincstr))
if plugin == nil {
2020-01-25 04:47:37 +13:00
return fmt.Errorf("required gstreamer plugin %s not found", pluginstr)
2020-01-19 12:30:09 +13:00
}
}
return nil
}
2020-01-13 21:05:38 +13:00
//export goHandlePipelineBuffer
func goHandlePipelineBuffer(buffer unsafe.Pointer, bufferLen C.int, duration C.int, pipelineID C.int) {
2022-09-17 22:43:17 +12:00
defer C.free(buffer)
2020-01-13 21:05:38 +13:00
pipelinesLock.Lock()
pipeline, ok := pipelines[int(pipelineID)]
pipelinesLock.Unlock()
if ok {
2022-09-17 22:43:17 +12:00
pipeline.Sample <- types.Sample{
2023-01-29 10:08:36 +13:00
Data: C.GoBytes(buffer, bufferLen),
Timestamp: time.Now(),
Duration: time.Duration(duration),
2022-09-17 22:43:17 +12:00
}
2020-01-13 21:05:38 +13:00
} else {
2022-09-17 22:43:17 +12:00
log.Warn().
Str("module", "capture").
Str("submodule", "gstreamer").
Int("pipeline_id", int(pipelineID)).
Msgf("discarding sample, pipeline not found")
2020-01-13 21:05:38 +13:00
}
2022-09-17 22:43:17 +12:00
}
//export goPipelineLog
func goPipelineLog(levelUnsafe *C.char, msgUnsafe *C.char, pipelineID C.int) {
levelStr := C.GoString(levelUnsafe)
msg := C.GoString(msgUnsafe)
level, _ := zerolog.ParseLevel(levelStr)
log.WithLevel(level).
Str("module", "capture").
Str("submodule", "gstreamer").
Int("pipeline_id", int(pipelineID)).
Msg(msg)
2020-01-13 21:05:38 +13:00
}