Merge 1d88a03b93 into 32341574dc

packages/input/src/controller.ts

@@ -56,7 +56,7 @@ export class Controller {
   // Polling configuration
   private readonly FULL_RATE_MS = 10; // 100 UPS
   private readonly IDLE_THRESHOLD = 100; // ms before considering idle/hands off controller
-  private readonly FULL_INTERVAL = 250; // ms before sending full state occassionally, to verify inputs are synced
+  private readonly FULL_INTERVAL= 250; // ms before sending full state occassionally, to verify inputs are synced
 
   // Polling state
   private pollingState: PollState = PollState.IDLE;
@@ -230,7 +230,7 @@ export class Controller {
       // Changing from running to idle..
       if (this.pollingState === PollState.RUNNING) {
         // Send full state on idle assumption
-        this.sendBatchedState(0xff, 0);
+        this.sendBatchedState(0xFF, 0);
         this.pollingState = PollState.IDLE;
       }
     }
@@ -364,15 +364,12 @@ export class Controller {
 
     // For FULL_STATE, include everything
     if (updateType === 0) {
-      message.changedFields = 0xff;
+      message.changedFields = 0xFF;
 
       message.buttonChangedMask = Object.fromEntries(
-        Array.from(this.state.buttonState)
+        Array.from(this.state.buttonState).map(([key, value]) => {
-          .map(
+          return [this.controllerButtonToVirtualKeyCode(key), value];
-            ([key, value]) =>
+        }),
-              [this.controllerButtonToVirtualKeyCode(key), value] as const,
-          )
-          .filter(([code]) => code !== undefined),
       );
       message.leftStickX = this.state.leftX;
       message.leftStickY = this.state.leftY;
@@ -405,12 +402,9 @@ export class Controller {
             })
             .map((key) => {
               const newValue = currentStateMap.get(key) ?? false;
-              return [
+
-                this.controllerButtonToVirtualKeyCode(key),
+              return [this.controllerButtonToVirtualKeyCode(key), newValue];
-                newValue,
+            }),
-              ] as const;
-            })
-            .filter(([code]) => code !== undefined),
         );
       }
       if (changedFields & this.CHANGED_LEFT_STICK_X) {
@@ -474,16 +468,14 @@ export class Controller {
       this.wrtc.removeDataChannelCallback(this._dcHandler);
       this._dcHandler = null;
     }
-    if (this.gamepad) {
+    // Gamepad disconnected
-      // Gamepad disconnected
+    const detachMsg = createMessage(
-      const detachMsg = createMessage(
+      create(ProtoControllerDetachSchema, {
-        create(ProtoControllerDetachSchema, {
+        sessionSlot: this.gamepad.index,
-          sessionSlot: this.gamepad.index,
+      }),
-        }),
+      "controllerInput",
-        "controllerInput",
+    );
-      );
+    this.wrtc.sendBinary(toBinary(ProtoMessageSchema, detachMsg));
-      this.wrtc.sendBinary(toBinary(ProtoMessageSchema, detachMsg));
-    }
   }
 
   private controllerButtonToVirtualKeyCode(code: number): number | undefined {
@@ -495,7 +487,7 @@ export class Controller {
 
     // Check if this rumble is for us
     if (
-      rumbleMsg.sessionId !== this.wrtc.getSessionID() ||
+      rumbleMsg.sessionId !== this.wrtc.getSessionID() &&
       rumbleMsg.sessionSlot !== this.gamepad.index
     )
       return;

packages/input/src/webrtc-stream.ts

@@ -172,7 +172,7 @@ export class WebRTCStream {
         const requestMsg = createMessage(
           create(ProtoClientRequestRoomStreamSchema, {
             roomName: roomName,
-            sessionId: clientId ?? "",
+            sessionId: clientId,
           }),
           "request-stream-room",
         );

packages/relay/internal/common/ice_helper.go

@@ -1,53 +0,0 @@
-package common
-
-import (
-	"log/slog"
-
-	"github.com/pion/webrtc/v4"
-)
-
-// ICEHelper holds webrtc.ICECandidateInit(s) until remote candidate is set for given webrtc.PeerConnection
-// Held candidates should be flushed at the end of negotiation to ensure all are available for connection
-type ICEHelper struct {
-	candidates []webrtc.ICECandidateInit
-	pc         *webrtc.PeerConnection
-}
-
-func NewICEHelper(pc *webrtc.PeerConnection) *ICEHelper {
-	return &ICEHelper{
-		pc:         pc,
-		candidates: make([]webrtc.ICECandidateInit, 0),
-	}
-}
-
-func (ice *ICEHelper) SetPeerConnection(pc *webrtc.PeerConnection) {
-	ice.pc = pc
-}
-
-func (ice *ICEHelper) AddCandidate(c webrtc.ICECandidateInit) {
-	if ice.pc != nil {
-		if ice.pc.RemoteDescription() != nil {
-			// Add immediately if remote is set
-			if err := ice.pc.AddICECandidate(c); err != nil {
-				slog.Error("Failed to add ICE candidate", "err", err)
-			}
-			// Also flush held candidates automatically
-			ice.FlushHeldCandidates()
-		} else {
-			// Hold in slice until remote is set
-			ice.candidates = append(ice.candidates, c)
-		}
-	}
-}
-
-func (ice *ICEHelper) FlushHeldCandidates() {
-	if ice.pc != nil && len(ice.candidates) > 0 {
-		for _, heldCandidate := range ice.candidates {
-			if err := ice.pc.AddICECandidate(heldCandidate); err != nil {
-				slog.Error("Failed to add held ICE candidate", "err", err)
-			}
-		}
-		// Clear the held candidates
-		ice.candidates = make([]webrtc.ICECandidateInit, 0)
-	}
-}

packages/relay/internal/core/protocol_stream.go

@@ -71,7 +71,7 @@ func (sp *StreamProtocol) handleStreamRequest(stream network.Stream) {
 	safeBRW := common.NewSafeBufioRW(brw)
 
 	var currentRoomName string // Track the current room for this stream
-	iceHelper := common.NewICEHelper(nil)
+	iceHolder := make([]webrtc.ICECandidateInit, 0)
 	for {
 		var msgWrapper gen.ProtoMessage
 		err := safeBRW.ReceiveProto(&msgWrapper)
@@ -177,7 +177,6 @@ func (sp *StreamProtocol) handleStreamRequest(stream network.Stream) {
 
 				// Assign peer connection
 				participant.PeerConnection = pc
-				iceHelper.SetPeerConnection(pc)
 
 				// Add audio/video tracks
 				{
@@ -345,7 +344,29 @@ func (sp *StreamProtocol) handleStreamRequest(stream network.Stream) {
 					SDPMLineIndex:    &smollified,
 					UsernameFragment: iceMsg.Candidate.UsernameFragment,
 				}
-				iceHelper.AddCandidate(cand)
+				// Use currentRoomName to get the connection from nested map
+				if len(currentRoomName) > 0 {
+					if roomMap, ok := sp.servedConns.Get(currentRoomName); ok {
+						if conn, ok := roomMap.Get(stream.Conn().RemotePeer()); ok && conn.pc.RemoteDescription() != nil {
+							if err = conn.pc.AddICECandidate(cand); err != nil {
+								slog.Error("Failed to add ICE candidate", "err", err)
+							}
+							for _, heldIce := range iceHolder {
+								if err := conn.pc.AddICECandidate(heldIce); err != nil {
+									slog.Error("Failed to add held ICE candidate", "err", err)
+								}
+							}
+							// Clear the held candidates
+							iceHolder = make([]webrtc.ICECandidateInit, 0)
+						} else {
+							// Hold the candidate until remote description is set
+							iceHolder = append(iceHolder, cand)
+						}
+					}
+				} else {
+					// Hold the candidate until remote description is set
+					iceHolder = append(iceHolder, cand)
+				}
 			} else {
 				slog.Error("Could not GetIce from ice-candidate")
 			}
@@ -365,8 +386,6 @@ func (sp *StreamProtocol) handleStreamRequest(stream network.Stream) {
 								continue
 							}
 							slog.Debug("Set remote description for answer")
-							// Flush held candidates now if missed before (race-condition)
-							iceHelper.FlushHeldCandidates()
 						} else {
 							slog.Warn("Received answer without active PeerConnection")
 						}
@@ -387,7 +406,7 @@ func (sp *StreamProtocol) handleStreamPush(stream network.Stream) {
 	safeBRW := common.NewSafeBufioRW(brw)
 
 	var room *shared.Room
-	iceHelper := common.NewICEHelper(nil)
+	iceHolder := make([]webrtc.ICECandidateInit, 0)
 	for {
 		var msgWrapper gen.ProtoMessage
 		err := safeBRW.ReceiveProto(&msgWrapper)
@@ -464,7 +483,21 @@ func (sp *StreamProtocol) handleStreamPush(stream network.Stream) {
 					SDPMLineIndex:    &smollified,
 					UsernameFragment: iceMsg.Candidate.UsernameFragment,
 				}
-				iceHelper.AddCandidate(cand)
+				if conn, ok := sp.incomingConns.Get(room.Name); ok && conn.pc.RemoteDescription() != nil {
+					if err = conn.pc.AddICECandidate(cand); err != nil {
+						slog.Error("Failed to add ICE candidate for pushed stream", "err", err)
+					}
+					for _, heldIce := range iceHolder {
+						if err = conn.pc.AddICECandidate(heldIce); err != nil {
+							slog.Error("Failed to add held ICE candidate for pushed stream", "err", err)
+						}
+					}
+					// Clear the held candidates
+					iceHolder = make([]webrtc.ICECandidateInit, 0)
+				} else {
+					// Hold the candidate until remote description is set
+					iceHolder = append(iceHolder, cand)
+				}
 			} else {
 				slog.Error("Failed to GetIce in pushed stream ice-candidate")
 			}
@@ -496,7 +529,6 @@ func (sp *StreamProtocol) handleStreamPush(stream network.Stream) {
 
 				// Assign room peer connection
 				room.PeerConnection = pc
-				iceHelper.SetPeerConnection(pc)
 
 				pc.OnDataChannel(func(dc *webrtc.DataChannel) {
 					// TODO: Is this the best way to handle DataChannel? Should we just use the map directly?
@@ -657,8 +689,6 @@ func (sp *StreamProtocol) handleStreamPush(stream network.Stream) {
 					continue
 				}
 				slog.Debug("Set remote description for pushed stream", "room", room.Name)
-				// Flush candidates now if they weren't before (race-condition)
-				iceHelper.FlushHeldCandidates()
 
 				// Create an answer
 				answer, err := pc.CreateAnswer(nil)

packages/relay/internal/shared/participant.go

@@ -63,13 +63,13 @@ func (p *Participant) SetTrack(trackType webrtc.RTPCodecType, track *webrtc.Trac
 		p.AudioTrack = track
 		_, err := p.PeerConnection.AddTrack(track)
 		if err != nil {
-			slog.Error("Failed to add Participant audio track", "participant", p.ID, "err", err)
+			slog.Error("Failed to add Participant audio track", err)
 		}
 	case webrtc.RTPCodecTypeVideo:
 		p.VideoTrack = track
 		_, err := p.PeerConnection.AddTrack(track)
 		if err != nil {
-			slog.Error("Failed to add Participant video track", "participant", p.ID, "err", err)
+			slog.Error("Failed to add Participant video track", err)
 		}
 	default:
 		slog.Warn("Unknown track type", "participant", p.ID, "trackType", trackType)
@@ -78,9 +78,6 @@ func (p *Participant) SetTrack(trackType webrtc.RTPCodecType, track *webrtc.Trac
 
 // Close cleans up participant resources
 func (p *Participant) Close() {
-	p.closeOnce.Do(func() {
-		close(p.packetQueue)
-	})
 	if p.DataChannel != nil {
 		err := p.DataChannel.Close()
 		if err != nil {