Add participant RPC for getting remote headers. (#699)

Author: dennwc

pkg/sip/inbound.go

@@ -1444,6 +1444,9 @@ func (c *inboundCall) createLiveKitParticipant(ctx context.Context, rconf RoomCo
 	if err != nil {
 		return err
 	}
+	if err := registerSignalingRPC(c.lkRoom, c.cc); err != nil {
+		return err
+	}
 	return nil
 }
 

pkg/sip/outbound.go

@@ -429,6 +429,9 @@ func (c *outboundCall) connectToRoom(ctx context.Context, lkNew RoomConfig, getR
 	}
 	c.lkRoom = r
 	c.lkRoomIn = local
+	if err := registerSignalingRPC(c.lkRoom, c.cc); err != nil {
+		return err
+	}
 	return nil
 }
 

pkg/sip/outbound_utilities_test.go

@@ -265,6 +265,10 @@ func (r *testRoom) NewTrack() *mixer.Input {
 	return r.room.NewTrack()
 }
 
+func (r *testRoom) RegisterRPC(method string, handler lksdk.RpcHandlerFunc) error {
+	return r.room.RegisterRPC(method, handler)
+}
+
 type testSIPClientTransaction struct {
 	log       logger.Logger
 	responses chan *sip.Response

pkg/sip/room.go

@@ -162,6 +162,7 @@ type RoomInterface interface {
 	NewParticipantTrack(sampleRate int) (msdk.WriteCloser[msdk.PCM16Sample], error)
 	SendData(data lksdk.DataPacket, opts ...lksdk.DataPublishOption) error
 	NewTrack() *mixer.Input
+	RegisterRPC(method string, handler lksdk.RpcHandlerFunc) error
 }
 
 type GetRoomFunc func(log logger.Logger, st *RoomStats) RoomInterface
@@ -455,6 +456,10 @@ func (r *Room) Connect(ctx context.Context, conf *config.Config, rconf RoomConfi
 	return nil
 }
 
+func (r *Room) RegisterRPC(method string, handler lksdk.RpcHandlerFunc) error {
+	return r.room.RegisterRpcMethod(method, handler)
+}
+
 func (r *Room) Subscribe() {
 	if r.room == nil {
 		return

pkg/sip/rpc.go

@@ -0,0 +1,101 @@
+package sip
+
+import (
+	"encoding/json"
+	"strings"
+
+	lksdk "github.com/livekit/server-sdk-go/v2"
+)
+
+func registerRPCMethodJSON[Req any, Resp any](r RoomInterface, name string, fnc func(r Req) (Resp, error)) error {
+	return r.RegisterRPC(name, func(data lksdk.RpcInvocationData) (string, error) {
+		var req Req
+		if len(data.Payload) != 0 {
+			if err := json.Unmarshal([]byte(data.Payload), &req); err != nil {
+				return "", err
+			}
+		}
+		resp, err := fnc(req)
+		if err != nil {
+			return "", err
+		}
+		out, err := json.Marshal(&resp)
+		return string(out), err
+	})
+}
+
+type getHeadersReq struct {
+	Include []string `json:"include,omitempty"` // list of headers to include
+	Exclude []string `json:"exclude,omitempty"` // list of headers to exclude
+}
+
+type getHeadersResp struct {
+	Headers map[string]string `json:"headers"`
+}
+
+// getHeadersIgnore is a set of header that are filtered out of the participant RPC response.
+var getHeadersIgnore = map[string]struct{}{
+	"route":               {},
+	"record-route":        {},
+	"via":                 {},
+	"supported":           {},
+	"accept":              {},
+	"allow":               {},
+	"allow-events":        {},
+	"content-type":        {},
+	"content-length":      {},
+	"content-encoding":    {},
+	"content-disposition": {},
+	"cseq":                {},
+	"event":               {},
+	"expires":             {},
+	"session-expires":     {},
+	"min-se":              {},
+	"max-forwards":        {},
+}
+
+func rpcGetHeaders(cc Signaling, r getHeadersReq) (getHeadersResp, error) {
+	var (
+		only map[string]struct{}
+		skip map[string]struct{}
+	)
+	if list := r.Include; len(list) != 0 {
+		only = make(map[string]struct{}, len(list))
+		for _, h := range list {
+			only[strings.ToLower(h)] = struct{}{}
+		}
+	}
+	if list := r.Exclude; len(list) != 0 {
+		skip = make(map[string]struct{}, len(list))
+		for _, h := range list {
+			skip[strings.ToLower(h)] = struct{}{}
+		}
+	}
+	out := make(map[string]string)
+	for _, h := range cc.RemoteHeaders() {
+		name := strings.ToLower(h.Name())
+		if _, ok := getHeadersIgnore[name]; ok {
+			continue
+		}
+		if _, ok := skip[name]; ok {
+			continue
+		}
+		if _, ok := only[name]; !ok && only != nil {
+			continue
+		}
+		out[h.Name()] = h.Value()
+	}
+	return getHeadersResp{
+		Headers: out,
+	}, nil
+}
+
+func registerSignalingRPC(r RoomInterface, cc Signaling) error {
+	err := registerRPCMethodJSON(r, "lk.sip.GetRemoteHeaders", func(r getHeadersReq) (getHeadersResp, error) {
+		return rpcGetHeaders(cc, r)
+	})
+	if err != nil {
+		return err
+	}
+	return nil
+}