Transfer participant integration test

go.mod

@@ -12,23 +12,23 @@ require (
 	github.com/icholy/digest v0.1.23
 	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/mediatransportutil v0.0.0-20240730083616-559fa5ece598
 	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/livekit/server-sdk-go/v2 v2.2.2-0.20240920185319-a83c50186010
 	github.com/mjibson/go-dsp v0.0.0-20180508042940-11479a337f12
 	github.com/ory/dockertest/v3 v3.10.0
-	github.com/pion/interceptor v0.1.29
-	github.com/pion/rtp v1.8.6
+	github.com/pion/interceptor v0.1.30
+	github.com/pion/rtp v1.8.9
 	github.com/pion/sdp/v3 v3.0.9
-	github.com/pion/webrtc/v3 v3.2.47
+	github.com/pion/webrtc/v3 v3.3.1
 	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
+	golang.org/x/exp v0.0.0-20240823005443-9b4947da3948
 	google.golang.org/protobuf v1.34.2
 	gopkg.in/hraban/opus.v2 v2.0.0-20230925203106-0188a62cb302
 	gopkg.in/yaml.v3 v3.0.1
@@ -86,17 +86,17 @@ require (
 	github.com/opencontainers/go-digest v1.0.0 // indirect
 	github.com/opencontainers/image-spec v1.0.2 // indirect
 	github.com/opencontainers/runc v1.1.12 // indirect
-	github.com/pion/datachannel v1.5.6 // indirect
-	github.com/pion/dtls/v2 v2.2.11 // indirect
-	github.com/pion/ice/v2 v2.3.29 // indirect
+	github.com/pion/datachannel v1.5.8 // indirect
+	github.com/pion/dtls/v2 v2.2.12 // indirect
+	github.com/pion/ice/v2 v2.3.34 // indirect
 	github.com/pion/logging v0.2.2 // indirect
 	github.com/pion/mdns v0.0.12 // indirect
 	github.com/pion/randutil v0.1.0 // indirect
 	github.com/pion/rtcp v1.2.14 // indirect
-	github.com/pion/sctp v1.8.16 // indirect
-	github.com/pion/srtp/v2 v2.0.18 // indirect
+	github.com/pion/sctp v1.8.19 // indirect
+	github.com/pion/srtp/v2 v2.0.20 // indirect
 	github.com/pion/stun v0.6.1 // indirect
-	github.com/pion/transport/v2 v2.2.5 // indirect
+	github.com/pion/transport/v2 v2.2.10 // indirect
 	github.com/pion/turn/v2 v2.1.6 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_model v0.5.0 // indirect
@@ -108,6 +108,7 @@ require (
 	github.com/satori/go.uuid v1.2.1-0.20181028125025-b2ce2384e17b // indirect
 	github.com/stoewer/go-strcase v1.3.0 // indirect
 	github.com/twitchtv/twirp v8.1.3+incompatible // indirect
+	github.com/wlynxg/anet v0.0.3 // indirect
 	github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb // indirect
 	github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 // indirect
 	github.com/xeipuuv/gojsonschema v1.2.0 // indirect
@@ -117,13 +118,13 @@ require (
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.27.0 // indirect
 	go.uber.org/zap/exp v0.2.0 // indirect
-	golang.org/x/crypto v0.25.0 // indirect
-	golang.org/x/mod v0.19.0 // indirect
-	golang.org/x/net v0.27.0 // indirect
-	golang.org/x/sync v0.7.0 // indirect
-	golang.org/x/sys v0.22.0 // indirect
-	golang.org/x/text v0.16.0 // indirect
-	golang.org/x/tools v0.23.0 // indirect
+	golang.org/x/crypto v0.26.0 // indirect
+	golang.org/x/mod v0.20.0 // indirect
+	golang.org/x/net v0.28.0 // indirect
+	golang.org/x/sync v0.8.0 // indirect
+	golang.org/x/sys v0.23.0 // indirect
+	golang.org/x/text v0.17.0 // indirect
+	golang.org/x/tools v0.24.0 // indirect
 	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

pkg/sip/inbound.go

@@ -744,6 +744,8 @@ func (c *inboundCall) transferCall(ctx context.Context, transferTo string) error
 		return err
 	}
 
+	c.log.Infow("inbound call tranferred", "transferTo", transferTo)
+
 	// This is needed to actually terminate the session before a media timeout
 	c.Close()
 
@@ -758,7 +760,7 @@ func (s *Server) newInbound(id LocalTag, invite *sip.Request, inviteTx sip.Serve
 		invite:    invite,
 		inviteTx:  inviteTx,
 		cancelled: make(chan struct{}),
-		referDone: make(chan error, 1),
+		referDone: make(chan error), // Do not buffer the channel to avoid reading a result for an old request
 	}
 	c.from, _ = invite.From()
 	if c.from != nil {
@@ -1094,14 +1096,14 @@ func (c *sipInbound) handleNotify(req *sip.Request, tx sip.ServerTransaction) er
 			// Success
 			select {
 			case c.referDone <- nil:
-			default:
+			case <-time.After(5 * time.Second):
 			}
 		default:
 			// Failure
 			select {
 			// TODO be more specific in the reported error
 			case c.referDone <- psrpc.NewErrorf(psrpc.Canceled, "call transfer failed"):
-			default:
+			case <-time.After(5 * time.Second):
 			}
 		}
 	}

pkg/sip/outbound.go

@@ -22,6 +22,7 @@ import (
 	"net/netip"
 	"sort"
 	"sync"
+	"time"
 
 	"github.com/emiago/sipgo/sip"
 	"github.com/frostbyte73/core"
@@ -391,6 +392,8 @@ func (c *outboundCall) transferCall(ctx context.Context, transferTo string) erro
 		return err
 	}
 
+	c.log.Infow("outbound l tranferred", "transferTo", transferTo)
+
 	// This is needed to actually terminate the session before a media timeout
 	c.CloseWithReason(CallHangup, "call transferred")
 
@@ -409,7 +412,7 @@ func (c *Client) newOutbound(id LocalTag, from URI) *sipOutbound {
 		c:         c,
 		id:        id,
 		from:      fromHeader,
-		referDone: make(chan error, 1),
+		referDone: make(chan error), // Do not buffer the channel to avoid reading a result for an old request
 	}
 }
 
@@ -725,14 +728,14 @@ func (c *sipOutbound) handleNotify(req *sip.Request, tx sip.ServerTransaction) e
 			// Success
 			select {
 			case c.referDone <- nil:
-			default:
+			case <-time.After(5 * time.Second):
 			}
 		default:
 			// Failure
 			select {
 			// TODO be more specific in the reported error
 			case c.referDone <- psrpc.NewErrorf(psrpc.Canceled, "call transfer failed"):
-			default:
+			case <-time.After(5 * time.Second):
 			}
 		}
 	}

