From 89c7cec2ad957856295a7ff1c48291b5ebcf3a8b Mon Sep 17 00:00:00 2001 From: Denys Smirnov Date: Wed, 24 Jan 2024 20:01:22 +0200 Subject: [PATCH] SIP: New protocol for creating participants. (#2404) --- go.mod | 4 +- go.sum | 10 ++-- pkg/service/interfaces.go | 5 -- pkg/service/redisstore.go | 38 +------------- pkg/service/sip.go | 105 +++++++++++--------------------------- 5 files changed, 38 insertions(+), 124 deletions(-) diff --git a/go.mod b/go.mod index e111fbb21..b13a7d01e 100644 --- a/go.mod +++ b/go.mod @@ -18,7 +18,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 github.com/livekit/mediatransportutil v0.0.0-20231213075826-cccbf2b93d3f - github.com/livekit/protocol v1.9.5-0.20240121141201-9e82495c0485 + github.com/livekit/protocol v1.9.5 github.com/livekit/psrpc v0.5.3-0.20231214055026-06ce27a934c9 github.com/mackerelio/go-osstat v0.2.4 github.com/magefile/mage v1.15.0 @@ -102,6 +102,6 @@ require ( golang.org/x/text v0.14.0 // indirect golang.org/x/tools v0.17.0 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20240116215550-a9fa1716bcac // indirect - google.golang.org/grpc v1.60.1 // indirect + google.golang.org/grpc v1.61.0 // indirect gopkg.in/yaml.v2 v2.4.0 // indirect ) diff --git a/go.sum b/go.sum index cb4e8d092..41f44c6e9 100644 --- a/go.sum +++ b/go.sum @@ -126,10 +126,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-20231213075826-cccbf2b93d3f h1:XHrwGwLNGQB3ZqolH1YdMH/22hgXKr4vm+2M7JKMMGg= github.com/livekit/mediatransportutil v0.0.0-20231213075826-cccbf2b93d3f/go.mod h1:GBzn9xL+mivI1pW+tyExcKgbc0VOc29I9yJsNcAVaAc= -github.com/livekit/protocol v1.9.5-0.20240118112540-cf33ad3861d8 h1:E9s9KFCuKgYWYgaKz0ZmC7K3cPr8Iij77HbnwhQ4JZw= -github.com/livekit/protocol v1.9.5-0.20240118112540-cf33ad3861d8/go.mod h1:Qv55+z0kD0NYp/G0qAaFA4Mjalxt7tsOJwrvV3HymsA= -github.com/livekit/protocol v1.9.5-0.20240121141201-9e82495c0485 h1:X75uVI0+YA7QN28NaVniP4IjhbcDWlktZ3Ec+PHjoHA= -github.com/livekit/protocol v1.9.5-0.20240121141201-9e82495c0485/go.mod h1:Qv55+z0kD0NYp/G0qAaFA4Mjalxt7tsOJwrvV3HymsA= +github.com/livekit/protocol v1.9.5 h1:/I6maM05euoUxrV6je16Qj5yCnCSPZ+nhHzm8akLCVk= +github.com/livekit/protocol v1.9.5/go.mod h1:daddOPw85C9nq6f9w1uiuc1i/He6X2gArlFcKUPELI4= github.com/livekit/psrpc v0.5.3-0.20231214055026-06ce27a934c9 h1:kXXV/NLVDHZ+Gn7xrR+UPpdwbH48n7WReBjLHAzqzhY= github.com/livekit/psrpc v0.5.3-0.20231214055026-06ce27a934c9/go.mod h1:cQjxg1oCxYHhxxv6KJH1gSvdtCHQoRZCHgPdm5N8v2g= github.com/mackerelio/go-osstat v0.2.4 h1:qxGbdPkFo65PXOb/F/nhDKpF2nGmGaCFDLXoZjJTtUs= @@ -425,8 +423,8 @@ golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8T golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= google.golang.org/genproto/googleapis/rpc v0.0.0-20240116215550-a9fa1716bcac h1:nUQEQmH/csSvFECKYRv6HWEyypysidKl2I6Qpsglq/0= google.golang.org/genproto/googleapis/rpc v0.0.0-20240116215550-a9fa1716bcac/go.mod h1:daQN87bsDqDoe316QbbvX60nMoJQa4r6Ds0ZuoAe5yA= -google.golang.org/grpc v1.60.1 h1:26+wFr+cNqSGFcOXcabYC0lUVJVRa2Sb2ortSK7VrEU= -google.golang.org/grpc v1.60.1/go.mod h1:OlCHIeLYqSSsLi6i49B5QGdzaMZK9+M7LXN2FKz4eGM= +google.golang.org/grpc v1.61.0 h1:TOvOcuXn30kRao+gfcvsebNEa5iZIiLkisYEkf7R7o0= +google.golang.org/grpc v1.61.0/go.mod h1:VUbo7IFqmF1QtCAstipjG0GIoq49KvMe9+h1jFLBNJs= google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8= google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0= google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM= diff --git a/pkg/service/interfaces.go b/pkg/service/interfaces.go index ec8359e9d..73c7d4a67 100644 --- a/pkg/service/interfaces.go +++ b/pkg/service/interfaces.go @@ -88,9 +88,4 @@ type SIPStore interface { LoadSIPDispatchRule(ctx context.Context, sipDispatchRuleID string) (*livekit.SIPDispatchRuleInfo, error) ListSIPDispatchRule(ctx context.Context) ([]*livekit.SIPDispatchRuleInfo, error) DeleteSIPDispatchRule(ctx context.Context, info *livekit.SIPDispatchRuleInfo) error - - StoreSIPParticipant(ctx context.Context, info *livekit.SIPParticipantInfo) error - LoadSIPParticipant(ctx context.Context, sipParticipantID string) (*livekit.SIPParticipantInfo, error) - ListSIPParticipant(ctx context.Context) ([]*livekit.SIPParticipantInfo, error) - DeleteSIPParticipant(ctx context.Context, info *livekit.SIPParticipantInfo) error } diff --git a/pkg/service/redisstore.go b/pkg/service/redisstore.go index dfb4c3626..91b437f40 100644 --- a/pkg/service/redisstore.go +++ b/pkg/service/redisstore.go @@ -26,11 +26,12 @@ import ( "github.com/redis/go-redis/v9" "google.golang.org/protobuf/proto" - "github.com/livekit/livekit-server/version" "github.com/livekit/protocol/ingress" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" "github.com/livekit/protocol/utils" + + "github.com/livekit/livekit-server/version" ) const ( @@ -53,7 +54,6 @@ const ( SIPTrunkKey = "sip_trunk" SIPDispatchRuleKey = "sip_dispatch_rule" - SIPParticipantKey = "sip_participant" // RoomParticipantsPrefix is hash of participant_name => ParticipantInfo RoomParticipantsPrefix = "room_participants:" @@ -908,37 +908,3 @@ func (s *RedisStore) ListSIPDispatchRule(ctx context.Context) (infos []*livekit. return infos, err } - -func (s *RedisStore) StoreSIPParticipant(ctx context.Context, info *livekit.SIPParticipantInfo) error { - data, err := proto.Marshal(info) - if err != nil { - return err - } - - return s.rc.HSet(s.ctx, SIPParticipantKey, info.SipParticipantId, data).Err() -} -func (s *RedisStore) LoadSIPParticipant(ctx context.Context, sipParticipantId string) (*livekit.SIPParticipantInfo, error) { - info := &livekit.SIPParticipantInfo{} - if err := s.loadOne(ctx, SIPParticipantKey, sipParticipantId, info, ErrSIPParticipantNotFound); err != nil { - return nil, err - } - - return info, nil -} - -func (s *RedisStore) DeleteSIPParticipant(ctx context.Context, info *livekit.SIPParticipantInfo) error { - return s.rc.HDel(s.ctx, SIPParticipantKey, info.SipParticipantId).Err() -} - -func (s *RedisStore) ListSIPParticipant(ctx context.Context) (infos []*livekit.SIPParticipantInfo, err error) { - err = s.loadMany(ctx, SIPParticipantKey, func() proto.Message { - infos = append(infos, &livekit.SIPParticipantInfo{}) - return infos[len(infos)-1] - }) - - return infos, err -} - -func (s *RedisStore) SendSIPParticipantDTMF(ctx context.Context, info *livekit.SendSIPParticipantDTMFRequest) (*livekit.SIPParticipantDTMFInfo, error) { - return nil, fmt.Errorf("TODO") -} diff --git a/pkg/service/sip.go b/pkg/service/sip.go index 8bba2fdca..a95dbad09 100644 --- a/pkg/service/sip.go +++ b/pkg/service/sip.go @@ -16,6 +16,7 @@ package service import ( "context" + "time" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" @@ -161,89 +162,43 @@ func (s *SIPService) CreateSIPParticipant(ctx context.Context, req *livekit.Crea return nil, ErrSIPNotConnected } - info := &livekit.SIPParticipantInfo{ - SipParticipantId: utils.NewGuid(utils.SIPParticipantPrefix), - SipTrunkId: req.SipTrunkId, - SipCallTo: req.SipCallTo, + AppendLogFields(ctx, "room", req.RoomName, "trunk", req.SipTrunkId, "to", req.SipCallTo) + ireq := &rpc.InternalCreateSIPParticipantRequest{ + CallTo: req.SipCallTo, RoomName: req.RoomName, ParticipantIdentity: req.ParticipantIdentity, } - - if err := s.store.StoreSIPParticipant(ctx, info); err != nil { - return nil, err - } - s.updateParticipant(ctx, info) - return info, nil -} - -func (s *SIPService) updateParticipant(ctx context.Context, info *livekit.SIPParticipantInfo) { - AppendLogFields(ctx, "participantId", info.SipParticipantId, "room", info.RoomName, "trunk", info.SipTrunkId, "to", info.SipCallTo) - req := &rpc.InternalUpdateSIPParticipantRequest{ - ParticipantId: info.SipParticipantId, - CallTo: info.SipCallTo, - RoomName: info.RoomName, - ParticipantIdentity: info.ParticipantIdentity, - } - if info.SipTrunkId != "" { - trunk, err := s.store.LoadSIPTrunk(ctx, info.SipTrunkId) + if req.SipTrunkId != "" { + trunk, err := s.store.LoadSIPTrunk(ctx, req.SipTrunkId) if err != nil { logger.Errorw("cannot get trunk to update sip participant", err) - return + return nil, err } - req.Address = trunk.OutboundAddress - req.Number = trunk.OutboundNumber - req.Username = trunk.OutboundUsername - req.Password = trunk.OutboundPassword + ireq.Address = trunk.OutboundAddress + ireq.Number = trunk.OutboundNumber + ireq.Username = trunk.OutboundUsername + ireq.Password = trunk.OutboundPassword } - if _, err := s.psrpcClient.UpdateSIPParticipant(ctx, "", req); err != nil { + + // CreateSIPParticipant will wait for LiveKit Participant to be created and that can take some time. + // Thus, we must set a higher deadline for it, if it's not set already. + // TODO: support context timeouts in psrpc + timeout := 30 * time.Second + if deadline, ok := ctx.Deadline(); ok { + timeout = time.Until(deadline) + } else { + var cancel func() + ctx, cancel = context.WithTimeout(ctx, timeout) + defer cancel() + } + resp, err := s.psrpcClient.CreateSIPParticipant(ctx, "", ireq, psrpc.WithRequestTimeout(timeout)) + if err != nil { logger.Errorw("cannot update sip participant", err) - } -} - -func (s *SIPService) ListSIPParticipant(ctx context.Context, req *livekit.ListSIPParticipantRequest) (*livekit.ListSIPParticipantResponse, error) { - if s.store == nil { - return nil, ErrSIPNotConnected - } - - participants, err := s.store.ListSIPParticipant(ctx) - if err != nil { return nil, err } - - return &livekit.ListSIPParticipantResponse{Items: participants}, nil -} - -func (s *SIPService) DeleteSIPParticipant(ctx context.Context, req *livekit.DeleteSIPParticipantRequest) (*livekit.SIPParticipantInfo, error) { - if s.store == nil { - return nil, ErrSIPNotConnected - } - - info, err := s.store.LoadSIPParticipant(ctx, req.SipParticipantId) - if err != nil { - return nil, err - } - - if err = s.store.DeleteSIPParticipant(ctx, info); err != nil { - return nil, err - } - // These indicate that the call should be disconnected - info.SipTrunkId = "" - info.SipCallTo = "" - s.updateParticipant(ctx, info) - return info, nil -} - -func (s *SIPService) SendSIPParticipantDTMF(ctx context.Context, req *livekit.SendSIPParticipantDTMFRequest) (*livekit.SIPParticipantDTMFInfo, error) { - if s.store == nil { - return nil, ErrSIPNotConnected - } - AppendLogFields(ctx, "participantId", req.SipParticipantId) - _, err := s.psrpcClient.SendSIPParticipantDTMF(ctx, &rpc.InternalSendSIPParticipantDTMFRequest{ - ParticipantId: req.SipParticipantId, - Digits: req.Digits, - }) - if err != nil { - logger.Errorw("cannot send dtmf to sip participant", err) - } - return nil, err + return &livekit.SIPParticipantInfo{ + ParticipantId: resp.ParticipantId, + ParticipantIdentity: resp.ParticipantIdentity, + RoomName: req.RoomName, + }, nil }