Merge 9fafcfb94c into 8f4081a5b1

hidrpc.go

@@ -53,13 +53,13 @@ func handleHidRPCMessage(message hidrpc.Message, session *Session) {
 		rpcCancelKeyboardMacro()
 		return
 
-	case hidrpc.TypeKeyboardMacroTokenState:
-		tokenState, err := message.KeyboardMacroTokenState()
+	case hidrpc.TypeCancelKeyboardMacroByTokenReport:
+		token, err := message.KeyboardMacroToken()
 		if err != nil {
 			logger.Warn().Err(err).Msg("failed to get keyboard macro token")
 			return
 		}
-		rpcCancelKeyboardMacroByToken(tokenState.Token)
+		rpcCancelKeyboardMacroByToken(token)
 		return
 
 	case hidrpc.TypeKeypressKeepAliveReport:
@@ -96,7 +96,6 @@ func onHidMessage(msg hidQueueMessage, session *Session) {
 
 	scopedLogger := hidRPCLogger.With().
 		Str("channel", msg.channel).
-		Dur("timelimit", msg.timelimit).
 		Int("data_len", dataLen).
 		Bytes("data", data[:min(dataLen, 32)]).
 		Logger()
@@ -126,7 +125,7 @@ func onHidMessage(msg hidQueueMessage, session *Session) {
 		r <- nil
 	}()
 	select {
-	case <-time.After(msg.timelimit * time.Second):
+	case <-time.After(1 * time.Second):
 		scopedLogger.Warn().Msg("HID RPC message timed out")
 	case <-r:
 		scopedLogger.Debug().Dur("duration", time.Since(t)).Msg("HID RPC message handled")
@@ -242,8 +241,6 @@ func reportHidRPC(params any, session *Session) {
 		message, err = hidrpc.NewKeydownStateMessage(params).Marshal()
 	case hidrpc.KeyboardMacroState:
 		message, err = hidrpc.NewKeyboardMacroStateMessage(params.State, params.IsPaste).Marshal()
-	case hidrpc.KeyboardMacroTokenState:
-		message, err = hidrpc.NewKeyboardMacroTokenMessage(params.Token).Marshal()
 	default:
 		err = fmt.Errorf("unknown HID RPC message type: %T", params)
 	}

internal/hidrpc/hidrpc.go

@@ -2,7 +2,6 @@ package hidrpc
 
 import (
 	"fmt"
-	"time"
 
 	"github.com/google/uuid"
 	"github.com/jetkvm/kvm/internal/usbgadget"
@@ -12,19 +11,19 @@ import (
 type MessageType byte
 
 const (
-	TypeHandshake                 MessageType = 0x01
-	TypeKeyboardReport            MessageType = 0x02
-	TypePointerReport             MessageType = 0x03
-	TypeWheelReport               MessageType = 0x04
-	TypeKeypressReport            MessageType = 0x05
-	TypeKeypressKeepAliveReport   MessageType = 0x09
-	TypeMouseReport               MessageType = 0x06
-	TypeKeyboardMacroReport       MessageType = 0x07
-	TypeCancelKeyboardMacroReport MessageType = 0x08
-	TypeKeyboardLedState          MessageType = 0x32
-	TypeKeydownState              MessageType = 0x33
-	TypeKeyboardMacroState        MessageType = 0x34
-	TypeKeyboardMacroTokenState   MessageType = 0x35
+	TypeHandshake                        MessageType = 0x01
+	TypeKeyboardReport                   MessageType = 0x02
+	TypePointerReport                    MessageType = 0x03
+	TypeWheelReport                      MessageType = 0x04
+	TypeKeypressReport                   MessageType = 0x05
+	TypeKeypressKeepAliveReport          MessageType = 0x09
+	TypeMouseReport                      MessageType = 0x06
+	TypeKeyboardMacroReport              MessageType = 0x07
+	TypeCancelKeyboardMacroReport        MessageType = 0x08
+	TypeKeyboardLedState                 MessageType = 0x32
+	TypeKeydownState                     MessageType = 0x33
+	TypeKeyboardMacroState               MessageType = 0x34
+	TypeCancelKeyboardMacroByTokenReport MessageType = 0x35
 )
 
 type QueueIndex int
@@ -32,26 +31,26 @@ type QueueIndex int
 const (
 	Version        byte = 0x01 // Version of the HID RPC protocol
 	HandshakeQueue int  = 0    // Queue index for handshake messages
-	KeyboardQueue  int  = 1    // Queue index for keyboard messages
+	KeyboardQueue  int  = 1    // Queue index for keyboard and macro messages
 	MouseQueue     int  = 2    // Queue index for mouse messages
-	MacroQueue     int  = 3    // Queue index for macro messages
+	MacroQueue     int  = 3    // Queue index for macro cancel messages
 	OtherQueue     int  = 4    // Queue index for other messages
 )
 
 // GetQueueIndex returns the index of the queue to which the message should be enqueued.
-func GetQueueIndex(messageType MessageType) (int, time.Duration) {
+func GetQueueIndex(messageType MessageType) int {
 	switch messageType {
 	case TypeHandshake:
-		return HandshakeQueue, 1
+		return HandshakeQueue
 	case TypeKeyboardReport, TypeKeypressReport, TypeKeyboardLedState, TypeKeydownState, TypeKeyboardMacroState:
-		return KeyboardQueue, 1
+		return KeyboardQueue
 	case TypePointerReport, TypeMouseReport, TypeWheelReport:
-		return MouseQueue, 1
+		return MouseQueue
 	// we don't want to block the queue for these messages
-	case TypeKeyboardMacroReport, TypeCancelKeyboardMacroReport, TypeKeyboardMacroTokenState:
-		return MacroQueue, 60 // 1 minute timeout
+	case TypeKeyboardMacroReport, TypeCancelKeyboardMacroReport, TypeCancelKeyboardMacroByTokenReport:
+		return MacroQueue
 	default:
-		return OtherQueue, 5
+		return OtherQueue
 	}
 }
 

internal/hidrpc/message.go

@@ -69,11 +69,11 @@ func (m *Message) String() string {
 			return fmt.Sprintf("CancelKeyboardMacroReport{Malformed: %v}", m.d)
 		}
 		return "CancelKeyboardMacroReport"
-	case TypeKeyboardMacroTokenState:
+	case TypeCancelKeyboardMacroByTokenReport:
 		if len(m.d) != 16 {
-			return fmt.Sprintf("KeyboardMacroTokenState{Malformed: %v}", m.d)
+			return fmt.Sprintf("CancelKeyboardMacroByTokenReport{Malformed: %v}", m.d)
 		}
-		return fmt.Sprintf("KeyboardMacroTokenState{Token: %s}", uuid.Must(uuid.FromBytes(m.d)).String())
+		return fmt.Sprintf("CancelKeyboardMacroByTokenReport{Token: %s}", uuid.Must(uuid.FromBytes(m.d)).String())
 	case TypeKeyboardLedState:
 		if len(m.d) < 1 {
 			return fmt.Sprintf("KeyboardLedState{Malformed: %v}", m.d)
@@ -246,28 +246,19 @@ func (m *Message) KeyboardMacroState() (KeyboardMacroState, error) {
 	}, nil
 }
 
-type KeyboardMacroTokenState struct {
-	Token uuid.UUID
-}
-
-// KeyboardMacroTokenState returns the keyboard macro token UUID from the message.
-func (m *Message) KeyboardMacroTokenState() (KeyboardMacroTokenState, error) {
-	if m.t != TypeKeyboardMacroTokenState {
-		return KeyboardMacroTokenState{}, fmt.Errorf("invalid message type: %d", m.t)
+// KeyboardMacroToken returns the keyboard macro token UUID from the message.
+func (m *Message) KeyboardMacroToken() (uuid.UUID, error) {
+	if m.t != TypeCancelKeyboardMacroByTokenReport {
+		return uuid.Nil, fmt.Errorf("invalid message type: %d", m.t)
 	}
 
 	if len(m.d) == 0 {
-		return KeyboardMacroTokenState{Token: uuid.Nil}, nil
+		return uuid.Nil, nil
 	}
 
 	if len(m.d) != 16 {
-		return KeyboardMacroTokenState{}, fmt.Errorf("invalid UUID length: %d", len(m.d))
+		return uuid.Nil, fmt.Errorf("invalid UUID length: %d", len(m.d))
 	}
 
-	token, err := uuid.FromBytes(m.d)
-	if err != nil {
-		return KeyboardMacroTokenState{}, fmt.Errorf("invalid UUID: %v", err)
-	}
-
-	return KeyboardMacroTokenState{Token: token}, nil
+	return uuid.FromBytes(m.d)
 }

jsonrpc.go

@@ -1091,10 +1091,10 @@ func getKeyboardMacroCancelMap() map[uuid.UUID]RunningMacro {
 func addKeyboardMacro(isPaste bool, cancel context.CancelFunc) uuid.UUID {
 	keyboardMacroLock.Lock()
 	defer keyboardMacroLock.Unlock()
-	cancelMap := getKeyboardMacroCancelMap()
+	keyboardMacroCancelMap := getKeyboardMacroCancelMap()
 
 	token := uuid.New() // Generate a unique token
-	cancelMap[token] = RunningMacro{
+	keyboardMacroCancelMap[token] = RunningMacro{
 		isPaste: isPaste,
 		cancel:  cancel,
 	}
@@ -1104,19 +1104,19 @@ func addKeyboardMacro(isPaste bool, cancel context.CancelFunc) uuid.UUID {
 func removeRunningKeyboardMacro(token uuid.UUID) {
 	keyboardMacroLock.Lock()
 	defer keyboardMacroLock.Unlock()
-	cancelMap := getKeyboardMacroCancelMap()
+	keyboardMacroCancelMap := getKeyboardMacroCancelMap()
 
-	delete(cancelMap, token)
+	delete(keyboardMacroCancelMap, token)
 }
 
 func cancelRunningKeyboardMacro(token uuid.UUID) {
 	keyboardMacroLock.Lock()
 	defer keyboardMacroLock.Unlock()
-	cancelMap := getKeyboardMacroCancelMap()
+	keyboardMacroCancelMap := getKeyboardMacroCancelMap()
 
-	if runningMacro, exists := cancelMap[token]; exists {
+	if runningMacro, exists := keyboardMacroCancelMap[token]; exists {
 		runningMacro.cancel()
-		delete(cancelMap, token)
+		delete(keyboardMacroCancelMap, token)
 		logger.Info().Interface("token", token).Msg("canceled keyboard macro by token")
 	} else {
 		logger.Debug().Interface("token", token).Msg("no running keyboard macro found for token")
@@ -1126,11 +1126,11 @@ func cancelRunningKeyboardMacro(token uuid.UUID) {
 func cancelAllRunningKeyboardMacros() {
 	keyboardMacroLock.Lock()
 	defer keyboardMacroLock.Unlock()
-	cancelMap := getKeyboardMacroCancelMap()
+	keyboardMacroCancelMap := getKeyboardMacroCancelMap()
 
-	for token, runningMacro := range cancelMap {
+	for token, runningMacro := range keyboardMacroCancelMap {
 		runningMacro.cancel()
-		delete(cancelMap, token)
+		delete(keyboardMacroCancelMap, token)
 		logger.Info().Interface("token", token).Msg("cancelled keyboard macro")
 	}
 }
@@ -1139,10 +1139,12 @@ func reportRunningMacrosState() {
 	if currentSession != nil {
 		keyboardMacroLock.Lock()
 		defer keyboardMacroLock.Unlock()
-		cancelMap := getKeyboardMacroCancelMap()
+		keyboardMacroCancelMap := getKeyboardMacroCancelMap()
 
 		isPaste := false
-		for _, runningMacro := range cancelMap {
+		anyRunning := false
+		for _, runningMacro := range keyboardMacroCancelMap {
+			anyRunning = true
 			if runningMacro.isPaste {
 				isPaste = true
 				break
@@ -1150,7 +1152,7 @@ func reportRunningMacrosState() {
 		}
 
 		state := hidrpc.KeyboardMacroState{
-			State:   len(cancelMap) > 0,
+			State:   anyRunning,
 			IsPaste: isPaste,
 		}
 
@@ -1192,10 +1194,7 @@ func rpcCancelKeyboardMacroByToken(token uuid.UUID) {
 }
 
 func executeKeyboardMacro(ctx context.Context, isPaste bool, macro []hidrpc.KeyboardMacroStep) error {
-	logger.Debug().
-		Int("macro_steps", len(macro)).
-		Bool("isPaste", isPaste).
-		Msg("Executing keyboard macro")
+	logger.Debug().Int("macro_steps", len(macro)).Msg("Executing keyboard macro")
 
 	// don't report keyboard state changes while executing the macro
 	gadget.SuspendKeyDownMessages()

webrtc.go

@@ -34,7 +34,7 @@ type Session struct {
 	lastTimerResetTime       time.Time  // Track when auto-release timer was last reset
 	keepAliveJitterLock      sync.Mutex // Protect jitter compensation timing state
 	hidQueueLock             sync.Mutex
-	hidQueues                []chan hidQueueMessage
+	hidQueue                 []chan hidQueueMessage
 
 	keysDownStateQueue chan usbgadget.KeysDownState
 }
@@ -48,8 +48,7 @@ func (s *Session) resetKeepAliveTime() {
 
 type hidQueueMessage struct {
 	webrtc.DataChannelMessage
-	channel   string
-	timelimit time.Duration
+	channel string
 }
 
 type SessionConfig struct {
@@ -94,20 +93,19 @@ func (s *Session) ExchangeOffer(offerStr string) (string, error) {
 	return base64.StdEncoding.EncodeToString(localDescription), nil
 }
 
-func (s *Session) initHidQueues() {
+func (s *Session) initQueues() {
 	s.hidQueueLock.Lock()
 	defer s.hidQueueLock.Unlock()
 
-	s.hidQueues = make([]chan hidQueueMessage, hidrpc.OtherQueue+1)
-	s.hidQueues[hidrpc.HandshakeQueue] = make(chan hidQueueMessage, 2) // we don't really want to queue many handshake messages
-	s.hidQueues[hidrpc.KeyboardQueue] = make(chan hidQueueMessage, 256)
-	s.hidQueues[hidrpc.MouseQueue] = make(chan hidQueueMessage, 256)
-	s.hidQueues[hidrpc.MacroQueue] = make(chan hidQueueMessage, 10) // macros can be long, but we don't want to queue too many
-	s.hidQueues[hidrpc.OtherQueue] = make(chan hidQueueMessage, 256)
+	s.hidQueue = make([]chan hidQueueMessage, 0)
+	for i := 0; i <= hidrpc.OtherQueue; i++ {
+		q := make(chan hidQueueMessage, 256)
+		s.hidQueue = append(s.hidQueue, q)
+	}
 }
 
-func (s *Session) handleQueue(queue chan hidQueueMessage) {
-	for msg := range queue {
+func (s *Session) handleQueues(index int) {
+	for msg := range s.hidQueue[index] {
 		onHidMessage(msg, s)
 	}
 }
@@ -162,18 +160,17 @@ func getOnHidMessageHandler(session *Session, scopedLogger *zerolog.Logger, chan
 		l.Trace().Msg("received data in HID RPC message handler")
 
 		// Enqueue to ensure ordered processing
-		queueIndex, timelimit := hidrpc.GetQueueIndex(hidrpc.MessageType(msg.Data[0]))
-		if queueIndex >= len(session.hidQueues) || queueIndex < 0 {
+		queueIndex := hidrpc.GetQueueIndex(hidrpc.MessageType(msg.Data[0]))
+		if queueIndex >= len(session.hidQueue) || queueIndex < 0 {
 			l.Warn().Int("queueIndex", queueIndex).Msg("received data in HID RPC message handler, but queue index not found")
 			queueIndex = hidrpc.OtherQueue
 		}
 
-		queue := session.hidQueues[queueIndex]
+		queue := session.hidQueue[queueIndex]
 		if queue != nil {
 			queue <- hidQueueMessage{
 				DataChannelMessage: msg,
 				channel:            channel,
-				timelimit:          timelimit,
 			}
 		} else {
 			l.Warn().Int("queueIndex", queueIndex).Msg("received data in HID RPC message handler, but queue is nil")
@@ -223,7 +220,7 @@ func newSession(config SessionConfig) (*Session, error) {
 
 	session := &Session{peerConnection: peerConnection}
 	session.rpcQueue = make(chan webrtc.DataChannelMessage, 256)
-	session.initHidQueues()
+	session.initQueues()
 	session.initKeysDownStateQueue()
 
 	go func() {
@@ -233,8 +230,8 @@ func newSession(config SessionConfig) (*Session, error) {
 		}
 	}()
 
-	for queue := range session.hidQueues {
-		go session.handleQueue(session.hidQueues[queue])
+	for i := 0; i < len(session.hidQueue); i++ {
+		go session.handleQueues(i)
 	}
 
 	peerConnection.OnDataChannel(func(d *webrtc.DataChannel) {
@@ -259,11 +256,7 @@ func newSession(config SessionConfig) (*Session, error) {
 			session.RPCChannel = d
 			d.OnMessage(func(msg webrtc.DataChannelMessage) {
 				// Enqueue to ensure ordered processing
-				if session.rpcQueue != nil {
-					session.rpcQueue <- msg
-				} else {
-					scopedLogger.Warn().Msg("RPC message received but rpcQueue is nil")
-				}
+				session.rpcQueue <- msg
 			})
 			triggerOTAStateUpdate()
 			triggerVideoStateUpdate()
@@ -332,23 +325,22 @@ func newSession(config SessionConfig) (*Session, error) {
 			_ = peerConnection.Close()
 		}
 		if connectionState == webrtc.ICEConnectionStateClosed {
-			scopedLogger.Debug().Msg("ICE Connection State is closed, tearing down session")
+			scopedLogger.Debug().Msg("ICE Connection State is closed, unmounting virtual media")
 			if session == currentSession {
 				// Cancel any ongoing keyboard report multi when session closes
 				cancelAllRunningKeyboardMacros()
 				currentSession = nil
 			}
-
 			// Stop RPC processor
 			if session.rpcQueue != nil {
 				close(session.rpcQueue)
 				session.rpcQueue = nil
 			}
 
-			// Stop HID RPC processors
-			for i := 0; i < len(session.hidQueues); i++ {
-				close(session.hidQueues[i])
-				session.hidQueues[i] = nil
+			// Stop HID RPC processor
+			for i := 0; i < len(session.hidQueue); i++ {
+				close(session.hidQueue[i])
+				session.hidQueue[i] = nil
 			}
 
 			close(session.keysDownStateQueue)