pkg/sip/protocol.go

@@ -171,6 +171,13 @@ func parseNotifyBody(body string) (int, error) {
 
 func handleNotify(req *sip.Request) (method sip.RequestMethod, cseq uint32, status int, err error) {
 	event := req.GetHeader("Event")
+	if event == nil {
+		event = req.GetHeader("o")
+	}
+	if event == nil {
+		return "", 0, 0, psrpc.NewErrorf(psrpc.MalformedRequest, "no event in NOTIFY request")
+	}
+
 	var cseq64 uint64
 
 	if m := referIdRegexp.FindStringSubmatch(strings.ToLower(event.Value())); len(m) > 0 {

pkg/sip/service.go

@@ -124,7 +124,7 @@ func (s *Service) CreateSIPParticipantAffinity(ctx context.Context, req *rpc.Int
 func (s *Service) TransferSIPParticipant(ctx context.Context, req *rpc.InternalTransferSIPParticipantRequest) (*emptypb.Empty, error) {
 	s.log.Infow("transfering SIP call", "callID", req.SipCallId, "transferTo", req.TransferTo)
 
-	ctx, done := context.WithTimeout(ctx, 30*time.Second)
+	ctx, done := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second)
 	defer done()
 
 	// Look for call both in client (outbound) and server (inbound)

pkg/siptest/client.go

@@ -34,6 +34,7 @@ import (
 	"github.com/at-wat/ebml-go/webm"
 	"github.com/emiago/sipgo"
 	"github.com/emiago/sipgo/sip"
+	"github.com/frostbyte73/core"
 	"github.com/icholy/digest"
 	"github.com/pion/sdp/v3"
 
@@ -56,6 +57,7 @@ type ClientConfig struct {
 	OnBye          func()
 	OnMediaTimeout func()
 	OnDTMF         func(ev dtmf.Event)
+	OnRefer        func(req *sip.Request)
 	Codec          string
 }
 
@@ -146,6 +148,14 @@ func NewClient(id string, conf ClientConfig) (*Client, error) {
 		default:
 		}
 	})
+	cli.sipServer.OnRefer(func(req *sip.Request, tx sip.ServerTransaction) {
+		if conf.OnRefer != nil {
+			conf.OnRefer(req)
+		}
+
+		err = tx.Respond(sip.NewResponseFromRequest(req, 202, "Accepted", nil))
+		tx.Terminate()
+	})
 
 	return cli, nil
 }
@@ -168,7 +178,8 @@ type Client struct {
 	inviteReq     *sip.Request
 	inviteResp    *sip.Response
 	recordHandler atomic.Pointer[rtp.Handler]
-	closed        atomic.Bool
+	lastCSeq      atomic.Uint32
+	closed        core.Fuse
 }
 
 func (c *Client) LocalIP() string {
@@ -183,23 +194,22 @@ func (c *Client) RemoteHeaders() []sip.Header {
 }
 
 func (c *Client) Close() {
-	if !c.closed.CompareAndSwap(false, true) {
-		return
-	}
-	if c.mediaConn != nil {
-		c.mediaConn.Close()
-	}
-	if c.inviteResp != nil {
-		c.sendBye()
-		c.inviteReq = nil
-		c.inviteResp = nil
-	}
-	if c.sipClient != nil {
-		c.sipClient.Close()
-	}
-	if c.sipServer != nil {
-		c.sipServer.Close()
-	}
+	c.closed.Once(func() {
+		if c.mediaConn != nil {
+			c.mediaConn.Close()
+		}
+		if c.inviteResp != nil {
+			c.sendBye()
+			c.inviteReq = nil
+			c.inviteResp = nil
+		}
+		if c.sipClient != nil {
+			c.sipClient.Close()
+		}
+		if c.sipServer != nil {
+			c.sipServer.Close()
+		}
+	})
 }
 
 func (c *Client) setupRTPReceiver() {
@@ -322,6 +332,11 @@ func (c *Client) Dial(ip string, uri string, number string, headers map[string]s
 	}
 	c.inviteReq = req
 	c.inviteResp = resp
+
+	if h, ok := req.CSeq(); ok {
+		c.lastCSeq.Store(h.SeqNo)
+	}
+
 	c.mediaConn.SetDestAddr(dstAddr)
 	c.log.Debug("client connected", "media-dst", dstAddr)
 	return nil
@@ -361,6 +376,10 @@ func (c *Client) sendBye() {
 	req := sip.NewByeRequest(c.inviteReq, c.inviteResp, nil)
 	req.AppendHeader(sip.NewHeader("User-Agent", "LiveKit"))
 
+	cseq := c.lastCSeq.Add(1)
+	cseqH, _ := req.CSeq()
+	cseqH.SeqNo = cseq
+
 	tx, err := c.sipClient.TransactionRequest(req)
 	if err != nil {
 		return
@@ -381,6 +400,96 @@ func (c *Client) SendDTMF(digits string) error {
 	return dtmf.Write(context.Background(), c.audioOut, c.mediaDTMF, c.mediaAudio.GetCurrentTimestamp(), digits)
 }
 
+func (c *Client) SendNotify(eventReq *sip.Request, notifyStatus string) error {
+	var recipient *sip.Uri
+
+	if contact, ok := eventReq.Contact(); ok {
+		recipient = &contact.Address
+	} else if from, ok := eventReq.From(); ok {
+		recipient = &from.Address
+	} else {
+		errors.New("missing destination address")
+	}
+
+	req := sip.NewRequest(sip.NOTIFY, recipient)
+
+	req.SipVersion = eventReq.SipVersion
+	sip.CopyHeaders("Via", eventReq, req)
+
+	if len(eventReq.GetHeaders("Route")) > 0 {
+		sip.CopyHeaders("Route", eventReq, req)
+	} else {
+		hdrs := c.inviteResp.GetHeaders("Record-Route")
+		for i := len(hdrs) - 1; i >= 0; i-- {
+			rrh, ok := hdrs[i].(*sip.RecordRouteHeader)
+			if !ok {
+				continue
+			}
+
+			h := rrh.Clone()
+			req.AppendHeader(h)
+		}
+	}
+
+	maxForwardsHeader := sip.MaxForwardsHeader(70)
+	req.AppendHeader(&maxForwardsHeader)
+
+	if to, ok := eventReq.To(); ok {
+		req.AppendHeader((*sip.FromHeader)(to))
+	} else {
+		return errors.New("missing To header in REFER request")
+	}
+
+	if from, ok := eventReq.From(); ok {
+		req.AppendHeader((*sip.ToHeader)(from))
+	} else {
+		return errors.New("missing From header in REFER request")
+	}
+
+	if callId, ok := eventReq.CallID(); ok {
+		req.AppendHeader(callId)
+	}
+
+	ct := sip.ContentTypeHeader("message/sipfrag")
+	req.AppendHeader(&ct)
+
+	cseq := c.lastCSeq.Add(1)
+	cseqH := &sip.CSeqHeader{
+		SeqNo:      cseq,
+		MethodName: sip.NOTIFY,
+	}
+	req.AppendHeader(cseqH)
+
+	req.SetTransport(eventReq.Transport())
+	req.SetSource(eventReq.Destination())
+	req.SetDestination(eventReq.Source())
+
+	if eventCSeq, ok := eventReq.CSeq(); ok {
+		req.AppendHeader(sip.NewHeader("Event", fmt.Sprintf("refer;id=%d", eventCSeq.SeqNo)))
+	} else {
+		return errors.New("missing CSeq header in REFER request")
+	}
+
+	req.SetBody([]byte(notifyStatus))
+
+	tx, err := c.sipClient.TransactionRequest(req)
+	if err != nil {
+		return err
+	}
+	defer tx.Terminate()
+
+	resp, err := getResponse(tx)
+	if err != nil {
+		return err
+	}
+
+	if resp.StatusCode != sip.StatusOK {
+		return fmt.Errorf("NOTIFY failed with status %d", resp.StatusCode)
+	}
+
+	return nil
+}
+
 func (c *Client) createOffer() ([]byte, error) {
 	sessionId := rand.Uint64()