Mirrors/cloud-game
1
0
Fork 0
mirror of https://github.com/giongto35/cloud-game.git synced 2026-08-04 07:43:58 +00:00
Code Issues Projects Releases Packages Wiki Activity

Use an interceptor for changing video timestamps (#255)

Browse source
This commit is contained in:
sergystepanov 2021-01-06 14:09:11 +03:00 • committed by Sergey Stepanov
parent 8bed364bb3
commit 53d15728fb
No known key found for this signature in database
GPG key ID: A56B4929BAA8556B
5 changed files with 76 additions and 118 deletions
Download patch file Download diff file Expand all files Collapse all files

1
go.mod vendored
View file

@ -19,6 +19,7 @@ require (
github.com/kkyr/fig v0.2.0
github.com/mitchellh/mapstructure v1.4.0 // indirect
github.com/pelletier/go-toml v1.8.1 // indirect
github.com/pion/interceptor v0.0.9
github.com/pion/rtp v1.6.2
github.com/pion/webrtc/v3 v3.0.3
github.com/prometheus/client_golang v1.9.0

24
pkg/webrtc/connection.go Normal file
View file

@ -0,0 +1,24 @@
package webrtc
import (
"github.com/pion/interceptor"
. "github.com/pion/webrtc/v3"
)
func NewInterceptedPeerConnection(conf Configuration, interceptors []interceptor.Interceptor) (*PeerConnection, error) {
m := &MediaEngine{}
if err := m.RegisterDefaultCodecs(); err != nil {
return nil, err
}
i := &interceptor.Registry{}
if err := RegisterDefaultInterceptors(m, i); err != nil {
return nil, err
}
for _, itc := range interceptors {
i.Add(itc)
}
api := NewAPI(WithMediaEngine(m), WithInterceptorRegistry(i))
return api.NewPeerConnection(conf)
}

36
pkg/webrtc/interceptor/retime.go Normal file
View file

@ -0,0 +1,36 @@
package interceptor
import (
"strings"
"sync/atomic"
. "github.com/pion/interceptor"
"github.com/pion/rtp"
)
// ReTime interceptor replaces timestamps of all outgoing video packets.
type ReTime struct {
NoOp
timestamp uint32
}
// BindLocalStream modifies any outgoing RTP packets.
func (i *ReTime) BindLocalStream(info *StreamInfo, writer RTPWriter) RTPWriter {
// use with video packets only
if strings.HasPrefix(info.MimeType, "video/") {
return RTPWriterFunc(func(header *rtp.Header, payload []byte, attributes Attributes) (int, error) {
h := *header
h.Timestamp = i.GetTimestamp()
return writer.Write(&h, payload, attributes)
})
}
return writer
}
func (i *ReTime) SetTimestamp(ts uint32) {
atomic.StoreUint32(&i.timestamp, ts)
}
func (i *ReTime) GetTimestamp() uint32 {
return atomic.LoadUint32(&i.timestamp)
}

103
pkg/webrtc/track.go
View file

@ -1,103 +0,0 @@
package webrtc
import (
"strings"
"github.com/pion/rtp"
"github.com/pion/rtp/codecs"
"github.com/pion/webrtc/v3"
"github.com/pion/webrtc/v3/pkg/media"
)
// CustomTrackSample is used just for adding custom timestamps
// into outgoing packets, since packetizer is not accessible anymore.
// Use webrtc.TrackLocalStaticSample instead if you use constant rate streams.
type CustomTrackSample struct {
packetizer rtp.Packetizer
rtpTrack *webrtc.TrackLocalStaticRTP
clockRate float64
}
func NewCustomTrackSample(c webrtc.RTPCodecCapability, id, streamID string) (*CustomTrackSample, error) {
rtpTrack, err := webrtc.NewTrackLocalStaticRTP(c, id, streamID)
if err != nil {
return nil, err
}
return &CustomTrackSample{rtpTrack: rtpTrack}, nil
}
func (s *CustomTrackSample) ID() string { return s.rtpTrack.ID() }
func (s *CustomTrackSample) StreamID() string { return s.rtpTrack.StreamID() }
func (s *CustomTrackSample) Kind() webrtc.RTPCodecType { return s.rtpTrack.Kind() }
func (s *CustomTrackSample) Codec() webrtc.RTPCodecCapability { return s.rtpTrack.Codec() }
func (s *CustomTrackSample) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters, error) {
rtpOutboundMTU := 1200
codec, err := s.rtpTrack.Bind(t)
if err != nil {
return codec, err
}
if s.packetizer != nil {
return codec, nil
}
payloader, err := payloaderForCodec(codec.RTPCodecCapability)
if err != nil {
return codec, err
}
s.packetizer = rtp.NewPacketizer(
rtpOutboundMTU,
0, // Value is handled when writing
0, // Value is handled when writing
payloader,
rtp.NewRandomSequencer(),
codec.ClockRate,
)
s.clockRate = float64(codec.RTPCodecCapability.ClockRate)
return codec, nil
}
func (s *CustomTrackSample) Unbind(t webrtc.TrackLocalContext) error {
return s.rtpTrack.Unbind(t)
}
func (s *CustomTrackSample) WriteSampleWithTimestamp(sample media.Sample, timestamp uint32) (err error) {
p, clockRate := s.packetizer, s.clockRate
if p == nil {
return nil
}
samples := sample.Duration.Seconds() * clockRate
packets := p.(rtp.Packetizer).Packetize(sample.Data, uint32(samples))
for _, p := range packets {
p.Timestamp = timestamp
err = s.rtpTrack.WriteRTP(p)
}
return
}
func payloaderForCodec(codec webrtc.RTPCodecCapability) (rtp.Payloader, error) {
switch strings.ToLower(codec.MimeType) {
case strings.ToLower(webrtc.MimeTypeH264):
return &codecs.H264Payloader{}, nil
case strings.ToLower(webrtc.MimeTypeOpus):
return &codecs.OpusPayloader{}, nil
case strings.ToLower(webrtc.MimeTypeVP8):
return &codecs.VP8Payloader{}, nil
case strings.ToLower(webrtc.MimeTypeVP9):
return &codecs.VP9Payloader{}, nil
case strings.ToLower(webrtc.MimeTypeG722):
return &codecs.G722Payloader{}, nil
case strings.ToLower(webrtc.MimeTypePCMU), strings.ToLower(webrtc.MimeTypePCMA):
return &codecs.G711Payloader{}, nil
default:
return nil, webrtc.ErrNoPayloaderForCodec
}
}

30
pkg/webrtc/webrtc.go
View file

@ -12,7 +12,9 @@ import (
webrtcConfig "github.com/giongto35/cloud-game/v2/pkg/config/webrtc"
"github.com/giongto35/cloud-game/v2/pkg/encoder"
"github.com/giongto35/cloud-game/v2/pkg/util"
itc "github.com/giongto35/cloud-game/v2/pkg/webrtc/interceptor"
"github.com/gofrs/uuid"
"github.com/pion/interceptor"
"github.com/pion/webrtc/v3"
"github.com/pion/webrtc/v3/pkg/media"
)
@ -29,10 +31,11 @@ type WebFrame struct {
type WebRTC struct {
ID string
connection *webrtc.PeerConnection
cfg webrtcConfig.Config
isConnected bool
isClosed bool
connection *webrtc.PeerConnection
cfg webrtcConfig.Config
tsInterceptor itc.ReTime
isConnected bool
isClosed bool
// for yuvI420 image
ImageChannel chan WebFrame
AudioChannel chan []byte
@ -110,7 +113,7 @@ func (w *WebRTC) StartClient(isMobile bool, iceCB OnIceCallback) (string, error)
}
}()
var err error
var videoTrack *CustomTrackSample
var videoTrack *webrtc.TrackLocalStaticSample
// reset client
if w.isConnected {
@ -119,7 +122,8 @@ func (w *WebRTC) StartClient(isMobile bool, iceCB OnIceCallback) (string, error)
}
log.Println("=== StartClient ===")
w.connection, err = webrtc.NewPeerConnection(webrtcconfig)
w.tsInterceptor = itc.ReTime{}
w.connection, err = NewInterceptedPeerConnection(webrtcconfig, []interceptor.Interceptor{&w.tsInterceptor})
if err != nil {
return "", err
}
@ -131,13 +135,11 @@ func (w *WebRTC) StartClient(isMobile bool, iceCB OnIceCallback) (string, error)
} else {
codec = webrtc.RTPCodecCapability{MimeType: "video/vp8"}
}
videoTrack, err = NewCustomTrackSample(codec, "video", "game-video")
if err != nil {
if videoTrack, err = webrtc.NewTrackLocalStaticSample(codec, "video", "game-video"); err != nil {
return "", err
}
_, err = w.connection.AddTrack(videoTrack)
if err != nil {
if _, err = w.connection.AddTrack(videoTrack); err != nil {
return "", err
}
log.Println("Add video track")
@ -313,7 +315,7 @@ func (w *WebRTC) IsConnected() bool {
return w.isConnected
}
func (w *WebRTC) startStreaming(vp8Track *CustomTrackSample, opusTrack *webrtc.TrackLocalStaticSample) {
func (w *WebRTC) startStreaming(vp8Track *webrtc.TrackLocalStaticSample, opusTrack *webrtc.TrackLocalStaticSample) {
log.Println("Start streaming")
// receive frame buffer
go func() {
@ -325,8 +327,8 @@ func (w *WebRTC) startStreaming(vp8Track *CustomTrackSample, opusTrack *webrtc.T
}()
for data := range w.ImageChannel {
err := vp8Track.WriteSampleWithTimestamp(media.Sample{Data: data.Data}, data.Timestamp)
if err != nil {
w.tsInterceptor.SetTimestamp(data.Timestamp)
if err := vp8Track.WriteSample(media.Sample{Data: data.Data}); err != nil {
log.Println("Warn: Err write sample: ", err)
break
}
@ -342,9 +344,7 @@ func (w *WebRTC) startStreaming(vp8Track *CustomTrackSample, opusTrack *webrtc.T
}
}()
//opusSamples := uint32(w.cfg.Encoder.Audio.GetFrameDuration() / w.cfg.Encoder.Audio.Channels)
audioDuration := time.Duration(w.cfg.Encoder.Audio.Frame) * time.Millisecond
for data := range w.AudioChannel {
if !w.isConnected {
return