add video support

pkg/media/h264/h264.go

@@ -0,0 +1,48 @@
+// Copyright 2023 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// 	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package h264
+
+import (
+	m "github.com/livekit/sip/pkg/media"
+	"github.com/pion/webrtc/v3/pkg/media"
+	"time"
+)
+
+type Sample []byte
+
+type Encoder struct {
+	w   m.Writer[Sample]
+	buf Sample
+}
+
+func (e *Encoder) WriteSample(in Sample) error {
+	return e.w.WriteSample(in)
+}
+
+func Encode(w m.Writer[Sample]) m.Writer[Sample] {
+	return &Encoder{w: w}
+}
+
+type SampleWriter interface {
+	WriteSample(sample media.Sample) error
+}
+
+func BuildSampleWriter[T ~[]byte](w SampleWriter, sampleDur time.Duration) m.Writer[T] {
+	return m.WriterFunc[T](func(in T) error {
+		data := make([]byte, len(in))
+		copy(data, in)
+		return w.WriteSample(media.Sample{Data: data})
+	})
+}

pkg/sip/inbound.go

@@ -17,6 +17,7 @@
 import (
 	"context"
 	"fmt"
+	"github.com/livekit/sip/pkg/media/h264"
 	"sync/atomic"
 	"time"
 
@@ -54,7 +55,6 @@
 
 	h := req.GetHeader("Proxy-Authorization")
 	if h == nil {
-		logger.Infow(fmt.Sprintf("Requesting inbound auth for %s", from), "from", from)
 		inviteState.challenge = digest.Challenge{
 			Realm:     UserAgent,
 			Nonce:     fmt.Sprintf("%d", time.Now().UnixMicro()),
@@ -117,14 +117,12 @@
 	src := req.Source()
 
 	s.mon.InviteReq(false, from.Address.String(), to.Address.String())
-	logger.Infow(fmt.Sprintf("INVITE from %q to %q", from.Address.String(), to.Address.String()),
-		"tag", tag, "from", from, "to", to)
+	logger.Infow("INVITE", "tag", tag, "from", from, "to", to)
 
 	username, password, err := s.authHandler(from.Address.User, to.Address.User, to.Address.Host, src)
 	if err != nil {
 		s.mon.InviteError(false, from.Address.String(), to.Address.String(), "no-rule")
-		logger.Warnw(fmt.Sprintf("Rejecting inbound call to %q, doesn't match any Trunks", to.Address.String()), err,
-			"tag", tag, "src", src, "from", from, "to", to, "to-host", to.Address.Host)
+		logger.Warnw("Rejecting inbound call, doesn't match any Trunks", err, "tag", tag, "src", src, "from", from, "to", to, "to-host", to.Address.Host)
 		sipErrorResponse(tx, req)
 		return
 	}
@@ -166,8 +164,10 @@
 	from         *sip.FromHeader
 	to           *sip.ToHeader
 	src          string
-	rtpConn      *MediaConn
+	audioRtpConn *MediaConn
+	videoRtpConn *MediaConn
 	audioHandler atomic.Pointer[rtp.Handler]
+	videoHandler atomic.Pointer[rtp.Handler]
 	dtmf         chan byte // buffered; DTMF digits as characters
 	lkRoom       *Room     // LiveKit room; only active after correct pin is entered
 	done         atomic.Bool
@@ -256,23 +256,34 @@
 }
 
 func (c *inboundCall) runMediaConn(offerData []byte, conf *config.Config) (answerData []byte, _ error) {
-	conn := NewMediaConn()
-	conn.OnRTP(c)
-	if err := conn.Start(conf.RTPPort.Start, conf.RTPPort.End, "0.0.0.0"); err != nil {
+	audioConn := NewMediaConn()
+	audioConn.OnRTP(c)
+	if err := audioConn.Start(conf.RTPPort.Start, conf.RTPPort.End, "0.0.0.0"); err != nil {
 		return nil, err
 	}
-	c.rtpConn = conn
+	c.audioRtpConn = audioConn
+
+	videoConn := NewMediaConn()
+	videoConn.OnRTP(c)
+	if err := videoConn.Start(conf.RTPPort.Start, conf.RTPPort.End, "0.0.0.0"); err != nil {
+		return nil, err
+	}
+	c.videoRtpConn = videoConn
+
 	offer := sdp.SessionDescription{}
 	if err := offer.Unmarshal(offerData); err != nil {
 		return nil, err
 	}
 
 	// Encoding pipeline (LK -> SIP)
 	// Need to be created earlier to send the pin prompts.
-	s := rtp.NewMediaStreamOut[ulaw.Sample](conn, rtpPacketDur)
-	c.lkRoom.SetOutput(ulaw.Encode(s))
+	aus := rtp.NewMediaStreamOut[ulaw.Sample](audioConn, rtpPacketDur)
+	c.lkRoom.SetAudioOutput(ulaw.Encode(aus))
+
+	vis := rtp.NewMediaStreamOut[h264.Sample](videoConn, rtpPacketDur)
+	c.lkRoom.SetVideoOutput(h264.Encode(vis))
 
-	return sdpGenerateAnswer(offer, c.s.signalingIp, conn.LocalAddr().Port)
+	return sdpGenerateAnswer(offer, c.s.signalingIp, audioConn.LocalAddr().Port, videoConn.LocalAddr().Port)
 }
 
 func (c *inboundCall) pinPrompt(ctx context.Context) {
@@ -343,9 +354,13 @@
 		p.Close()
 		c.lkRoom = nil
 	}
-	if c.rtpConn != nil {
-		c.rtpConn.Close()
-		c.rtpConn = nil
+	if c.audioRtpConn != nil {
+		c.audioRtpConn.Close()
+		c.audioRtpConn = nil
+	}
+	if c.videoRtpConn != nil {
+		c.videoRtpConn.Close()
+		c.videoRtpConn = nil
 	}
 	close(c.dtmf)
 }
@@ -356,9 +371,17 @@
 		return nil
 	}
 	// TODO: Audio data appears to be coming with PayloadType=0, so maybe enforce it?
-	if h := c.audioHandler.Load(); h != nil {
-		return (*h).HandleRTP(p)
+	//logger.Infow("------------ pt == ", p.PayloadType)
+	if p.PayloadType < 96 {
+		if h := c.audioHandler.Load(); h != nil {
+			return (*h).HandleRTP(p)
+		}
+	} else {
+		if h := c.videoHandler.Load(); h != nil {
+			return (*h).HandleRTP(p)
+		}
 	}
+
 	return nil
 }
 
@@ -372,17 +395,21 @@
 	if err != nil {
 		return err
 	}
-	local, err := c.lkRoom.NewParticipant()
+	audioTrack, videoTrack, err := c.lkRoom.NewParticipant()
 	if err != nil {
 		_ = c.lkRoom.Close()
 		return err
 	}
 
-	// Decoding pipeline (SIP -> LK)
-	law := ulaw.Decode(local)
+	// Decoding pipeline (SIP -> LK) audio
+	law := ulaw.Decode(audioTrack)
+	//构造NewMediaStreamIn  通过readLoop方法传入rtp byte数据，通过law decode为int16数据传给local  (pw)
 	var h rtp.Handler = rtp.NewMediaStreamIn(law)
 	c.audioHandler.Store(&h)
 
+	var vh rtp.Handler = rtp.NewMediaStreamIn(videoTrack)
+	c.videoHandler.Store(&vh)
+
 	return nil
 }
 

pkg/sip/outbound.go

@@ -17,6 +17,7 @@
 import (
 	"context"
 	"fmt"
+	"github.com/livekit/sip/pkg/media/h264"
 	"sync"
 
 	"github.com/emiago/sipgo/sip"
@@ -40,13 +41,15 @@
 type outboundCall struct {
 	c             *Client
 	participantID string
-	rtpConn       *MediaConn
+	audioRtpConn  *MediaConn
+	videoRtpConn  *MediaConn
 
 	mu            sync.RWMutex
 	mediaRunning  bool
 	lkCur         lkRoomConfig
 	lkRoom        *Room
-	lkRoomIn      media.Writer[media.PCM16Sample]
+	lkRoomAudioIn media.Writer[media.PCM16Sample]
+	lkRoomVideoIn media.Writer[h264.Sample]
 	sipCur        sipOutboundConfig
 	sipInviteReq  *sip.Request
 	sipInviteResp *sip.Response
@@ -79,7 +82,8 @@
 	call := &outboundCall{
 		c:             c,
 		participantID: participantId,
-		rtpConn:       NewMediaConn(),
+		audioRtpConn:  NewMediaConn(),
+		videoRtpConn:  NewMediaConn(),
 	}
 	return call
 }
@@ -92,19 +96,23 @@
 }
 
 func (c *outboundCall) close() {
-	c.rtpConn.OnRTP(nil)
-	c.lkRoom.SetOutput(nil)
+	c.audioRtpConn.OnRTP(nil)
+	c.videoRtpConn.OnRTP(nil)
+	c.lkRoom.SetAudioOutput(nil)
+	c.lkRoom.SetVideoOutput(nil)
 
 	if c.mediaRunning {
-		_ = c.rtpConn.Close()
+		_ = c.audioRtpConn.Close()
+		_ = c.videoRtpConn.Close()
 	}
 	c.mediaRunning = false
 
 	if c.lkRoom != nil {
 		_ = c.lkRoom.Close()
 	}
 	c.lkRoom = nil
-	c.lkRoomIn = nil
+	c.lkRoomAudioIn = nil
+	c.lkRoomVideoIn = nil
 	c.lkCur = lkRoomConfig{}
 
 	c.stopSIP()
@@ -155,7 +163,10 @@
 	if c.mediaRunning {
 		return nil
 	}
-	if err := c.rtpConn.Start(conf.RTPPort.Start, conf.RTPPort.End, "0.0.0.0"); err != nil {
+	if err := c.audioRtpConn.Start(conf.RTPPort.Start, conf.RTPPort.End, "0.0.0.0"); err != nil {
+		return err
+	}
+	if err := c.videoRtpConn.Start(conf.RTPPort.Start, conf.RTPPort.End, "0.0.0.0"); err != nil {
 		return err
 	}
 	c.mediaRunning = true
@@ -169,19 +180,21 @@
 	if c.lkRoom != nil {
 		_ = c.lkRoom.Close()
 		c.lkRoom = nil
-		c.lkRoomIn = nil
+		c.lkRoomAudioIn = nil
+		c.lkRoomVideoIn = nil
 	}
 	r, err := ConnectToRoom(c.c.conf, lkNew.roomName, lkNew.identity)
 	if err != nil {
 		return err
 	}
-	local, err := r.NewParticipant()
+	audioTrack, videoTrack, err := r.NewParticipant()
 	if err != nil {
 		_ = r.Close()
 		return err
 	}
 	c.lkRoom = r
-	c.lkRoomIn = local
+	c.lkRoomAudioIn = audioTrack
+	c.lkRoomVideoIn = videoTrack
 	c.lkCur = lkNew
 	return nil
 }
@@ -201,17 +214,25 @@
 
 func (c *outboundCall) relinkMedia() {
 	if c.lkRoom == nil || !c.mediaRunning {
-		c.lkRoom.SetOutput(nil)
-		c.rtpConn.OnRTP(nil)
+		c.lkRoom.SetAudioOutput(nil)
+		c.lkRoom.SetVideoOutput(nil)
+		c.audioRtpConn.OnRTP(nil)
+		c.videoRtpConn.OnRTP(nil)
 		return
 	}
 	// Encoding pipeline (LK -> SIP)
-	s := rtp.NewMediaStreamOut[ulaw.Sample](c.rtpConn, rtpPacketDur)
-	c.lkRoom.SetOutput(ulaw.Encode(s))
+	aus := rtp.NewMediaStreamOut[ulaw.Sample](c.audioRtpConn, rtpPacketDur)
+	c.lkRoom.SetAudioOutput(ulaw.Encode(aus))
+
+	vis := rtp.NewMediaStreamOut[h264.Sample](c.videoRtpConn, rtpPacketDur)
+	c.lkRoom.SetVideoOutput(h264.Encode(vis))
 
 	// Decoding pipeline (SIP -> LK)
-	law := ulaw.Decode(c.lkRoomIn)
-	c.rtpConn.OnRTP(rtp.NewMediaStreamIn(law))
+	law := ulaw.Decode(c.lkRoomAudioIn)
+	c.audioRtpConn.OnRTP(rtp.NewMediaStreamIn(law))
+
+	var vh rtp.Handler = rtp.NewMediaStreamIn(c.lkRoomVideoIn)
+	c.videoRtpConn.OnRTP(vh)
 }
 
 func (c *outboundCall) SendDTMF(ctx context.Context, digits string) error {
@@ -250,7 +271,7 @@
 }
 
 func (c *outboundCall) sipSignal(conf sipOutboundConfig) error {
-	offer, err := sdpGenerateOffer(c.c.signalingIp, c.rtpConn.LocalAddr().Port)
+	offer, err := sdpGenerateOffer(c.c.signalingIp, c.audioRtpConn.LocalAddr().Port, c.videoRtpConn.LocalAddr().Port)
 	if err != nil {
 		return err
 	}