iridium/neko
2
2
Fork 0
mirror of https://github.com/m1k1o/neko.git synced 2024-07-24 14:40:50 +12:00
Code Issues Packages Projects Releases Wiki Activity

remove go-events

Browse Source
This commit is contained in:
mbattista
2023-01-21 23:43:04 +01:00
committed by Miroslav Šedivý
parent cfc6bd417f
commit 5690a849e2
24 changed files with 16046 additions and 9271 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
server/internal/capture/gst/gst.go
View File

@ -51,6 +51,7 @@ func CreatePipeline(pipelineStr string) (*Pipeline, error) {
if gstError != nil {
defer C.g_error_free(gstError)
fmt.Printf("(pipeline error) %s", C.GoString(gstError.message))
return nil, fmt.Errorf("(pipeline error) %s", C.GoString(gstError.message))
}
@ -177,6 +178,7 @@ func goHandlePipelineBuffer(buffer unsafe.Pointer, bufferLen C.int, duration C.i
if ok {
pipeline.Sample <- types.Sample{
Data: C.GoBytes(buffer, bufferLen),
Timestamp: time.Now(),
Duration: time.Duration(duration),
}
} else {

53
server/internal/capture/manager.go
View File

@ -18,12 +18,12 @@ type CaptureManagerCtx struct {
broadcast *BroacastManagerCtx
audio *StreamSinkManagerCtx
video *StreamSinkManagerCtx
}
func New(desktop types.DesktopManager, config *config.Capture) *CaptureManagerCtx {
logger := log.With().Str("module", "capture").Logger()
return &CaptureManagerCtx{
manager := &CaptureManagerCtx{
logger: logger,
desktop: desktop,
@ -38,6 +38,10 @@ func New(desktop types.DesktopManager, config *config.Capture) *CaptureManagerCt
return NewVideoPipeline(config.VideoCodec, config.Display, config.VideoPipeline, config.VideoMaxFPS, config.VideoBitrate, config.VideoHWEnc)
}, "video"),
}
manager.Video().SetAdaptiveFramerate(config.VideoAdaptiveFramerate)
return manager
}
func (manager *CaptureManagerCtx) Start() {
@ -47,31 +51,38 @@ func (manager *CaptureManagerCtx) Start() {
}
}
manager.desktop.OnBeforeScreenSizeChange(func() {
if manager.video.Started() {
manager.video.destroyPipeline()
}
go func() {
for {
_ = <- manager.desktop.GetBeforeScreenSizeChangeChannel()
if manager.video.Started() {
manager.video.destroyPipeline()
}
if manager.broadcast.Started() {
manager.broadcast.destroyPipeline()
}
})
manager.desktop.OnAfterScreenSizeChange(func() {
if manager.video.Started() {
err := manager.video.createPipeline()
if err != nil && !errors.Is(err, types.ErrCapturePipelineAlreadyExists) {
manager.logger.Panic().Err(err).Msg("unable to recreate video pipeline")
if manager.broadcast.Started() {
manager.broadcast.destroyPipeline()
}
}
}()
if manager.broadcast.Started() {
err := manager.broadcast.createPipeline()
if err != nil && !errors.Is(err, types.ErrCapturePipelineAlreadyExists) {
manager.logger.Panic().Err(err).Msg("unable to recreate broadcast pipeline")
go func() {
for {
framerate := <- manager.desktop.GetAfterScreenSizeChangeChannel();
if manager.video.Started() {
manager.video.SetChangeFramerate(framerate)
err := manager.video.createPipeline()
if err != nil && !errors.Is(err, types.ErrCapturePipelineAlreadyExists) {
manager.logger.Panic().Err(err).Msg("unable to recreate video pipeline")
}
}
if manager.broadcast.Started() {
err := manager.broadcast.createPipeline()
if err != nil && !errors.Is(err, types.ErrCapturePipelineAlreadyExists) {
manager.logger.Panic().Err(err).Msg("unable to recreate broadcast pipeline")
}
}
}
})
}()
}
func (manager *CaptureManagerCtx) Shutdown() error {

25
server/internal/capture/pipelines.go
View File

@ -53,7 +53,7 @@ func NewBroadcastPipeline(device string, display string, pipelineSrc string, url
}
func NewVideoPipeline(rtpCodec codec.RTPCodec, display string, pipelineSrc string, fps int16, bitrate uint, hwenc string) (string, error) {
pipelineStr := " ! appsink name=appsink"
pipelineStr := " ! appsink name=appsinkvideo"
// if using custom pipeline
if pipelineSrc != "" {
@ -106,6 +106,27 @@ func NewVideoPipeline(rtpCodec codec.RTPCodec, display string, pipelineSrc strin
}
pipelineStr = fmt.Sprintf(videoSrc+"vp9enc target-bitrate=%d cpu-used=-5 threads=4 deadline=1 keyframe-max-dist=30 auto-alt-ref=true"+pipelineStr, display, fps, bitrate*1000)
case codec.AV1().Name:
// https://gstreamer.freedesktop.org/documentation/aom/av1enc.html?gi-language=c
// gstreamer1.0-plugins-bad
// av1enc usage-profile=1
if err := gst.CheckPlugins([]string{"ximagesrc", "vpx"}); err != nil {
return "", err
}
pipelineStr = strings.Join([]string{
fmt.Sprintf(videoSrc, display, fps),
"av1enc",
fmt.Sprintf("target-bitrate=%d", bitrate*650),
"cpu-used=4",
"end-usage=cbr",
// "usage-profile=realtime",
"undershoot=95",
"keyframe-max-dist=25",
"min-quantizer=4",
"max-quantizer=20",
pipelineStr,
}, " ")
case codec.H264().Name:
if err := gst.CheckPlugins([]string{"ximagesrc"}); err != nil {
return "", err
@ -149,7 +170,7 @@ func NewVideoPipeline(rtpCodec codec.RTPCodec, display string, pipelineSrc strin
}
func NewAudioPipeline(rtpCodec codec.RTPCodec, device string, pipelineSrc string, bitrate uint) (string, error) {
pipelineStr := " ! appsink name=appsink"
pipelineStr := " ! appsink name=appsinkaudio"
// if using custom pipeline
if pipelineSrc != "" {

53
server/internal/capture/streamsink.go
View File

@ -3,6 +3,8 @@ package capture
import (
"errors"
"sync"
"regexp"
"strconv"
"github.com/rs/zerolog"
"github.com/rs/zerolog/log"
@ -17,15 +19,17 @@ type StreamSinkManagerCtx struct {
mu sync.Mutex
wg sync.WaitGroup
codec codec.RTPCodec
pipeline *gst.Pipeline
pipelineMu sync.Mutex
pipelineFn func() (string, error)
codec codec.RTPCodec
pipeline *gst.Pipeline
pipelineMu sync.Mutex
pipelineFn func() (string, error)
adaptiveFramerate bool
listeners int
listenersMu sync.Mutex
sampleFn func(sample types.Sample)
changeFramerate int16
sampleChannel chan types.Sample
}
func streamSinkNew(codec codec.RTPCodec, pipelineFn func() (string, error), video_id string) *StreamSinkManagerCtx {
@ -35,9 +39,12 @@ func streamSinkNew(codec codec.RTPCodec, pipelineFn func() (string, error), vide
Str("video_id", video_id).Logger()
manager := &StreamSinkManagerCtx{
logger: logger,
codec: codec,
pipelineFn: pipelineFn,
logger: logger,
codec: codec,
pipelineFn: pipelineFn,
changeFramerate: 0,
adaptiveFramerate: false,
sampleChannel: make(chan types.Sample, 100),
}
return manager
@ -50,10 +57,6 @@ func (manager *StreamSinkManagerCtx) shutdown() {
manager.wg.Wait()
}
func (manager *StreamSinkManagerCtx) OnSample(listener func(sample types.Sample)) {
manager.sampleFn = listener
}
func (manager *StreamSinkManagerCtx) Codec() codec.RTPCodec {
return manager.codec
}
@ -142,6 +145,11 @@ func (manager *StreamSinkManagerCtx) createPipeline() error {
return err
}
if manager.changeFramerate > 0 && manager.adaptiveFramerate {
m1 := regexp.MustCompile(`framerate=\d+/1`)
pipelineStr = m1.ReplaceAllString(pipelineStr, "framerate=" + strconv.FormatInt(int64(manager.changeFramerate), 10) + "/1")
}
manager.logger.Info().
Str("codec", manager.codec.Name).
Str("src", pipelineStr).
@ -152,7 +160,12 @@ func (manager *StreamSinkManagerCtx) createPipeline() error {
return err
}
manager.pipeline.AttachAppsink("appsink")
appsinkSubfix := "audio"
if codec.IsVideo(manager.codec.Type) {
appsinkSubfix = "video"
}
manager.pipeline.AttachAppsink("appsink" + appsinkSubfix)
manager.pipeline.Play()
manager.wg.Add(1)
@ -169,7 +182,7 @@ func (manager *StreamSinkManagerCtx) createPipeline() error {
return
}
manager.sampleFn(sample)
manager.sampleChannel <- sample
}
}()
@ -188,3 +201,15 @@ func (manager *StreamSinkManagerCtx) destroyPipeline() {
manager.logger.Info().Msgf("destroying pipeline")
manager.pipeline = nil
}
func (manager *StreamSinkManagerCtx) GetSampleChannel() (chan types.Sample) {
return manager.sampleChannel
}
func (manager *StreamSinkManagerCtx) SetChangeFramerate(rate int16) {
manager.changeFramerate = rate
}
func (manager *StreamSinkManagerCtx) SetAdaptiveFramerate(allow bool) {
manager.adaptiveFramerate = allow
}

19
server/internal/config/capture.go
View File

@ -17,6 +17,7 @@ type Capture struct {
VideoBitrate uint // TODO: Pipeline builder.
VideoMaxFPS int16 // TODO: Pipeline builder.
VideoPipeline string
VideoAdaptiveFramerate bool
// audio
AudioDevice string
@ -56,6 +57,12 @@ func (Capture) Init(cmd *cobra.Command) error {
return err
}
// DEPRECATED: video codec
cmd.PersistentFlags().Bool("av1", false, "DEPRECATED: use video_codec")
if err := viper.BindPFlag("av1", cmd.PersistentFlags().Lookup("av1")); err != nil {
return err
}
// DEPRECATED: video codec
cmd.PersistentFlags().Bool("h264", false, "DEPRECATED: use video_codec")
if err := viper.BindPFlag("h264", cmd.PersistentFlags().Lookup("h264")); err != nil {
@ -82,6 +89,11 @@ func (Capture) Init(cmd *cobra.Command) error {
return err
}
cmd.PersistentFlags().Bool("adaptive_framerate", false, "use the framerate given from the source display")
if err := viper.BindPFlag("adaptive_framerate", cmd.PersistentFlags().Lookup("adaptive_framerate")); err != nil {
return err
}
//
// audio
//
@ -173,6 +185,9 @@ func (s *Capture) Set() {
} else if viper.GetBool("h264") {
s.VideoCodec = codec.H264()
log.Warn().Msg("you are using deprecated config setting 'NEKO_H264=true', use 'NEKO_VIDEO_CODEC=h264' instead")
} else if viper.GetBool("av1") {
s.VideoCodec = codec.AV1()
log.Warn().Msg("you are using deprecated config setting 'NEKO_AV1=true', use 'NEKO_VIDEO_CODEC=av1' instead")
}
videoHWEnc := ""
@ -183,7 +198,11 @@ func (s *Capture) Set() {
s.VideoBitrate = viper.GetUint("video_bitrate")
s.VideoMaxFPS = int16(viper.GetInt("max_fps"))
if s.VideoMaxFPS == 0 {
s.VideoMaxFPS = 60
}
s.VideoPipeline = viper.GetString("video")
s.VideoAdaptiveFramerate = viper.GetBool("adaptive_framerate")
//
// audio

44
server/internal/desktop/manager.go
View File

@ -9,7 +9,6 @@ import (
"m1k1o/neko/internal/desktop/xevent"
"m1k1o/neko/internal/desktop/xorg"
"github.com/kataras/go-events"
"github.com/rs/zerolog"
"github.com/rs/zerolog/log"
)
@ -17,18 +16,20 @@ import (
var mu = sync.Mutex{}
type DesktopManagerCtx struct {
logger zerolog.Logger
wg sync.WaitGroup
shutdown chan struct{}
emmiter events.EventEmmiter
config *config.Desktop
logger zerolog.Logger
wg sync.WaitGroup
shutdown chan struct{}
beforeScreenSizeChangeChannel chan bool
afterScreenSizeChangeChannel chan int16
config *config.Desktop
}
func New(config *config.Desktop) *DesktopManagerCtx {
return &DesktopManagerCtx{
logger: log.With().Str("module", "desktop").Logger(),
shutdown: make(chan struct{}),
emmiter: events.New(),
beforeScreenSizeChangeChannel: make (chan bool),
afterScreenSizeChangeChannel: make (chan int16),
config: config,
}
}
@ -47,14 +48,17 @@ func (manager *DesktopManagerCtx) Start() {
go xevent.EventLoop(manager.config.Display)
manager.OnEventError(func(error_code uint8, message string, request_code uint8, minor_code uint8) {
manager.logger.Warn().
Uint8("error_code", error_code).
Str("message", message).
Uint8("request_code", request_code).
Uint8("minor_code", minor_code).
go func() {
for {
desktopErrorMessage := <- xevent.EventErrorChannel
manager.logger.Warn().
Uint8("error_code", desktopErrorMessage.Error_code).
Str("message", desktopErrorMessage.Message).
Uint8("request_code", desktopErrorMessage.Request_code).
Uint8("minor_code", desktopErrorMessage.Minor_code).
Msg("X event error occurred")
})
}
}()
manager.wg.Add(1)
@ -75,16 +79,12 @@ func (manager *DesktopManagerCtx) Start() {
}()
}
func (manager *DesktopManagerCtx) OnBeforeScreenSizeChange(listener func()) {
manager.emmiter.On("before_screen_size_change", func(payload ...any) {
listener()
})
func (manager *DesktopManagerCtx) GetBeforeScreenSizeChangeChannel() (chan bool) {
return manager.beforeScreenSizeChangeChannel
}
func (manager *DesktopManagerCtx) OnAfterScreenSizeChange(listener func()) {
manager.emmiter.On("after_screen_size_change", func(payload ...any) {
listener()
})
func (manager *DesktopManagerCtx) GetAfterScreenSizeChangeChannel() (chan int16) {
return manager.afterScreenSizeChangeChannel
}
func (manager *DesktopManagerCtx) Shutdown() error {

35
server/internal/desktop/xevent.go
View File

@ -1,33 +1,26 @@
package desktop
import "m1k1o/neko/internal/desktop/xevent"
import (
"m1k1o/neko/internal/desktop/xevent"
"m1k1o/neko/internal/types"
)
func (manager *DesktopManagerCtx) OnCursorChanged(listener func(serial uint64)) {
xevent.Emmiter.On("cursor-changed", func(payload ...any) {
listener(payload[0].(uint64))
})
func (manager *DesktopManagerCtx) GetCursorChangedChannel() (chan uint64) {
return xevent.CursorChangedChannel
}
func (manager *DesktopManagerCtx) OnClipboardUpdated(listener func()) {
xevent.Emmiter.On("clipboard-updated", func(payload ...any) {
listener()
})
func (manager *DesktopManagerCtx) GetClipboardUpdatedChannel() (chan bool) {
return xevent.ClipboardUpdatedChannel
}
func (manager *DesktopManagerCtx) OnFileChooserDialogOpened(listener func()) {
xevent.Emmiter.On("file-chooser-dialog-opened", func(payload ...any) {
listener()
})
func (manager *DesktopManagerCtx) GetFileChooserDialogOpenedChannel() (chan bool) {
return xevent.FileChooserDialogOpenedChannel
}
func (manager *DesktopManagerCtx) OnFileChooserDialogClosed(listener func()) {
xevent.Emmiter.On("file-chooser-dialog-closed", func(payload ...any) {
listener()
})
func (manager *DesktopManagerCtx) GetFileChooserDialogClosedChannel() (chan bool) {
return xevent.FileChooserDialogClosedChannel
}
func (manager *DesktopManagerCtx) OnEventError(listener func(error_code uint8, message string, request_code uint8, minor_code uint8)) {
xevent.Emmiter.On("event-error", func(payload ...any) {
listener(payload[0].(uint8), payload[1].(string), payload[2].(uint8), payload[3].(uint8))
})
func (manager *DesktopManagerCtx) GetEventErrorChannel() (chan types.DesktopErrorMessage) {
return xevent.EventErrorChannel
}

38
server/internal/desktop/xevent/xevent.go
View File

@ -9,14 +9,38 @@ import "C"
import (
"unsafe"
"github.com/kataras/go-events"
"m1k1o/neko/internal/types"
)
var Emmiter events.EventEmmiter
var CursorChangedChannel chan uint64
var ClipboardUpdatedChannel chan bool
var FileChooserDialogClosedChannel chan bool
var FileChooserDialogOpenedChannel chan bool
var EventErrorChannel chan types.DesktopErrorMessage
func init() {
Emmiter = events.New()
CursorChangedChannel = make(chan uint64)
ClipboardUpdatedChannel = make(chan bool)
FileChooserDialogClosedChannel = make(chan bool)
FileChooserDialogOpenedChannel = make(chan bool)
EventErrorChannel = make(chan types.DesktopErrorMessage)
// Dummy goroutines since there is no consumer for the channel otherwise
go func() {
for {
_ = <-CursorChangedChannel
}
}()
go func() {
for {
_ = <-FileChooserDialogClosedChannel
}
}()
go func() {
for {
_ = <-FileChooserDialogOpenedChannel
}
}()
}
func EventLoop(display string) {
@ -28,12 +52,12 @@ func EventLoop(display string) {
//export goXEventCursorChanged
func goXEventCursorChanged(event C.XFixesCursorNotifyEvent) {
Emmiter.Emit("cursor-changed", uint64(event.cursor_serial))
CursorChangedChannel <- uint64(event.cursor_serial)
}
//export goXEventClipboardUpdated
func goXEventClipboardUpdated() {
Emmiter.Emit("clipboard-updated")
ClipboardUpdatedChannel <- true
}
//export goXEventConfigureNotify
@ -48,7 +72,7 @@ func goXEventUnmapNotify(window C.Window) {
//export goXEventError
func goXEventError(event *C.XErrorEvent, message *C.char) {
Emmiter.Emit("event-error", uint8(event.error_code), C.GoString(message), uint8(event.request_code), uint8(event.minor_code))
EventErrorChannel <- types.DesktopErrorMessage{ uint8(event.error_code), C.GoString(message), uint8(event.request_code), uint8(event.minor_code) }
}
//export goXEventActive

4
server/internal/desktop/xorg.go
View File

@ -72,10 +72,10 @@ func (manager *DesktopManagerCtx) ScreenConfigurations() map[int]types.ScreenCon
func (manager *DesktopManagerCtx) SetScreenSize(size types.ScreenSize) error {
mu.Lock()
manager.emmiter.Emit("before_screen_size_change")
manager.GetBeforeScreenSizeChangeChannel() <- true
defer func() {
manager.emmiter.Emit("after_screen_size_change")
manager.GetAfterScreenSizeChangeChannel() <- size.Rate
mu.Unlock()
}()

64
server/internal/session/manager.go
View File

@ -4,7 +4,6 @@ import (
"fmt"
"sync"
"github.com/kataras/go-events"
"github.com/rs/zerolog"
"github.com/rs/zerolog/log"
@ -17,20 +16,22 @@ func New(capture types.CaptureManager) *SessionManager {
logger: log.With().Str("module", "session").Logger(),
host: "",
capture: capture,
sessionChannel: make(chan types.SessionInformation, 10),
hostChannel: make(chan types.HostInformation, 10),
members: make(map[string]*Session),
emmiter: events.New(),
}
}
type SessionManager struct {
mu sync.Mutex
logger zerolog.Logger
host string
capture types.CaptureManager
members map[string]*Session
emmiter events.EventEmmiter
mu sync.Mutex
logger zerolog.Logger
host string
capture types.CaptureManager
members map[string]*Session
sessionChannel chan types.SessionInformation
hostChannel chan types.HostInformation
// TODO: Handle locks in sessions as flags.
controlLocked bool
controlLocked bool
}
func (manager *SessionManager) New(id string, admin bool, socket types.WebSocket) types.Session {
@ -49,7 +50,13 @@ func (manager *SessionManager) New(id string, admin bool, socket types.WebSocket
manager.capture.Video().AddListener()
manager.mu.Unlock()
manager.emmiter.Emit("created", id, session)
manager.sessionChannel <- types.SessionInformation{ "created", id, session }
go func() {
for {
_ = <- manager.hostChannel
}
}()
return session
}
@ -68,7 +75,8 @@ func (manager *SessionManager) SetHost(id string) error {
if ok {
manager.host = id
manager.emmiter.Emit("host", id)
manager.hostChannel <- types.HostInformation{ "host", id }
return nil
}
@ -86,7 +94,7 @@ func (manager *SessionManager) GetHost() (types.Session, bool) {
func (manager *SessionManager) ClearHost() {
id := manager.host
manager.host = ""
manager.emmiter.Emit("host_cleared", id)
manager.hostChannel <- types.HostInformation{ "host_cleared", id }
}
func (manager *SessionManager) Has(id string) bool {
@ -163,7 +171,7 @@ func (manager *SessionManager) Destroy(id string) {
manager.capture.Video().RemoveListener()
manager.mu.Unlock()
manager.emmiter.Emit("destroyed", id, session)
manager.sessionChannel <- types.SessionInformation{ "destroyed", id, session }
manager.logger.Err(err).Str("session_id", id).Msg("destroying session")
return
}
@ -221,32 +229,10 @@ func (manager *SessionManager) AdminBroadcast(v interface{}, exclude interface{}
return nil
}
func (manager *SessionManager) OnHost(listener func(id string)) {
manager.emmiter.On("host", func(payload ...interface{}) {
listener(payload[0].(string))
})
func (manager *SessionManager) GetSessionChannel() (chan types.SessionInformation) {
return manager.sessionChannel
}
func (manager *SessionManager) OnHostCleared(listener func(id string)) {
manager.emmiter.On("host_cleared", func(payload ...interface{}) {
listener(payload[0].(string))
})
}
func (manager *SessionManager) OnDestroy(listener func(id string, session types.Session)) {
manager.emmiter.On("destroyed", func(payload ...interface{}) {
listener(payload[0].(string), payload[1].(*Session))
})
}
func (manager *SessionManager) OnCreated(listener func(id string, session types.Session)) {
manager.emmiter.On("created", func(payload ...interface{}) {
listener(payload[0].(string), payload[1].(*Session))
})
}
func (manager *SessionManager) OnConnected(listener func(id string, session types.Session)) {
manager.emmiter.On("connected", func(payload ...interface{}) {
listener(payload[0].(string), payload[1].(*Session))
})
func (manager *SessionManager) GetHostChannel() (chan types.HostInformation) {
return manager.hostChannel
}

2
server/internal/session/session.go
View File

@ -78,7 +78,7 @@ func (session *Session) SetPeer(peer types.Peer) error {
func (session *Session) SetConnected(connected bool) error {
session.connected = connected
if connected {
session.manager.emmiter.Emit("connected", session.id, session)
session.manager.sessionChannel <- types.SessionInformation{ "connected", session.id, session }
}
return nil
}

4
server/internal/types/capture.go
View File

@ -19,13 +19,15 @@ type BroadcastManager interface {
type StreamSinkManager interface {
Codec() codec.RTPCodec
OnSample(listener func(sample Sample))
AddListener() error
RemoveListener() error
ListenersCount() int
Started() bool
GetSampleChannel() (chan Sample)
SetChangeFramerate(rate int16)
SetAdaptiveFramerate(allow bool)
}
type CaptureManager interface {

23
server/internal/types/codec/codecs.go
View File

@ -31,6 +31,8 @@ func ParseStr(codecName string) (codec RTPCodec, ok bool) {
codec = VP8()
case VP9().Name:
codec = VP9()
case AV1().Name:
codec = AV1()
case H264().Name:
codec = H264()
case Opus().Name:
@ -48,6 +50,12 @@ func ParseStr(codecName string) (codec RTPCodec, ok bool) {
return
}
func IsVideo(codecType webrtc.RTPCodecType) (ok bool) {
ok = codecType == webrtc.RTPCodecTypeVideo
return
}
type RTPCodec struct {
Name string
PayloadType webrtc.PayloadType
@ -108,6 +116,21 @@ func H264() RTPCodec {
},
}
}
// TODO: Profile ID.
func AV1() RTPCodec {
return RTPCodec{
Name: "av1",
PayloadType: 96,
Type: webrtc.RTPCodecTypeVideo,
Capability: webrtc.RTPCodecCapability{
MimeType: webrtc.MimeTypeAV1,
ClockRate: 90000,
Channels: 0,
SDPFmtpLine: "",
RTCPFeedback: RTCPFeedback,
},
}
}
func Opus() RTPCodec {
return RTPCodec{

21
server/internal/types/desktop.go
View File

@ -33,11 +33,18 @@ type KeyboardMap struct {
Variant string
}
type DesktopErrorMessage struct {
Error_code uint8
Message string
Request_code uint8
Minor_code uint8
}
type DesktopManager interface {
Start()
Shutdown() error
OnBeforeScreenSizeChange(listener func())
OnAfterScreenSizeChange(listener func())
GetBeforeScreenSizeChangeChannel() (chan bool)
GetAfterScreenSizeChangeChannel() (chan int16)
// clipboard
ReadClipboard() string
@ -65,9 +72,9 @@ type DesktopManager interface {
GetScreenshotImage() *image.RGBA
// xevent
OnCursorChanged(listener func(serial uint64))
OnClipboardUpdated(listener func())
OnFileChooserDialogOpened(listener func())
OnFileChooserDialogClosed(listener func())
OnEventError(listener func(error_code uint8, message string, request_code uint8, minor_code uint8))
GetCursorChangedChannel() (chan uint64)
GetClipboardUpdatedChannel() (chan bool)
GetFileChooserDialogOpenedChannel() (chan bool)
GetFileChooserDialogClosedChannel() (chan bool)
GetEventErrorChannel() (chan DesktopErrorMessage)
}

18
server/internal/types/session.go
View File

@ -7,6 +7,17 @@ type Member struct {
Muted bool `json:"muted"`
}
type SessionInformation struct {
Type string
Id string
Session Session
}
type HostInformation struct {
Type string
Id string
}
type Session interface {
ID() string
Name() string
@ -46,9 +57,6 @@ type SessionManager interface {
Clear() error
Broadcast(v interface{}, exclude interface{}) error
AdminBroadcast(v interface{}, exclude interface{}) error
OnHost(listener func(id string))
OnHostCleared(listener func(id string))
OnDestroy(listener func(id string, session Session))
OnCreated(listener func(id string, session Session))
OnConnected(listener func(id string, session Session))
GetSessionChannel() (chan SessionInformation)
GetHostChannel() (chan HostInformation)
}

26
server/internal/webrtc/webrtc.go
View File

@ -55,12 +55,15 @@ func (manager *WebRTCManager) Start() {
manager.logger.Panic().Err(err).Msg("unable to create audio track")
}
manager.capture.Audio().OnSample(func(sample types.Sample) {
err := manager.audioTrack.WriteSample(media.Sample(sample))
if err != nil && errors.Is(err, io.ErrClosedPipe) {
manager.logger.Warn().Err(err).Msg("audio pipeline failed to write")
go func() {
for {
newSample := <- manager.capture.Audio().GetSampleChannel()
err := manager.audioTrack.WriteSample(media.Sample(newSample))
if err != nil && errors.Is(err, io.ErrClosedPipe) {
manager.logger.Warn().Err(err).Msg("audio pipeline failed to write")
}
}
})
}()
//
// video
@ -72,12 +75,15 @@ func (manager *WebRTCManager) Start() {
manager.logger.Panic().Err(err).Msg("unable to create video track")
}
manager.capture.Video().OnSample(func(sample types.Sample) {
err := manager.videoTrack.WriteSample(media.Sample(sample))
if err != nil && errors.Is(err, io.ErrClosedPipe) {
manager.logger.Warn().Err(err).Msg("video pipeline failed to write")
go func() {
for {
newSample := <- manager.capture.Video().GetSampleChannel()
err := manager.videoTrack.WriteSample(media.Sample(newSample))
if err != nil && errors.Is(err, io.ErrClosedPipe) {
manager.logger.Warn().Err(err).Msg("video pipeline failed to write")
}
}
})
}()
//
// api

182
server/internal/websocket/websocket.go
View File

@ -101,102 +101,108 @@ type WebSocketHandler struct {
}
func (ws *WebSocketHandler) Start() {
ws.sessions.OnCreated(func(id string, session types.Session) {
if err := ws.handler.SessionCreated(id, session); err != nil {
ws.logger.Warn().Str("id", id).Err(err).Msg("session created with and error")
} else {
ws.logger.Debug().Str("id", id).Msg("session created")
}
})
go func () {
for {
channelMessage := <- ws.sessions.GetSessionChannel()
ws.sessions.OnConnected(func(id string, session types.Session) {
if err := ws.handler.SessionConnected(id, session); err != nil {
ws.logger.Warn().Str("id", id).Err(err).Msg("session connected with and error")
} else {
ws.logger.Debug().Str("id", id).Msg("session connected")
}
switch channelMessage.Type {
case "created":
if err := ws.handler.SessionCreated(channelMessage.Id, channelMessage.Session); err != nil {
ws.logger.Warn().Str("id", channelMessage.Id).Err(err).Msg("session created with and error")
} else {
ws.logger.Debug().Str("id", channelMessage.Id).Msg("session created")
}
case "connected":
if err := ws.handler.SessionConnected(channelMessage.Id, channelMessage.Session); err != nil {
ws.logger.Warn().Str("id", channelMessage.Id).Err(err).Msg("session connected with and error")
} else {
ws.logger.Debug().Str("id", channelMessage.Id).Msg("session connected")
}
// if control protection is enabled and at least one admin
// and if room was locked on behalf control protection, unlock
sess, ok := ws.state.GetLocked("control")
if ok && ws.conf.ControlProtection && sess == CONTROL_PROTECTION_SESSION && len(ws.sessions.Admins()) > 0 {
ws.state.Unlock("control")
ws.sessions.SetControlLocked(false) // TODO: Handle locks in sessions as flags.
ws.logger.Info().Msgf("control unlocked on behalf of control protection")
// if control protection is enabled and at least one admin
// and if room was locked on behalf control protection, unlock
sess, ok := ws.state.GetLocked("control")
if ok && ws.conf.ControlProtection && sess == CONTROL_PROTECTION_SESSION && len(ws.sessions.Admins()) > 0 {
ws.state.Unlock("control")
ws.sessions.SetControlLocked(false) // TODO: Handle locks in sessions as flags.
ws.logger.Info().Msgf("control unlocked on behalf of control protection")
if err := ws.sessions.Broadcast(
message.AdminLock{
Event: event.ADMIN_UNLOCK,
ID: id,
Resource: "control",
}, nil); err != nil {
ws.logger.Warn().Err(err).Msgf("broadcasting event %s has failed", event.ADMIN_UNLOCK)
if err := ws.sessions.Broadcast(
message.AdminLock{
Event: event.ADMIN_UNLOCK,
ID: channelMessage.Id,
Resource: "control",
}, nil); err != nil {
ws.logger.Warn().Err(err).Msgf("broadcasting event %s has failed", event.ADMIN_UNLOCK)
}
}
// remove outdated stats
if channelMessage.Session.Admin() {
ws.lastAdminLeftAt = nil
} else {
ws.lastUserLeftAt = nil
}
case "destroyed":
if err := ws.handler.SessionDestroyed(channelMessage.Id); err != nil {
ws.logger.Warn().Str("id", channelMessage.Id).Err(err).Msg("session destroyed with and error")
} else {
ws.logger.Debug().Str("id", channelMessage.Id).Msg("session destroyed")
}
membersCount := len(ws.sessions.Members())
adminCount := len(ws.sessions.Admins())
// if control protection is enabled and no admin
// and room is not locked, lock
ok := ws.state.IsLocked("control")
if !ok && ws.conf.ControlProtection && adminCount == 0 {
ws.state.Lock("control", CONTROL_PROTECTION_SESSION)
ws.sessions.SetControlLocked(true) // TODO: Handle locks in sessions as flags.
ws.logger.Info().Msgf("control locked and released on behalf of control protection")
ws.handler.AdminRelease(channelMessage.Id, channelMessage.Session)
if err := ws.sessions.Broadcast(
message.AdminLock{
Event: event.ADMIN_LOCK,
ID: channelMessage.Id,
Resource: "control",
}, nil); err != nil {
ws.logger.Warn().Err(err).Msgf("broadcasting event %s has failed", event.ADMIN_LOCK)
}
}
// if this was the last admin
if channelMessage.Session.Admin() && adminCount == 0 {
now := time.Now()
ws.lastAdminLeftAt = &now
}
// if this was the last user
if !channelMessage.Session.Admin() && membersCount-adminCount == 0 {
now := time.Now()
ws.lastUserLeftAt = &now
}
}
}
}()
// remove outdated stats
if session.Admin() {
ws.lastAdminLeftAt = nil
} else {
ws.lastUserLeftAt = nil
}
})
ws.sessions.OnDestroy(func(id string, session types.Session) {
if err := ws.handler.SessionDestroyed(id); err != nil {
ws.logger.Warn().Str("id", id).Err(err).Msg("session destroyed with and error")
} else {
ws.logger.Debug().Str("id", id).Msg("session destroyed")
}
membersCount := len(ws.sessions.Members())
adminCount := len(ws.sessions.Admins())
// if control protection is enabled and no admin
// and room is not locked, lock
ok := ws.state.IsLocked("control")
if !ok && ws.conf.ControlProtection && adminCount == 0 {
ws.state.Lock("control", CONTROL_PROTECTION_SESSION)
ws.sessions.SetControlLocked(true) // TODO: Handle locks in sessions as flags.
ws.logger.Info().Msgf("control locked and released on behalf of control protection")
ws.handler.AdminRelease(id, session)
if err := ws.sessions.Broadcast(
message.AdminLock{
Event: event.ADMIN_LOCK,
ID: id,
Resource: "control",
}, nil); err != nil {
ws.logger.Warn().Err(err).Msgf("broadcasting event %s has failed", event.ADMIN_LOCK)
go func() {
for {
_ = <- ws.desktop.GetClipboardUpdatedChannel()
session, ok := ws.sessions.GetHost()
if !ok {
return
}
err := session.Send(message.Clipboard{
Event: event.CONTROL_CLIPBOARD,
Text: ws.desktop.ReadClipboard(),
})
ws.logger.Err(err).Msg("sync clipboard")
}
// if this was the last admin
if session.Admin() && adminCount == 0 {
now := time.Now()
ws.lastAdminLeftAt = &now
}
// if this was the last user
if !session.Admin() && membersCount-adminCount == 0 {
now := time.Now()
ws.lastUserLeftAt = &now
}
})
ws.desktop.OnClipboardUpdated(func() {
session, ok := ws.sessions.GetHost()
if !ok {
return
}
err := session.Send(message.Clipboard{
Event: event.CONTROL_CLIPBOARD,
Text: ws.desktop.ReadClipboard(),
})
ws.logger.Err(err).Msg("sync clipboard")
})
}()
// watch for file changes and send file list if file transfer is enabled
if ws.conf.FileTransferEnabled {