// Assert that we have an audio or video file
_, err := os.Stat(videoFileName)
haveVideoFile := !os.IsNotExist(err)
_, err = os.Stat(audioFileName)
haveAudioFile := !os.IsNotExist(err)
if !haveAudioFile && !haveVideoFile {
panic("Could not find `" + audioFileName + "` or `" + videoFileName + "`")
}
先看看本地视频和音频文件在不在
// Create a new RTCPeerConnection
peerConnection, err := webrtc.NewPeerConnection(webrtc.Configuration{
ICEServers: []webrtc.ICEServer{
{
URLs: []string{"stun:stun.l.google.com:19302"},
},
},
})
if err != nil {
panic(err)
}
defer func() {
if cErr := peerConnection.Close(); cErr != nil {
fmt.Printf("cannot close peerConnection: %v\n", cErr)
}
}()
webrtc.NewPeerConnection创建默认配置的WebRTC标准PeerConnection。
iceConnectedCtx, iceConnectedCtxCancel := context.WithCancel(context.Background())
创建context，后面用到。
绑定视频流
if haveVideoFile {
如果视频存在就开始绑定视频流
// Create a video track
videoTrack, videoTrackErr := webrtc.NewTrackLocalStaticSample(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8}, "video", "pion")
if videoTrackErr != nil {
panic(videoTrackErr)
}
首先是创建track，pion里的track分为TrackLocal和TrackRemote两种interface。其中TrackLocal是本地发给别人的track，TrackRemote是别人发给本地的track。这里是要发送给远端的，当然是创建TrackLocal。
这里用的webrtc.NewTrackLocalStaticSample是pion里提供的创建TrackLocal的范例。
rtpSender, videoTrackErr := peerConnection.AddTrack(videoTrack)
if videoTrackErr != nil {
panic(videoTrackErr)
}
// Read incoming RTCP packets
// Before these packets are returned they are processed by interceptors. For things
// like NACK this needs to be called.
go func() {
rtcpBuf := make([]byte, 1500)
for {
if _, _, rtcpErr := rtpSender.Read(rtcpBuf); rtcpErr != nil {
return
}
}
}()
用AddTrack添加刚才创建好的track
视频流的发送过程
go func() {
这里一个协程独立运行视频流的发送过程
// Open a IVF file and start reading using our IVFReader
file, ivfErr := os.Open(videoFileName)
if ivfErr != nil {
panic(ivfErr)
}
先打开文件
ivf, header, ivfErr := ivfreader.NewWith(file)
if ivfErr != nil {
panic(ivfErr)
}
这就是创建一个读取器
// Wait for connection established
<-iceConnectedCtx.Done()
等待连接建立后正式开始操作
可以看到，操作不过就是将ivf格式的视频数据读出来用WriteSample写进track里。
}()
}
协程结束。
绑定音频流、音频流的发送过程
if haveAudioFile {
// Create a audio track
audioTrack, audioTrackErr := webrtc.NewTrackLocalStaticSample(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus}, "audio", "pion")
if audioTrackErr != nil {
panic(audioTrackErr)
}
rtpSender, audioTrackErr := peerConnection.AddTrack(audioTrack)
if audioTrackErr != nil {
panic(audioTrackErr)
}
// Read incoming RTCP packets
// Before these packets are returned they are processed by interceptors. For things
// like NACK this needs to be called.
go func() {
rtcpBuf := make([]byte, 1500)
for {
if _, _, rtcpErr := rtpSender.Read(rtcpBuf); rtcpErr != nil {
return
}
}
}()
go func() {
// Open a IVF file and start reading using our IVFReader
file, oggErr := os.Open(audioFileName)
if oggErr != nil {
panic(oggErr)
}
// Open on oggfile in non-checksum mode.
ogg, _, oggErr := oggreader.NewWith(file)
if oggErr != nil {
panic(oggErr)
}
// Wait for connection established
<-iceConnectedCtx.Done()
// Keep track of last granule, the difference is the amount of samples in the buffer
var lastGranule uint64
// It is important to use a time.Ticker instead of time.Sleep because
// * avoids accumulating skew, just calling time.Sleep didn't compensate for the time spent parsing the data
// * works around latency issues with Sleep (see https://github.com/golang/go/issues/44343)
ticker := time.NewTicker(oggPageDuration)
for ; true; <-ticker.C {
pageData, pageHeader, oggErr := ogg.ParseNextPage()
if oggErr == io.EOF {
fmt.Printf("All audio pages parsed and sent")
os.Exit(0)
}
if oggErr != nil {
panic(oggErr)
}
// The amount of samples is the difference between the last and current timestamp
sampleCount := float64(pageHeader.GranulePosition - lastGranule)
lastGranule = pageHeader.GranulePosition
sampleDuration := time.Duration((sampleCount/48000)*1000) * time.Millisecond
if oggErr = audioTrack.WriteSample(media.Sample{Data: pageData, Duration: sampleDuration}); oggErr != nil {
panic(oggErr)
}
}
}()
}
可以看到，操作都和视频流一毛一样。
启动PeerConnection
// Set the handler for ICE connection state
// This will notify you when the peer has connected/disconnected
peerConnection.OnICEConnectionStateChange(func(connectionState webrtc.ICEConnectionState) {
fmt.Printf("Connection State has changed %s \n", connectionState.String())
if connectionState == webrtc.ICEConnectionStateConnected {
iceConnectedCtxCancel()
}
})
// Set the handler for Peer connection state
// This will notify you when the peer has connected/disconnected
peerConnection.OnConnectionStateChange(func(s webrtc.PeerConnectionState) {
fmt.Printf("Peer Connection State has changed: %s\n", s.String())
if s == webrtc.PeerConnectionStateFailed {
// Wait until PeerConnection has had no network activity for 30 seconds or another failure. It may be reconnected using an ICE Restart.
// Use webrtc.PeerConnectionStateDisconnected if you are interested in detecting faster timeout.
// Note that the PeerConnection may come back from PeerConnectionStateDisconnected.
fmt.Println("Peer Connection has gone to failed exiting")
os.Exit(0)
}
})
// Wait for the offer to be pasted
offer := webrtc.SessionDescription{}
signal.Decode(signal.MustReadStdin(), &offer)
// Set the remote SessionDescription
if err = peerConnection.SetRemoteDescription(offer); err != nil {
panic(err)
}
// Create answer
answer, err := peerConnection.CreateAnswer(nil)
if err != nil {
panic(err)
}
// Create channel that is blocked until ICE Gathering is complete
gatherComplete := webrtc.GatheringCompletePromise(peerConnection)
// Sets the LocalDescription, and starts our UDP listeners
if err = peerConnection.SetLocalDescription(answer); err != nil {
panic(err)
}
// Block until ICE Gathering is complete, disabling trickle ICE
// we do this because we only can exchange one signaling message
// in a production application you should exchange ICE Candidates via OnICECandidate
<-gatherComplete
// Output the answer in base64 so we can paste it in browser
fmt.Println(signal.Encode(*peerConnection.LocalDescription()))
// Block forever
select {}