fix jitter buffer (#325)

Author: frostbyte73

.gitignore

@@ -4,4 +4,4 @@ bin/
 cmd/server/server
 
 test/config.yaml
-test/client/*.mkv
+test/*/*.mkv

pkg/media/rtp/jitter.go

@@ -39,7 +39,7 @@ type jitterHandler struct {
 }
 
 func (h *jitterHandler) HandleRTP(p *rtp.Packet) error {
-	h.buf.Push(p)
+	h.buf.Push(p.Clone())
 	var last error
 	for _, p := range h.buf.Pop(false) {
 		if err := h.h.HandleRTP(p); err != nil {

pkg/siptest/client.go

@@ -538,6 +538,7 @@ func (c *Client) createOffer() ([]byte, error) {
 	return offer.Marshal()
 }
 
+// Sends PCM audio from a webm file
 func (c *Client) SendAudio(path string) error {
 	f, err := os.Open(path)
 	if err != nil {

test/cloud/cloud.go

@@ -0,0 +1,67 @@
+package cloud
+
+import (
+	"context"
+	"time"
+
+	"github.com/livekit/protocol/livekit"
+	"github.com/livekit/protocol/logger"
+	"github.com/livekit/protocol/rpc"
+	"github.com/livekit/protocol/sip"
+	"github.com/livekit/psrpc"
+)
+
+type CloudTestService struct {
+	conf *IntegrationConfig
+
+	psrpcClient rpc.SIPInternalClient
+}
+
+func NewCloudTestService(conf *IntegrationConfig, bus psrpc.MessageBus) (*CloudTestService, error) {
+	c, err := rpc.NewSIPInternalClient(bus)
+	if err != nil {
+		return nil, err
+	}
+
+	return &CloudTestService{
+		conf:        conf,
+		psrpcClient: c,
+	}, nil
+}
+
+func (s *CloudTestService) CreateSIPParticipant(ctx context.Context, req *livekit.CreateSIPParticipantRequest) (*livekit.SIPParticipantInfo, error) {
+	token, err := sip.BuildSIPToken(sip.SIPTokenParams{
+		APIKey:                s.conf.ApiKey,
+		APISecret:             s.conf.ApiSecret,
+		RoomName:              req.RoomName,
+		ParticipantIdentity:   req.ParticipantIdentity,
+		ParticipantName:       req.ParticipantName,
+		ParticipantMetadata:   req.ParticipantMetadata,
+		ParticipantAttributes: req.ParticipantAttributes,
+	})
+	if err != nil {
+		logger.Errorw("failed to create SIP token", err)
+		return nil, err
+	}
+
+	callID := sip.NewCallID()
+	trunk := &livekit.SIPOutboundTrunkInfo{}
+	r, err := rpc.NewCreateSIPParticipantRequest(ProjectID, callID, Host, s.conf.WsUrl, token, req, trunk)
+	if err != nil {
+		logger.Errorw("failed to build CreateSIPParticipantRequest", err)
+		return nil, err
+	}
+
+	resp, err := s.psrpcClient.CreateSIPParticipant(ctx, s.conf.ClusterID, r, psrpc.WithRequestTimeout(time.Second*30))
+	if err != nil {
+		logger.Errorw("failed to create SIP participant", err)
+		return nil, err
+	}
+
+	return &livekit.SIPParticipantInfo{
+		ParticipantId:       resp.ParticipantId,
+		ParticipantIdentity: resp.ParticipantIdentity,
+		RoomName:            req.RoomName,
+		SipCallId:           r.SipCallId,
+	}, nil
+}

test/cloud/config.go

@@ -0,0 +1,36 @@
+package cloud
+
+import (
+	"os"
+
+	"github.com/livekit/sip/pkg/config"
+)
+
+const (
+	Host      = "integration.sip.livekit.cloud"
+	ProjectID = "sip_integration"
+	Uri       = "integration.pstn.twilio.com"
+)
+
+type IntegrationConfig struct {
+	*config.Config
+
+	RoomName string
+}
+
+func NewIntegrationConfig() (*IntegrationConfig, error) {
+	c := &IntegrationConfig{
+		Config: &config.Config{
+			ApiKey:             os.Getenv("LIVEKIT_API_KEY"),
+			ApiSecret:          os.Getenv("LIVEKIT_API_SECRET"),
+			WsUrl:              os.Getenv("LIVEKIT_WS_URL"),
+			ServiceName:        "sip",
+			EnableJitterBuffer: true,
+		},
+		RoomName: "test",
+	}
+	if err := c.Config.Init(); err != nil {
+		return nil, err
+	}
+	return c, nil
+}

test/cloud/integration_test.go

@@ -0,0 +1,53 @@
+package cloud
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/livekit/protocol/logger"
+	"github.com/livekit/psrpc"
+)
+
+func TestSIP(t *testing.T) {
+	logger.InitFromConfig(&logger.Config{
+		JSON:  false,
+		Level: "debug",
+	}, "sip")
+
+	conf, err := NewIntegrationConfig()
+	require.NoError(t, err)
+
+	if conf.ApiKey == "" || conf.ApiSecret == "" || conf.WsUrl == "" {
+		t.Skip("missing env vars")
+	}
+
+	bus := psrpc.NewLocalMessageBus()
+	svc, err := NewService(conf, bus)
+	require.NoError(t, err)
+	defer svc.Stop(true)
+
+	go func() {
+		_ = svc.Run()
+	}()
+
+	a, err := NewPhoneClient(false)
+	require.NoError(t, err)
+	defer a.Close()
+
+	b, err := NewPhoneClient(true)
+	require.NoError(t, err)
+	defer b.Close()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go b.SendSilence(ctx)
+
+	go a.SendAudio("audio-pcm.mkv")
+
+	time.Sleep(time.Second * 5)
+	_ = a.SendDTMF("2345")
+	time.Sleep(time.Second * 5)
+}

test/cloud/io.go

@@ -0,0 +1,68 @@
+package cloud
+
+import (
+	"context"
+	"fmt"
+
+	"google.golang.org/protobuf/types/known/emptypb"
+
+	"github.com/livekit/protocol/logger"
+	"github.com/livekit/protocol/rpc"
+	"github.com/livekit/protocol/sip"
+	"github.com/livekit/psrpc"
+)
+
+const (
+	TrunkID        = "ST_INTEGRATION"
+	DispatchRuleID = "SDR_INTEGRATION"
+)
+
+type IOTestClient struct {
+	rpc.IOInfoClient
+
+	conf *IntegrationConfig
+}
+
+func NewIOTestClient(conf *IntegrationConfig) rpc.IOInfoClient {
+	return &IOTestClient{conf: conf}
+}
+
+func (c *IOTestClient) GetSIPTrunkAuthentication(_ context.Context, _ *rpc.GetSIPTrunkAuthenticationRequest, _ ...psrpc.RequestOption) (*rpc.GetSIPTrunkAuthenticationResponse, error) {
+	return &rpc.GetSIPTrunkAuthenticationResponse{
+		SipTrunkId: TrunkID,
+		ProjectId:  ProjectID,
+	}, nil
+}
+
+func (c *IOTestClient) EvaluateSIPDispatchRules(_ context.Context, _ *rpc.EvaluateSIPDispatchRulesRequest, _ ...psrpc.RequestOption) (*rpc.EvaluateSIPDispatchRulesResponse, error) {
+	identity := fmt.Sprintf("phone-%d", num.Load())
+	token, err := sip.BuildSIPToken(sip.SIPTokenParams{
+		APIKey:              c.conf.ApiKey,
+		APISecret:           c.conf.ApiSecret,
+		RoomName:            c.conf.RoomName,
+		ParticipantIdentity: identity,
+		ParticipantName:     identity,
+		RoomConfig:          nil,
+	})
+	if err != nil {
+		logger.Errorw("failed to build SIP token", err)
+		return nil, err
+	}
+
+	return &rpc.EvaluateSIPDispatchRulesResponse{
+		RoomName:            c.conf.RoomName,
+		ParticipantIdentity: identity,
+		ParticipantName:     identity,
+		Token:               token,
+		WsUrl:               c.conf.WsUrl,
+		Result:              rpc.SIPDispatchResult_ACCEPT,
+		SipTrunkId:          TrunkID,
+		SipDispatchRuleId:   DispatchRuleID,
+		ProjectId:           ProjectID,
+		RoomConfig:          nil,
+	}, nil
+}
+
+func (c *IOTestClient) UpdateSIPCallState(_ context.Context, _ *rpc.UpdateSIPCallStateRequest, _ ...psrpc.RequestOption) (*emptypb.Empty, error) {
+	return &emptypb.Empty{}, nil
+}

test/cloud/phone.go

@@ -0,0 +1,65 @@
+package cloud
+
+import (
+	"fmt"
+	"os"
+	"sync/atomic"
+
+	"github.com/livekit/sip/pkg/media/dtmf"
+	"github.com/livekit/sip/pkg/media/g711"
+	"github.com/livekit/sip/pkg/siptest"
+)
+
+const (
+	to    = 15550100000
+	codec = g711.ULawSDPName
+)
+
+var num atomic.Int64
+
+type PhoneClient struct {
+	*siptest.Client
+
+	rec *os.File
+}
+
+func NewPhoneClient(record bool) (*PhoneClient, error) {
+	n := num.Add(1)
+	id := fmt.Sprintf("phone-%d", n)
+
+	c, err := siptest.NewClient(id, siptest.ClientConfig{
+		Number: fmt.Sprintf("+%d", to+n),
+		Codec:  codec,
+		OnBye:  func() {},
+		OnDTMF: func(ev dtmf.Event) {
+			fmt.Println("DTMF C:", ev.Code, " D:", string(ev.Digit))
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	p := &PhoneClient{
+		Client: c,
+	}
+	if record {
+		p.rec, err = os.Create(fmt.Sprintf("%s.mkv", id))
+		if err != nil {
+			return nil, err
+		}
+		c.Record(p.rec)
+	}
+
+	if err = p.Client.Dial(p.LocalIP()+":5060", Uri, fmt.Sprintf("+%d", to), nil); err != nil {
+		return nil, err
+	}
+
+	return p, nil
+}
+
+func (p *PhoneClient) Close() {
+	p.Client.Close()
+	if rec := p.rec; rec != nil {
+		_ = p.rec.Close()
+	}
+}

test/cloud/service.go

@@ -0,0 +1,32 @@
+package cloud
+
+import (
+	"github.com/livekit/protocol/logger"
+	"github.com/livekit/protocol/rpc"
+	"github.com/livekit/psrpc"
+	"github.com/livekit/sip/pkg/service"
+	"github.com/livekit/sip/pkg/sip"
+	"github.com/livekit/sip/pkg/stats"
+)
+
+func NewService(conf *IntegrationConfig, bus psrpc.MessageBus) (*service.Service, error) {
+	psrpcClient := NewIOTestClient(conf)
+
+	mon, err := stats.NewMonitor(conf.Config)
+	if err != nil {
+		return nil, err
+	}
+
+	sipsrv, err := sip.NewService("", conf.Config, mon, logger.GetLogger(), func(projectID string) rpc.IOInfoClient { return psrpcClient })
+	if err != nil {
+		return nil, err
+	}
+	svc := service.NewService(conf.Config, logger.GetLogger(), sipsrv, sipsrv.Stop, sipsrv.ActiveCalls, psrpcClient, bus, mon)
+	sipsrv.SetHandler(svc)
+
+	if err = sipsrv.Start(); err != nil {
+		return nil, err
+	}
+
+	return svc, nil
+}