用实例学习pion - play-from-disk
Howard Yin 2021-09-02 13:34:02  WebRTC编程框架pion实操
// 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 + "`")
}
 1
2
3
4
5
6
7
8
9
10
2
3
4
5
6
7
8
9
10
先看看本地视频和音频文件在不在
// 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)
    }
}()
 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
webrtc.NewPeerConnection创建默认配置的WebRTC标准PeerConnection。
iceConnectedCtx, iceConnectedCtxCancel := context.WithCancel(context.Background())
 1
创建context,后面用到。
# 绑定视频流
if haveVideoFile {
 1
如果视频存在就开始绑定视频流
    // Create a video track
    videoTrack, videoTrackErr := webrtc.NewTrackLocalStaticSample(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8}, "video", "pion")
    if videoTrackErr != nil {
        panic(videoTrackErr)
    }
 1
2
3
4
5
2
3
4
5
首先是创建track,pion里的track分为TrackLocal和TrackRemote两种interface。其中TrackLocal是本地发给别人的track,TrackRemote是别人发给本地的track,更多分析可见《pion中的TrackLocal》和《pion中的TrackRemote》。这里是要发送给远端的,当然是创建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
            }
        }
    }()
 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
用AddTrack添加刚才创建好的track,和《用实例学习pion - rtp-to-webrtc》里一样
# 视频流的发送过程
    go func() {
 1
这里一个协程独立运行视频流的发送过程
        // Open a IVF file and start reading using our IVFReader
        file, ivfErr := os.Open(videoFileName)
        if ivfErr != nil {
            panic(ivfErr)
        }
 1
2
3
4
5
2
3
4
5
先打开文件
        ivf, header, ivfErr := ivfreader.NewWith(file)
        if ivfErr != nil {
            panic(ivfErr)
        }
 1
2
3
4
2
3
4
这就是创建一个读取器
        // Wait for connection established
        <-iceConnectedCtx.Done()
 1
2
2
等待连接建立后正式开始操作
        // Send our video file frame at a time. Pace our sending so we send it at the same speed it should be played back as.
        // This isn't required since the video is timestamped, but we will such much higher loss if we send all at once.
        //
        // 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(time.Millisecond * time.Duration((float32(header.TimebaseNumerator)/float32(header.TimebaseDenominator))*1000))
        for ; true; <-ticker.C {
            frame, _, ivfErr := ivf.ParseNextFrame()
            if ivfErr == io.EOF {
                fmt.Printf("All video frames parsed and sent")
                os.Exit(0)
            }
            if ivfErr != nil {
                panic(ivfErr)
            }
            if ivfErr = videoTrack.WriteSample(media.Sample{Data: frame, Duration: time.Second}); ivfErr != nil {
                panic(ivfErr)
            }
        }
 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
可以看到,操作不过就是将ivf格式的视频数据读出来用WriteSample写进track里。
    }()
}
 1
2
2
协程结束。
# 绑定音频流、音频流的发送过程
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)
            }
        }
    }()
}
 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
可以看到,操作都和视频流一毛一样。
# 启动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 {}
 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
和《用实例学习pion - rtp-forwarder》里一样的等待链接的方式,不必多讲。