Support for request/response signaling messages

engine.go

@@ -77,6 +77,7 @@ type engineHandler interface {
 	OnSubscribedQualityUpdate(subscribedQualityUpdate *livekit.SubscribedQualityUpdate)
 	OnSubscribedAudioCodecUpdate(subscribedAudioCodecUpdate *livekit.SubscribedAudioCodecUpdate)
 	OnMediaSectionsRequirement(mediaSectionsRequirement *livekit.MediaSectionsRequirement)
+	OnRequestResponse(requestResponse *livekit.RequestResponse)
 }
 
 // -------------------------------------------
@@ -130,6 +131,11 @@ type RTCEngine struct {
 	trackPublishedListenersLock sync.Mutex
 	trackPublishedListeners     map[string]chan *livekit.TrackPublishedResponse
 
+	// signal requests waiting for a response, keyed by request id
+	pendingRequestsLock sync.Mutex
+	nextRequestID       uint32 // guarded by pendingRequestsLock
+	pendingRequests     map[uint32]chan proto.Message
+
 	subscriberPrimary bool
 	hasPublish        atomic.Bool
 	closed            atomic.Bool
@@ -154,6 +160,7 @@ func NewRTCEngine(
 		engineHandler:            engineHandler,
 		cbGetLocalParticipantSID: getLocalParticipantSID,
 		trackPublishedListeners:  make(map[string]chan *livekit.TrackPublishedResponse),
+		pendingRequests:          make(map[uint32]chan proto.Message),
 		reliableMsgSeq:           1,
 		connectionManager:        newConnectionManager(regionProvider),
 	}
@@ -359,6 +366,7 @@ func (e *RTCEngine) Close() {
 	}
 
 	e.connectionManager.setClosed()
+	e.abortPendingRequests()
 
 	e.pclock.Lock()
 	e.pendingPublisherOffer = webrtc.SessionDescription{}
@@ -755,6 +763,76 @@ func (e *RTCEngine) UnregisterTrackPublishedListener(cid string) {
 	e.trackPublishedListenersLock.Unlock()
 }
 
+func (e *RTCEngine) newPendingRequest() *pendingRequest {
+	p := &pendingRequest{
+		engine: e,
+		ch:     make(chan proto.Message, 1),
+	}
+
+	e.pendingRequestsLock.Lock()
+	p.id = e.allocateRequestIDLocked()
+	if e.closed.Load() {
+		// nothing will answer, fail Await right away
+		close(p.ch)
+	} else {
+		e.pendingRequests[p.id] = p.ch
+	}
+	e.pendingRequestsLock.Unlock()
+
+	return p
+}
+
+// allocateRequestIDLocked returns the next request id not in use by a pending request, skipping
+// zero. Wrapping around is harmless: an id is reused only once its previous request is done.
+func (e *RTCEngine) allocateRequestIDLocked() uint32 {
+	for {
+		e.nextRequestID++
+		if e.nextRequestID == 0 {
+			continue
+		}
+		if _, inUse := e.pendingRequests[e.nextRequestID]; !inUse {
+			return e.nextRequestID
+		}
+	}
+}
+
+func (e *RTCEngine) removePendingRequest(requestID uint32) {
+	e.pendingRequestsLock.Lock()
+	delete(e.pendingRequests, requestID)
+	e.pendingRequestsLock.Unlock()
+}
+
+// deliverResponse hands res to the request waiting for requestID, if any.
+func (e *RTCEngine) deliverResponse(requestID uint32, res proto.Message) bool {
+	if requestID == 0 {
+		return false
+	}
+
+	e.pendingRequestsLock.Lock()
+	ch, ok := e.pendingRequests[requestID]
+	delete(e.pendingRequests, requestID)
+	e.pendingRequestsLock.Unlock()
+	if !ok {
+		return false
+	}
+
+	// buffered, and removed from the map above so this is the only send
+	ch <- res
+	return true
+}
+
+// abortPendingRequests fails every request still waiting for a response.
+func (e *RTCEngine) abortPendingRequests() {
+	e.pendingRequestsLock.Lock()
+	pending := e.pendingRequests
+	e.pendingRequests = make(map[uint32]chan proto.Message)
+	e.pendingRequestsLock.Unlock()
+
+	for _, ch := range pending {
+		close(ch)
+	}
+}
+
 func (e *RTCEngine) handleDataPacket(msg webrtc.DataChannelMessage) {
 	packet, err := e.readDataPacket(msg)
 	if err != nil {
@@ -1043,6 +1121,8 @@ func (e *RTCEngine) cleanupConnection() {
 }
 
 func (e *RTCEngine) restartConnection() error {
+	// the new session will never answer requests made on the old one
+	e.abortPendingRequests()
 	if err := e.join(nil, nil); err != nil {
 		return err
 	}
@@ -1677,6 +1757,14 @@ func (e *RTCEngine) OnMediaSectionsRequirement(mediaSectionsRequirement *livekit
 	e.engineHandler.OnMediaSectionsRequirement(mediaSectionsRequirement)
 }
 
+func (e *RTCEngine) OnRequestResponse(res *livekit.RequestResponse) {
+	if e.deliverResponse(res.GetRequestId(), res) {
+		return
+	}
+
+	e.engineHandler.OnRequestResponse(res)
+}
+
 // ------------------------------------
 
 func setConfiguration(pcTransport *PCTransport, configuration webrtc.Configuration) {

requestresponse.go

@@ -0,0 +1,97 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package lksdk
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/livekit/protocol/livekit"
+	"google.golang.org/protobuf/proto"
+)
+
+// SignalRequestError is returned when the server rejects a signal request.
+type SignalRequestError struct {
+	RequestID uint32
+	Reason    livekit.RequestResponse_Reason
+	Message   string
+}
+
+func (e *SignalRequestError) Error() string {
+	return fmt.Sprintf("signal request rejected (%s): %s", e.Reason, e.Message)
+}
+
+// pendingRequest reserves the response to a signal request that carries a request id: create it
+// before sending the request, put ID() in the request_id field, then Await the response. The
+// engine owns the registry these are tracked in, see (*RTCEngine).newPendingRequest.
+type pendingRequest struct {
+	engine *RTCEngine
+	id     uint32
+	ch     chan proto.Message
+}
+
+func (p *pendingRequest) ID() uint32 {
+	return p.id
+}
+
+// Await blocks until a response arrives, ctx is done, or the engine is closed. It has no timeout
+// of its own, see withDefaultTimeout. A RequestResponse with a reason other than OK is returned
+// as a *SignalRequestError.
+func (p *pendingRequest) Await(ctx context.Context) (proto.Message, error) {
+	defer p.engine.removePendingRequest(p.id)
+
+	select {
+	case res, ok := <-p.ch:
+		if !ok {
+			return nil, ErrAborted
+		}
+		if rr, isRequestResponse := res.(*livekit.RequestResponse); isRequestResponse && rr.GetReason() != livekit.RequestResponse_OK {
+			return nil, &SignalRequestError{
+				RequestID: rr.GetRequestId(),
+				Reason:    rr.GetReason(),
+				Message:   rr.GetMessage(),
+			}
+		}
+		return res, nil
+
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+}
+
+// awaitResponse is Await for requests whose success is reported by a dedicated message type.
+func awaitResponse[T proto.Message](ctx context.Context, p *pendingRequest) (T, error) {
+	var zero T
+
+	res, err := p.Await(ctx)
+	if err != nil {
+		return zero, err
+	}
+	typed, ok := res.(T)
+	if !ok {
+		return zero, fmt.Errorf("unexpected response %T to signal request %d", res, p.id)
+	}
+	return typed, nil
+}
+
+// withDefaultTimeout applies deadline d to ctx when it has none. A deadline the caller set is
+// never shortened. The returned cancel must be called.
+func withDefaultTimeout(ctx context.Context, d time.Duration) (context.Context, context.CancelFunc) {
+	if _, ok := ctx.Deadline(); ok {
+		return ctx, func() {}
+	}
+	return context.WithTimeout(ctx, d)
+}