cursor image/position from WebRTC.

internal/desktop/xorg.go

@@ -9,16 +9,21 @@ import (
 	"demodesk/neko/internal/desktop/xorg"
 )
 
-var mousePosX int
+// TODO: Refactor.
-var mousePosY int
+var cursorListeners []func(x, y int)
 
 func (manager *DesktopManagerCtx) Move(x, y int) {
 	xorg.Move(x, y)
-	mousePosX, mousePosY = x, y
+
+	// TODO: Refactor.
+	for _, listener := range cursorListeners {
+		listener(x, y)
+	}
 }
 
-func (manager *DesktopManagerCtx) GetMousePositon() (x, y int) {
+// TODO: Refactor.
-	return mousePosX, mousePosY
+func (manager *DesktopManagerCtx) OnCursorPosition(listener func(x, y int)) {
+	cursorListeners = append(cursorListeners, listener)
 }
 
 func (manager *DesktopManagerCtx) Scroll(x, y int) {

internal/types/desktop.go

@@ -48,7 +48,7 @@ type DesktopManager interface {
 
 	// xorg
 	Move(x, y int)
-	GetMousePositon() (x, y int)
+	OnCursorPosition(listener func(x, y int))
 	Scroll(x, y int)
 	ButtonDown(code int) error
 	KeyDown(code uint64) error

internal/types/event/events.go

@@ -46,11 +46,6 @@ const (
 	KEYBOARD_MAP       = "keyboard/map"
 )
 
-const (
-	CURSOR_IMAGE    = "cursor/image"
-	CURSOR_POSITION = "cursor/position"
-)
-
 const (
 	BORADCAST_STATUS = "broadcast/status"
 )

internal/types/message/messages.go

@@ -22,7 +22,6 @@ type SystemInit struct {
 	ScreenSize      ScreenSize            `json:"screen_size"`
 	Members         map[string]MemberData `json:"members"`
 	ImplicitHosting bool                  `json:"implicit_hosting"`
-	CursorImage     *CursorImage          `json:"cursor_image"`
 }
 
 type SystemAdmin struct {
@@ -159,26 +158,6 @@ type KeyboardModifiers struct {
 	ScrollLock *bool  `json:"scroll_lock"`	
 }
 
-/////////////////////////////
-// Cursor
-/////////////////////////////
-
-type CursorImage struct {
-	Event  string `json:"event,omitempty"`
-	Uri    string `json:"uri"`
-	Width  uint16 `json:"width"`
-	Height uint16 `json:"height"`
-	X      uint16 `json:"x"`
-	Y      uint16 `json:"y"`
-}
-
-type CursorPosition struct {
-	Event     string `json:"event,omitempty"`
-	MemberId  string `json:"member_id"`
-	X         uint16 `json:"x"`
-	Y         uint16 `json:"y"`
-}
-
 /////////////////////////////
 // Broadcast
 /////////////////////////////

internal/webrtc/manager.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"io"
 	"strings"
+	"reflect"
 
 	"github.com/pion/webrtc/v3"
 	"github.com/pion/webrtc/v3/pkg/media"
@@ -19,21 +20,27 @@ import (
 
 func New(desktop types.DesktopManager, capture types.CaptureManager, config *config.WebRTC) *WebRTCManagerCtx {
 	return &WebRTCManagerCtx{
-		logger:   log.With().Str("module", "webrtc").Logger(),
+		logger:           log.With().Str("module", "webrtc").Logger(),
-		desktop:  desktop,
+		desktop:          desktop,
-		capture:  capture,
+		capture:          capture,
-		config:   config,
+		config:           config,
+		// TODO: Refactor.
+		curImgListeners:  map[uintptr]*func(cur *types.CursorImage){},
+		curPosListeners:  map[uintptr]*func(x, y int){},
 	}
 }
 
 type WebRTCManagerCtx struct {
-	logger       zerolog.Logger
+	logger          zerolog.Logger
-	audioTrack   *webrtc.TrackLocalStaticSample
+	audioTrack      *webrtc.TrackLocalStaticSample
-	audioCodec   codec.RTPCodec
+	audioCodec      codec.RTPCodec
-	audioStop    func()
+	audioStop       func()
-	desktop      types.DesktopManager
+	desktop         types.DesktopManager
-	capture      types.CaptureManager
+	capture         types.CaptureManager
-	config       *config.WebRTC
+	config          *config.WebRTC
+	// TODO: Refactor.
+	curImgListeners map[uintptr]*func(cur *types.CursorImage)
+	curPosListeners map[uintptr]*func(x, y int)
 }
 
 func (manager *WebRTCManagerCtx) Start() {
@@ -64,6 +71,21 @@ func (manager *WebRTCManagerCtx) Start() {
 		Str("ephemeral_port_range", fmt.Sprintf("%d-%d", manager.config.EphemeralMin, manager.config.EphemeralMax)).
 		Str("nat_ips", strings.Join(manager.config.NAT1To1IPs, ",")).
 		Msgf("webrtc starting")
+
+	// TODO: Refactor.
+	manager.desktop.OnCursorChanged(func(serial uint64) {
+		cur := manager.desktop.GetCursorImage()
+		for _, emit := range manager.curImgListeners {
+			(*emit)(cur)
+		}
+	})
+
+	// TODO: Refactor.
+	manager.desktop.OnCursorPosition(func(x, y int) {
+		for _, emit := range manager.curPosListeners {
+			(*emit)(x, y)
+		}
+	})
 }
 
 func (manager *WebRTCManagerCtx) Shutdown() error {
@@ -213,6 +235,33 @@ func (manager *WebRTCManagerCtx) CreatePeer(session types.Session, videoID strin
 		}
 	}
 
+	peer := &WebRTCPeerCtx{
+		api:          api,
+		connection:   connection,
+		changeVideo:  changeVideo,
+		dataChannel:  dataChannel,
+	}
+
+	cursorChange := func(cur *types.CursorImage) {
+		if err := peer.SendCursorImage(cur); err != nil {
+			manager.logger.Warn().Err(err).Msg("could not send cursor image")
+		}
+	}
+
+	cursorPosition := func(x, y int) {
+		if session.IsHost() {
+			return
+		}
+
+		if err := peer.SendCursorPosition(x, y); err != nil {
+			manager.logger.Warn().Err(err).Msg("could not send cursor position")
+		}
+	}
+
+	// TODO: Refactor.
+	cursorChangePtr := reflect.ValueOf(&cursorChange).Pointer()
+	cursorPositionPtr := reflect.ValueOf(&cursorPosition).Pointer()
+
 	connection.OnConnectionStateChange(func(state webrtc.PeerConnectionState) {
 		switch state {
 		case webrtc.PeerConnectionStateConnected:
@@ -229,9 +278,19 @@ func (manager *WebRTCManagerCtx) CreatePeer(session types.Session, videoID strin
 			if videoStream.ListenersCount() == 0 {
 				videoStream.Stop()
 			}
+
+			// TODO: Refactor.
+			delete(manager.curImgListeners, cursorChangePtr)
+			delete(manager.curPosListeners, cursorPositionPtr)
 		}
 	})
 
+	dataChannel.OnOpen(func() {
+		// TODO: Refactor.
+		manager.curImgListeners[cursorChangePtr] = &cursorChange
+		manager.curPosListeners[cursorPositionPtr] = &cursorPosition
+	})
+
 	dataChannel.OnMessage(func(message webrtc.DataChannelMessage) {
 		if !session.IsHost() {
 			return
@@ -242,13 +301,7 @@ func (manager *WebRTCManagerCtx) CreatePeer(session types.Session, videoID strin
 		}
 	})
 
-	session.SetWebRTCPeer(&WebRTCPeerCtx{
+	session.SetWebRTCPeer(peer)
-		api:          api,
-		connection:   connection,
-		changeVideo:  changeVideo,
-		dataChannel:  dataChannel,
-	})
-
 	return connection.LocalDescription(), nil
 }
 

internal/websocket/handler/system.go

@@ -4,7 +4,6 @@ import (
 	"demodesk/neko/internal/types"
 	"demodesk/neko/internal/types/event"
 	"demodesk/neko/internal/types/message"
-	"demodesk/neko/internal/utils"
 )
 
 func (h *MessageHandlerCtx) systemInit(session types.Session) error {
@@ -33,20 +32,6 @@ func (h *MessageHandlerCtx) systemInit(session types.Session) error {
 		}
 	}
 
-	var cursorImage *message.CursorImage
-	cur := h.desktop.GetCursorImage()
-	uri, err := utils.GetCursorImageURI(cur)
-	if err == nil {
-		cursorImage = &message.CursorImage{
-			Event:  event.CURSOR_IMAGE,
-			Uri:    uri,
-			Width:  cur.Width,
-			Height: cur.Height,
-			X:      cur.Xhot,
-			Y:      cur.Yhot,
-		}
-	}
-
 	return session.Send(
 		message.SystemInit{
 			Event:           event.SYSTEM_INIT,
@@ -59,7 +44,6 @@ func (h *MessageHandlerCtx) systemInit(session types.Session) error {
 			},
 			Members:         members,
 			ImplicitHosting: h.sessions.ImplicitHosting(),
-			CursorImage:     cursorImage,
 		})
 }
 

internal/websocket/manager.go

@@ -12,7 +12,6 @@ import (
 	"demodesk/neko/internal/types"
 	"demodesk/neko/internal/types/event"
 	"demodesk/neko/internal/types/message"
-	"demodesk/neko/internal/utils"
 )
 
 func New(
@@ -93,25 +92,6 @@ func (ws *WebSocketManagerCtx) Start() {
 		}
 	})
 
-	// TOOD: Throttle events.
-	ws.desktop.OnCursorChanged(func(serial uint64) {
-		cur := ws.desktop.GetCursorImage()
-		uri, err := utils.GetCursorImageURI(cur)
-		if err != nil {
-			ws.logger.Warn().Err(err).Msg("could create cursor image")
-			return
-		}
-
-		ws.sessions.Broadcast(message.CursorImage{
-			Event:  event.CURSOR_IMAGE,
-			Uri:    uri,
-			Width:  cur.Width,
-			Height: cur.Height,
-			X:      cur.Xhot,
-			Y:      cur.Yhot,
-		}, nil)
-	})
-
 	ws.desktop.OnClipboardUpdated(func() {
 		session := ws.sessions.GetHost()
 		if session == nil || !session.CanAccessClipboard() {
@@ -134,47 +114,6 @@ func (ws *WebSocketManagerCtx) Start() {
 	})
 
 	ws.fileChooserDialogEvents()
-
-	go func() {
-		ws.logger.Debug().Msg("cursor position broadcast start")
-
-		defer func() {
-			ws.logger.Debug().Msg("cursor position broadcast shutdown")
-		}()
-
-		var posX int
-		var posY int
-
-		for {
-			select {
-			case <-ws.shutdown:
-				return
-			default:
-				time.Sleep(40 * time.Millisecond)
-
-				session := ws.sessions.GetHost()
-				if session == nil {
-					continue
-				}
-
-				x, y := ws.desktop.GetMousePositon()
-				if posX == x && posY == y {
-					continue
-				}
-
-				memberId := session.ID()
-				posX = x
-				posY = y
-			
-				ws.sessions.Broadcast(message.CursorPosition{
-					Event: event.CURSOR_POSITION,
-					MemberId: memberId,
-					X:        uint16(x),
-					Y:        uint16(y),
-				}, []string{ memberId })
-			}
-		}
-	}()
 }
 
 func (ws *WebSocketManagerCtx) Shutdown() error {