Set extra headers earlier for outbound calls. (#324)

Author: dennwc

pkg/sip/inbound.go

@@ -356,7 +356,7 @@ func (s *Server) newInboundCall(
 	extra map[string]string,
 ) *inboundCall {
 	// Map known headers immediately on join. The rest of the mapping will be available later.
-	extra = HeadersToAttrs(extra, nil, 0, cc)
+	extra = HeadersToAttrs(extra, nil, 0, cc, nil)
 	c := &inboundCall{
 		s:          s,
 		log:        log,
@@ -472,7 +472,7 @@ func (c *inboundCall) handleInvite(ctx context.Context, req *sip.Request, trunkI
 		}
 	}
 	p := &disp.Room.Participant
-	p.Attributes = HeadersToAttrs(p.Attributes, disp.HeadersToAttributes, disp.IncludeHeaders, c.cc)
+	p.Attributes = HeadersToAttrs(p.Attributes, disp.HeadersToAttributes, disp.IncludeHeaders, c.cc, nil)
 	if disp.MaxCallDuration <= 0 || disp.MaxCallDuration > maxCallDuration {
 		disp.MaxCallDuration = maxCallDuration
 	}

pkg/sip/outbound.go

@@ -397,7 +397,9 @@ func (c *outboundCall) connectMedia() {
 	c.media.HandleDTMF(c.handleDTMF)
 }
 
-func sipResponse(ctx context.Context, tx sip.ClientTransaction, stop <-chan struct{}, setState func(code sip.StatusCode)) (*sip.Response, error) {
+type sipRespFunc func(code sip.StatusCode, hdrs Headers)
+
+func sipResponse(ctx context.Context, tx sip.ClientTransaction, stop <-chan struct{}, setState sipRespFunc) (*sip.Response, error) {
 	cnt := 0
 	for {
 		select {
@@ -411,12 +413,12 @@ func sipResponse(ctx context.Context, tx sip.ClientTransaction, stop <-chan stru
 			return nil, psrpc.NewErrorf(psrpc.Canceled, "transaction failed to complete (%d intermediate responses)", cnt)
 		case res := <-tx.Responses():
 			status := res.StatusCode
+			if setState != nil {
+				setState(res.StatusCode, res.Headers())
+			}
 			if status/100 != 1 { // != 1xx
 				return res, nil
 			}
-			if setState != nil {
-				setState(res.StatusCode)
-			}
 			// continue
 			cnt++
 		}
@@ -442,6 +444,18 @@ func (c *outboundCall) setStatus(v CallStatus) {
 	})
 }
 
+func (c *outboundCall) setExtraAttrs(hdrToAttr map[string]string, opts livekit.SIPHeaderOptions, cc Signaling, hdrs Headers) {
+	extra := HeadersToAttrs(nil, hdrToAttr, opts, cc, hdrs)
+	if c.lkRoom != nil && len(extra) != 0 {
+		room := c.lkRoom.Room()
+		if room != nil {
+			room.LocalParticipant.SetAttributes(extra)
+		} else {
+			c.log.Warnw("could not set attributes on nil room", nil, "attrs", extra)
+		}
+	}
+}
+
 func (c *outboundCall) sipSignal(ctx context.Context) error {
 	ctx, span := tracer.Start(ctx, "outboundCall.sipSignal")
 	defer span.End()
@@ -479,11 +493,15 @@ func (c *outboundCall) sipSignal(ctx context.Context) error {
 	toUri := CreateURIFromUserAndAddress(c.sipConf.to, c.sipConf.address, TransportFrom(c.sipConf.transport))
 
 	ringing := false
-	sdpResp, err := c.cc.Invite(ctx, toUri, c.sipConf.user, c.sipConf.pass, c.sipConf.headers, sdpOfferData, func(code sip.StatusCode) {
-		if !ringing && code >= sip.StatusRinging {
+	sdpResp, err := c.cc.Invite(ctx, toUri, c.sipConf.user, c.sipConf.pass, c.sipConf.headers, sdpOfferData, func(code sip.StatusCode, hdrs Headers) {
+		if code == sip.StatusOK {
+			return // is set separately
+		}
+		if !ringing && code >= sip.StatusRinging && code < sip.StatusOK {
 			ringing = true
 			c.setStatus(CallRinging)
 		}
+		c.setExtraAttrs(nil, 0, nil, hdrs)
 	})
 	if err != nil {
 		// TODO: should we retry? maybe new offer will work
@@ -526,15 +544,7 @@ func (c *outboundCall) sipSignal(ctx context.Context) error {
 	}
 	joinDur()
 
-	extra := HeadersToAttrs(nil, c.sipConf.headersToAttrs, c.sipConf.includeHeaders, c.cc)
-	if c.lkRoom != nil && len(extra) != 0 {
-		room := c.lkRoom.Room()
-		if room != nil {
-			room.LocalParticipant.SetAttributes(extra)
-		} else {
-			c.log.Warnw("could not set attributes on nil room", nil, "attrs", extra)
-		}
-	}
+	c.setExtraAttrs(c.sipConf.headersToAttrs, c.sipConf.includeHeaders, c.cc, nil)
 	c.state.DeferUpdate(func(info *livekit.SIPCallInfo) {
 		info.AudioCodec = mc.Audio.Codec.Info().SDPName
 		if r := c.lkRoom.Room(); r != nil {
@@ -684,7 +694,7 @@ func (c *sipOutbound) RemoteHeaders() Headers {
 	return c.inviteOk.Headers()
 }
 
-func (c *sipOutbound) Invite(ctx context.Context, to URI, user, pass string, headers map[string]string, sdpOffer []byte, setState func(code sip.StatusCode)) ([]byte, error) {
+func (c *sipOutbound) Invite(ctx context.Context, to URI, user, pass string, headers map[string]string, sdpOffer []byte, setState sipRespFunc) ([]byte, error) {
 	ctx, span := tracer.Start(ctx, "sipOutbound.Invite")
 	defer span.End()
 	c.mu.Lock()
@@ -814,7 +824,7 @@ func (c *sipOutbound) AckInviteOK(ctx context.Context) error {
 	return c.c.sipCli.WriteRequest(sip.NewAckRequest(c.invite, c.inviteOk, nil))
 }
 
-func (c *sipOutbound) attemptInvite(ctx context.Context, callID sip.CallIDHeader, dest string, to *sip.ToHeader, offer []byte, authHeaderName, authHeader string, headers Headers, setState func(code sip.StatusCode)) (*sip.Request, *sip.Response, error) {
+func (c *sipOutbound) attemptInvite(ctx context.Context, callID sip.CallIDHeader, dest string, to *sip.ToHeader, offer []byte, authHeaderName, authHeader string, headers Headers, setState sipRespFunc) (*sip.Request, *sip.Response, error) {
 	ctx, span := tracer.Start(ctx, "sipOutbound.attemptInvite")
 	defer span.End()
 	req := sip.NewRequest(sip.INVITE, to.Address)

pkg/sip/types.go

@@ -241,11 +241,13 @@ func LoggerWithHeaders(log logger.Logger, c Signaling) logger.Logger {
 	return log
 }
 
-func HeadersToAttrs(attrs, hdrToAttr map[string]string, opts livekit.SIPHeaderOptions, c Signaling) map[string]string {
+func HeadersToAttrs(attrs, hdrToAttr map[string]string, opts livekit.SIPHeaderOptions, c Signaling, headers Headers) map[string]string {
 	if attrs == nil {
 		attrs = make(map[string]string)
 	}
-	headers := c.RemoteHeaders()
+	if c != nil {
+		headers = c.RemoteHeaders()
+	}
 	// Map all headers, if requested
 	if opts != livekit.SIPHeaderOptions_SIP_NO_HEADERS {
 		for _, h := range headers {
@@ -277,12 +279,14 @@ func HeadersToAttrs(attrs, hdrToAttr map[string]string, opts livekit.SIPHeaderOp
 			attrs[name] = h.Value()
 		}
 	}
-	// Other metadata
-	if tag := c.Tag(); tag != "" {
-		attrs[AttrSIPCallTag] = string(tag)
-	}
-	if cid := c.CallID(); cid != "" {
-		attrs[AttrSIPCallIDFull] = cid
+	if c != nil {
+		// Other metadata
+		if tag := c.Tag(); tag != "" {
+			attrs[AttrSIPCallTag] = string(tag)
+		}
+		if cid := c.CallID(); cid != "" {
+			attrs[AttrSIPCallIDFull] = cid
+		}
 	}
 	return attrs
 }