Implement TransferParticipant using SIP REFER messages (#182)

Author: biglittlebigben

cmd/livekit-sip/main.go

@@ -91,7 +91,7 @@ func runService(c *cli.Context) error {
 	}
 
 	sipsrv := sip.NewService(conf, mon, log)
-	svc := service.NewService(conf, log, sipsrv.InternalServerImpl(), sipsrv.Stop, sipsrv.ActiveCalls, psrpcClient, bus, mon)
+	svc := service.NewService(conf, log, sipsrv, sipsrv.Stop, sipsrv.ActiveCalls, psrpcClient, bus, mon)
 	sipsrv.SetHandler(svc)
 
 	if err = sipsrv.Start(); err != nil {

go.mod

@@ -13,7 +13,7 @@ require (
 	github.com/jfreymuth/oggvorbis v1.0.5
 	github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1
 	github.com/livekit/mediatransportutil v0.0.0-20240625074155-301bb4a816b7
-	github.com/livekit/protocol v1.21.1-0.20240913074525-1f5de7d620c4
+	github.com/livekit/protocol v1.22.1-0.20240920184753-71b9c184e5c8
 	github.com/livekit/psrpc v0.5.3-0.20240616012458-ac39c8549a0a
 	github.com/livekit/server-sdk-go/v2 v2.2.1-0.20240726160203-3f7f396734c3
 	github.com/mjibson/go-dsp v0.0.0-20180508042940-11479a337f12
@@ -22,12 +22,14 @@ require (
 	github.com/pion/rtp v1.8.6
 	github.com/pion/sdp/v3 v3.0.9
 	github.com/pion/webrtc/v3 v3.2.47
+	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.19.1
 	github.com/sirupsen/logrus v1.9.3
 	github.com/stretchr/testify v1.9.0
 	github.com/urfave/cli/v2 v2.27.2
 	github.com/zaf/resample v1.5.0
 	golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56
+	google.golang.org/protobuf v1.34.2
 	gopkg.in/hraban/opus.v2 v2.0.0-20230925203106-0188a62cb302
 	gopkg.in/yaml.v3 v3.0.1
 )
@@ -96,7 +98,6 @@ require (
 	github.com/pion/stun v0.6.1 // indirect
 	github.com/pion/transport/v2 v2.2.5 // indirect
 	github.com/pion/turn/v2 v2.1.6 // indirect
-	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_model v0.5.0 // indirect
 	github.com/prometheus/common v0.48.0 // indirect
@@ -126,7 +127,6 @@ require (
 	google.golang.org/genproto/googleapis/api v0.0.0-20240528184218-531527333157 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240725223205-93522f1f2a9f // indirect
 	google.golang.org/grpc v1.65.0 // indirect
-	google.golang.org/protobuf v1.34.2 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 )
 

go.sum

@@ -124,8 +124,8 @@ github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 h1:jm09419p0lqTkD
 github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ=
 github.com/livekit/mediatransportutil v0.0.0-20240625074155-301bb4a816b7 h1:F1L8inJoynwIAYpZENNYS+1xHJMF5RFRorsnAlcxfSY=
 github.com/livekit/mediatransportutil v0.0.0-20240625074155-301bb4a816b7/go.mod h1:jwKUCmObuiEDH0iiuJHaGMXwRs3RjrB4G6qqgkr/5oE=
-github.com/livekit/protocol v1.21.1-0.20240913074525-1f5de7d620c4 h1:JhLovaAv+UxOTYpjQxN3IRV3kuJFb9f3LZmg/KqvHkc=
-github.com/livekit/protocol v1.21.1-0.20240913074525-1f5de7d620c4/go.mod h1:AFuwk3+uIWFeO5ohKjx5w606Djl940+wktaZ441VoCI=
+github.com/livekit/protocol v1.22.1-0.20240920184753-71b9c184e5c8 h1:Tt/INVn5HOgyy/OknLzEK46sWDKRnQ+NvsjFkMZDbWc=
+github.com/livekit/protocol v1.22.1-0.20240920184753-71b9c184e5c8/go.mod h1:AFuwk3+uIWFeO5ohKjx5w606Djl940+wktaZ441VoCI=
 github.com/livekit/psrpc v0.5.3-0.20240616012458-ac39c8549a0a h1:EQAHmcYEGlc6V517cQ3Iy0+jHgP6+tM/B4l2vGuLpQo=
 github.com/livekit/psrpc v0.5.3-0.20240616012458-ac39c8549a0a/go.mod h1:CQUBSPfYYAaevg1TNCc6/aYsa8DJH4jSRFdCeSZk5u0=
 github.com/livekit/server-sdk-go/v2 v2.2.1-0.20240726160203-3f7f396734c3 h1:I/XGsUSnB4ajiWNokzpuRLl8VirnaYyLPwKdjHxOHiE=

pkg/service/service.go

@@ -193,3 +193,17 @@ func (s *Service) DeregisterCreateSIPParticipantTopic() {
 		s.rpcSIPServer.DeregisterCreateSIPParticipantTopic(s.conf.ClusterID)
 	}
 }
+
+func (s *Service) RegisterTransferSIPParticipantTopic(sipCallId string) error {
+	if s.rpcSIPServer != nil {
+		return s.rpcSIPServer.RegisterTransferSIPParticipantTopic(sipCallId)
+	}
+
+	return psrpc.NewErrorf(psrpc.Internal, "RPC server not started")
+}
+
+func (s *Service) DeregisterTransferSIPParticipantTopic(sipCallId string) {
+	if s.rpcSIPServer != nil {
+		s.rpcSIPServer.DeregisterTransferSIPParticipantTopic(sipCallId)
+	}
+}

pkg/sip/client.go

@@ -29,7 +29,7 @@ import (
 	"github.com/livekit/protocol/rpc"
 	"github.com/livekit/protocol/tracer"
 	"github.com/livekit/sip/pkg/config"
-	"github.com/livekit/sip/pkg/errors"
+	siperrors "github.com/livekit/sip/pkg/errors"
 	"github.com/livekit/sip/pkg/stats"
 )
 
@@ -46,6 +46,8 @@ type Client struct {
 	cmu         sync.Mutex
 	activeCalls map[LocalTag]*outboundCall
 	byRemote    map[RemoteTag]*outboundCall
+
+	handler Handler
 }
 
 func NewClient(conf *config.Config, log logger.Logger, mon *stats.Monitor) *Client {
@@ -119,6 +121,10 @@ func (c *Client) Stop() {
 	}
 }
 
+func (c *Client) SetHandler(handler Handler) {
+	c.handler = handler
+}
+
 func (c *Client) CreateSIPParticipant(ctx context.Context, req *rpc.InternalCreateSIPParticipantRequest) (*rpc.InternalCreateSIPParticipantResponse, error) {
 	ctx, span := tracer.Start(ctx, "Client.CreateSIPParticipant")
 	defer span.End()
@@ -127,7 +133,7 @@ func (c *Client) CreateSIPParticipant(ctx context.Context, req *rpc.InternalCrea
 
 func (c *Client) createSIPParticipant(ctx context.Context, req *rpc.InternalCreateSIPParticipantRequest) (*rpc.InternalCreateSIPParticipantResponse, error) {
 	if !c.mon.CanAccept() {
-		return nil, errors.ErrUnavailable
+		return nil, siperrors.ErrUnavailable
 	}
 	if req.CallTo == "" {
 		return nil, fmt.Errorf("call-to number must be set")
@@ -193,6 +199,7 @@ func (c *Client) createSIPParticipant(ctx context.Context, req *rpc.InternalCrea
 		ParticipantIdentity: p.Identity,
 		SipCallId:           req.SipCallId,
 	}, nil
+
 }
 
 func (c *Client) OnRequest(req *sip.Request, tx sip.ServerTransaction) bool {
@@ -201,6 +208,8 @@ func (c *Client) OnRequest(req *sip.Request, tx sip.ServerTransaction) bool {
 		return false
 	case "BYE":
 		return c.onBye(req, tx)
+	case "NOTIFY":
+		return c.onNotify(req, tx)
 	}
 }
 
@@ -219,7 +228,29 @@ func (c *Client) onBye(req *sip.Request, tx sip.ServerTransaction) bool {
 	return true
 }
 
-func (c *Client) CreateSIPParticipantAffinity(ctx context.Context, req *rpc.InternalCreateSIPParticipantRequest) float32 {
-	// TODO: scale affinity based on a number or active calls?
-	return 0.5
+func (c *Client) onNotify(req *sip.Request, tx sip.ServerTransaction) bool {
+	tag, _ := getFromTag(req)
+	c.cmu.Lock()
+	call := c.byRemote[tag]
+	c.cmu.Unlock()
+	if call == nil {
+		return false
+	}
+	call.log.Infow("NOTIFY")
+	go func() {
+		err := call.cc.handleNotify(req, tx)
+
+		code, msg := sipCodeAndMessageFromError(err)
+
+		tx.Respond(sip.NewResponseFromRequest(req, code, msg, nil))
+	}()
+	return true
+}
+
+func (c *Client) RegisterTransferSIPParticipant(sipCallID string, o *outboundCall) error {
+	return c.handler.RegisterTransferSIPParticipantTopic(sipCallID)
+}
+
+func (c *Client) DeregisterTransferSIPParticipant(sipCallID string) {
+	c.handler.DeregisterTransferSIPParticipantTopic(sipCallID)
 }