neko/pkg/gst/gst.go

159 lines
3.3 KiB
Go
Raw Normal View History

package gst
/*
#cgo pkg-config: gstreamer-1.0 gstreamer-app-1.0
#include "gst.h"
*/
import "C"
import (
"fmt"
"sync"
2021-12-06 09:31:40 +13:00
"sync/atomic"
2021-02-15 02:40:17 +13:00
"time"
"unsafe"
2021-12-06 05:52:25 +13:00
"github.com/rs/zerolog"
"github.com/rs/zerolog/log"
2022-03-20 23:43:00 +13:00
"gitlab.com/demodesk/neko/server/pkg/types"
)
type Pipeline struct {
2021-12-06 06:16:26 +13:00
id int
Src string
Ctx *C.GstPipelineCtx
Sample chan types.Sample
}
2021-12-06 09:31:40 +13:00
var pSerial int32
var pipelines = make(map[int]*Pipeline)
var pipelinesLock sync.Mutex
var registry *C.GstRegistry
func init() {
C.gstreamer_init()
2021-12-06 05:52:25 +13:00
go C.gstreamer_loop()
registry = C.gst_registry_get()
}
2021-02-02 11:50:18 +13:00
func CreatePipeline(pipelineStr string) (*Pipeline, error) {
2021-12-06 09:31:40 +13:00
id := atomic.AddInt32(&pSerial, 1)
pipelineStrUnsafe := C.CString(pipelineStr)
defer C.free(unsafe.Pointer(pipelineStrUnsafe))
pipelinesLock.Lock()
defer pipelinesLock.Unlock()
2021-12-06 06:16:26 +13:00
var gstError *C.GError
ctx := C.gstreamer_pipeline_create(pipelineStrUnsafe, C.int(id), &gstError)
2021-01-13 04:12:05 +13:00
2021-01-13 04:24:54 +13:00
if gstError != nil {
2021-01-15 02:15:17 +13:00
defer C.g_error_free(gstError)
2021-02-15 02:40:17 +13:00
return nil, fmt.Errorf("(pipeline error) %s", C.GoString(gstError.message))
2021-01-13 04:12:05 +13:00
}
p := &Pipeline{
2021-12-06 09:31:40 +13:00
id: int(id),
2021-12-06 06:16:26 +13:00
Src: pipelineStr,
Ctx: ctx,
Sample: make(chan types.Sample),
}
pipelines[p.id] = p
return p, nil
}
2021-11-29 10:19:06 +13:00
func (p *Pipeline) AttachAppsink(sinkName string) {
sinkNameUnsafe := C.CString(sinkName)
defer C.free(unsafe.Pointer(sinkNameUnsafe))
2021-12-06 06:16:26 +13:00
C.gstreamer_pipeline_attach_appsink(p.Ctx, sinkNameUnsafe)
}
2021-12-06 10:06:42 +13:00
func (p *Pipeline) AttachAppsrc(srcName string) {
srcNameUnsafe := C.CString(srcName)
defer C.free(unsafe.Pointer(srcNameUnsafe))
C.gstreamer_pipeline_attach_appsrc(p.Ctx, srcNameUnsafe)
}
func (p *Pipeline) Play() {
2021-12-06 06:16:26 +13:00
C.gstreamer_pipeline_play(p.Ctx)
}
func (p *Pipeline) Pause() {
C.gstreamer_pipeline_pause(p.Ctx)
}
2021-12-06 06:16:26 +13:00
func (p *Pipeline) Destroy() {
C.gstreamer_pipeline_destory(p.Ctx)
2021-12-06 09:31:40 +13:00
pipelinesLock.Lock()
delete(pipelines, p.id)
pipelinesLock.Unlock()
2021-12-06 06:16:26 +13:00
close(p.Sample)
2021-12-06 09:31:40 +13:00
C.free(unsafe.Pointer(p.Ctx))
p = nil
}
2021-12-06 10:06:42 +13:00
func (p *Pipeline) Push(buffer []byte) {
2021-11-29 10:37:17 +13:00
bytes := C.CBytes(buffer)
defer C.free(bytes)
2021-12-06 10:06:42 +13:00
C.gstreamer_pipeline_push(p.Ctx, bytes, C.int(len(buffer)))
2021-11-29 10:37:17 +13:00
}
// gst-inspect-1.0
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 {
return fmt.Errorf("required gstreamer plugin %s not found", pluginstr)
}
}
return nil
}
//export goHandlePipelineBuffer
func goHandlePipelineBuffer(buffer unsafe.Pointer, bufferLen C.int, duration C.int, pipelineID C.int) {
2021-01-23 06:13:32 +13:00
defer C.free(buffer)
pipelinesLock.Lock()
pipeline, ok := pipelines[int(pipelineID)]
pipelinesLock.Unlock()
if ok {
2021-01-23 06:13:32 +13:00
pipeline.Sample <- types.Sample{
2021-02-15 02:40:17 +13:00
Data: C.GoBytes(buffer, bufferLen),
2021-02-02 11:50:18 +13:00
Duration: time.Duration(duration),
2021-01-23 06:13:32 +13:00
}
} else {
2021-12-06 05:52:25 +13:00
log.Warn().
Str("module", "capture").
Str("submodule", "gstreamer").
2022-02-13 06:55:56 +13:00
Int("pipeline_id", int(pipelineID)).
2021-12-06 09:33:30 +13:00
Msgf("discarding sample, pipeline not found")
}
}
2021-12-06 05:52:25 +13:00
//export goPipelineLog
2021-12-06 06:16:26 +13:00
func goPipelineLog(levelUnsafe *C.char, msgUnsafe *C.char, pipelineID C.int) {
2021-12-06 05:52:25 +13:00
levelStr := C.GoString(levelUnsafe)
msg := C.GoString(msgUnsafe)
2021-12-06 09:33:30 +13:00
level, _ := zerolog.ParseLevel(levelStr)
log.WithLevel(level).
2021-12-06 05:52:25 +13:00
Str("module", "capture").
Str("submodule", "gstreamer").
2022-02-13 06:55:56 +13:00
Int("pipeline_id", int(pipelineID)).
2021-12-06 09:33:30 +13:00
Msg(msg)
2021-12-06 05:52:25 +13:00
}