read batched signal messages (#1583)

* batch signal messages

* update protcol

pkg/routing/redisrouter.go

@@ -149,10 +149,6 @@ func (r *RedisRouter) StartParticipantSignal(ctx context.Context, roomName livek
 		return
 	}
 
-	if r.usePSRPCSignal {
-		return r.StartParticipantSignalWithNodeID(ctx, roomName, pi, livekit.NodeID(rtcNode.Id))
-	}
-
 	// create a new connection id
 	connectionID = livekit.ConnectionID(utils.NewGuid("CO_"))
 	pKey := participantKeyLegacy(roomName, pi.Identity)
@@ -163,6 +159,10 @@ func (r *RedisRouter) StartParticipantSignal(ctx context.Context, roomName livek
 		return
 	}
 
+	if r.usePSRPCSignal {
+		return r.StartParticipantSignalWithNodeID(ctx, roomName, pi, livekit.NodeID(rtcNode.Id))
+	}
+
 	// index by connectionID, since there may be multiple connections for the participant
 	// set up response channel before sending StartSession and be ready to receive responses.
 	resChan := r.getOrCreateMessageChannel(r.responseChannels, string(connectionID))

pkg/routing/signal.go

@@ -105,6 +105,11 @@ func (r *signalClient) StartParticipantSignal(
 			if err = resChan.WriteMessage(msg.Response); err != nil {
 				break
 			}
+			for _, res := range msg.Responses {
+				if err = resChan.WriteMessage(res); err != nil {
+					break
+				}
+			}
 		}
 
 		logger.Debugw("participant signal stream closed",