mirror of https://github.com/ossrs/srs.git
SquashSRS4: Regine DTLS and add regression tests. 4.0.84
winlin
4 years ago
parent
dc93836489
commit
e74810230a
@ -1,17 +0,0 @@
|
||||
#!/bin/bash
|
||||
|
||||
# check exists.
|
||||
if [[ -f /usr/local/bin/ccache ]]; then
|
||||
echo "ccache is ok";
|
||||
exit 0;
|
||||
fi
|
||||
|
||||
# check sudoer.
|
||||
sudo echo "ok" > /dev/null 2>&1;
|
||||
ret=$?; if [[ 0 -ne ${ret} ]]; then echo "you must be sudoer"; exit 1; fi
|
||||
|
||||
unzip ccache-3.1.9.zip && cd ccache-3.1.9 && ./configure && make
|
||||
ret=$?; if [[ $ret -ne 0 ]]; then echo "build ccache failed."; exit $ret; fi
|
||||
|
||||
sudo cp ccache /usr/local/bin && sudo ln -s ccache /usr/local/bin/gcc && sudo ln -s ccache /usr/local/bin/g++ && sudo ln -s ccache /usr/local/bin/cc && sudo ln -s ccache /usr/local/bin/c++
|
||||
ret=$?; if [[ $ret -ne 0 ]]; then echo "install ccache failed."; exit $ret; fi
|
||||
Binary file not shown.
@ -0,0 +1,20 @@
|
||||
The MIT License (MIT)
|
||||
|
||||
Copyright (c) 2021 srs-bench(ossrs)
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy of
|
||||
this software and associated documentation files (the "Software"), to deal in
|
||||
the Software without restriction, including without limitation the rights to
|
||||
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
|
||||
the Software, and to permit persons to whom the Software is furnished to do so,
|
||||
subject to the following conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall be included in all
|
||||
copies or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
|
||||
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
|
||||
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
|
||||
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
|
||||
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
Binary file not shown.
@ -1,5 +0,0 @@
|
||||
package rtc
|
||||
|
||||
const (
|
||||
rtpOutboundMTU = 1200
|
||||
)
|
||||
@ -1,27 +0,0 @@
|
||||
package rtc
|
||||
|
||||
import (
|
||||
"github.com/pion/rtp"
|
||||
"github.com/pion/rtp/codecs"
|
||||
"github.com/pion/webrtc/v3"
|
||||
"strings"
|
||||
)
|
||||
|
||||
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
|
||||
}
|
||||
}
|
||||
@ -1,27 +0,0 @@
|
||||
package rtc
|
||||
|
||||
import (
|
||||
"github.com/pion/webrtc/v3"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// Do a fuzzy find for a codec in the list of codecs
|
||||
// Used for lookup up a codec in an existing list to find a match
|
||||
func codecParametersFuzzySearch(needle webrtc.RTPCodecParameters, haystack []webrtc.RTPCodecParameters) (webrtc.RTPCodecParameters, error) {
|
||||
// First attempt to match on MimeType + SDPFmtpLine
|
||||
for _, c := range haystack {
|
||||
if strings.EqualFold(c.RTPCodecCapability.MimeType, needle.RTPCodecCapability.MimeType) &&
|
||||
c.RTPCodecCapability.SDPFmtpLine == needle.RTPCodecCapability.SDPFmtpLine {
|
||||
return c, nil
|
||||
}
|
||||
}
|
||||
|
||||
// Fallback to just MimeType
|
||||
for _, c := range haystack {
|
||||
if strings.EqualFold(c.RTPCodecCapability.MimeType, needle.RTPCodecCapability.MimeType) {
|
||||
return c, nil
|
||||
}
|
||||
}
|
||||
|
||||
return webrtc.RTPCodecParameters{}, webrtc.ErrCodecNotFound
|
||||
}
|
||||
@ -1,246 +0,0 @@
|
||||
package rtc
|
||||
|
||||
import (
|
||||
"github.com/pion/rtp"
|
||||
"github.com/pion/webrtc/v3"
|
||||
"github.com/pion/webrtc/v3/pkg/media"
|
||||
"strings"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// trackBinding is a single bind for a Track
|
||||
// Bind can be called multiple times, this stores the
|
||||
// result for a single bind call so that it can be used when writing
|
||||
type trackBinding struct {
|
||||
id string
|
||||
ssrc webrtc.SSRC
|
||||
payloadType webrtc.PayloadType
|
||||
writeStream webrtc.TrackLocalWriter
|
||||
}
|
||||
|
||||
// TrackLocalStaticRTP is a TrackLocal that has a pre-set codec and accepts RTP Packets.
|
||||
// If you wish to send a media.Sample use TrackLocalStaticSample
|
||||
type TrackLocalStaticRTP struct {
|
||||
mu sync.RWMutex
|
||||
bindings []trackBinding
|
||||
codec webrtc.RTPCodecCapability
|
||||
id, streamID string
|
||||
}
|
||||
|
||||
// NewTrackLocalStaticRTP returns a TrackLocalStaticRTP.
|
||||
func NewTrackLocalStaticRTP(c webrtc.RTPCodecCapability, id, streamID string) (*TrackLocalStaticRTP, error) {
|
||||
return &TrackLocalStaticRTP{
|
||||
codec: c,
|
||||
bindings: []trackBinding{},
|
||||
id: id,
|
||||
streamID: streamID,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Bind is called by the PeerConnection after negotiation is complete
|
||||
// This asserts that the code requested is supported by the remote peer.
|
||||
// If so it setups all the state (SSRC and PayloadType) to have a call
|
||||
func (s *TrackLocalStaticRTP) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
parameters := webrtc.RTPCodecParameters{RTPCodecCapability: s.codec}
|
||||
if codec, err := codecParametersFuzzySearch(parameters, t.CodecParameters()); err == nil {
|
||||
s.bindings = append(s.bindings, trackBinding{
|
||||
ssrc: t.SSRC(),
|
||||
payloadType: codec.PayloadType,
|
||||
writeStream: t.WriteStream(),
|
||||
id: t.ID(),
|
||||
})
|
||||
return codec, nil
|
||||
}
|
||||
|
||||
return webrtc.RTPCodecParameters{}, webrtc.ErrUnsupportedCodec
|
||||
}
|
||||
|
||||
// Unbind implements the teardown logic when the track is no longer needed. This happens
|
||||
// because a track has been stopped.
|
||||
func (s *TrackLocalStaticRTP) Unbind(t webrtc.TrackLocalContext) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
for i := range s.bindings {
|
||||
if s.bindings[i].id == t.ID() {
|
||||
s.bindings[i] = s.bindings[len(s.bindings)-1]
|
||||
s.bindings = s.bindings[:len(s.bindings)-1]
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
return webrtc.ErrUnbindFailed
|
||||
}
|
||||
|
||||
// ID is the unique identifier for this Track. This should be unique for the
|
||||
// stream, but doesn't have to globally unique. A common example would be 'audio' or 'video'
|
||||
// and StreamID would be 'desktop' or 'webcam'
|
||||
func (s *TrackLocalStaticRTP) ID() string { return s.id }
|
||||
|
||||
// StreamID is the group this track belongs too. This must be unique
|
||||
func (s *TrackLocalStaticRTP) StreamID() string { return s.streamID }
|
||||
|
||||
// Kind controls if this TrackLocal is audio or video
|
||||
func (s *TrackLocalStaticRTP) Kind() webrtc.RTPCodecType {
|
||||
switch {
|
||||
case strings.HasPrefix(s.codec.MimeType, "audio/"):
|
||||
return webrtc.RTPCodecTypeAudio
|
||||
case strings.HasPrefix(s.codec.MimeType, "video/"):
|
||||
return webrtc.RTPCodecTypeVideo
|
||||
default:
|
||||
return webrtc.RTPCodecType(0)
|
||||
}
|
||||
}
|
||||
|
||||
// Codec gets the Codec of the track
|
||||
func (s *TrackLocalStaticRTP) Codec() webrtc.RTPCodecCapability {
|
||||
return s.codec
|
||||
}
|
||||
|
||||
// WriteRTP writes a RTP Packet to the TrackLocalStaticRTP
|
||||
// If one PeerConnection fails the packets will still be sent to
|
||||
// all PeerConnections. The error message will contain the ID of the failed
|
||||
// PeerConnections so you can remove them
|
||||
func (s *TrackLocalStaticRTP) WriteRTP(p *rtp.Packet) error {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
|
||||
writeErrs := []error{}
|
||||
outboundPacket := *p
|
||||
|
||||
for _, b := range s.bindings {
|
||||
outboundPacket.Header.SSRC = uint32(b.ssrc)
|
||||
outboundPacket.Header.PayloadType = uint8(b.payloadType)
|
||||
if _, err := b.writeStream.WriteRTP(&outboundPacket.Header, outboundPacket.Payload); err != nil {
|
||||
writeErrs = append(writeErrs, err)
|
||||
}
|
||||
}
|
||||
|
||||
return FlattenErrs(writeErrs)
|
||||
}
|
||||
|
||||
// Write writes a RTP Packet as a buffer to the TrackLocalStaticRTP
|
||||
// If one PeerConnection fails the packets will still be sent to
|
||||
// all PeerConnections. The error message will contain the ID of the failed
|
||||
// PeerConnections so you can remove them
|
||||
func (s *TrackLocalStaticRTP) Write(b []byte) (n int, err error) {
|
||||
packet := &rtp.Packet{}
|
||||
if err = packet.Unmarshal(b); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
return len(b), s.WriteRTP(packet)
|
||||
}
|
||||
|
||||
// TrackLocalStaticSample is a TrackLocal that has a pre-set codec and accepts Samples.
|
||||
// If you wish to send a RTP Packet use TrackLocalStaticRTP
|
||||
type TrackLocalStaticSample struct {
|
||||
packetizer rtp.Packetizer
|
||||
rtpTrack *TrackLocalStaticRTP
|
||||
clockRate float64
|
||||
|
||||
// Set the callback before write RTP packet.
|
||||
OnBeforeWritePacket func(rtp *rtp.Packet)
|
||||
}
|
||||
|
||||
// NewTrackLocalStaticSample returns a TrackLocalStaticSample
|
||||
func NewTrackLocalStaticSample(c webrtc.RTPCodecCapability, id, streamID string) (*TrackLocalStaticSample, error) {
|
||||
rtpTrack, err := NewTrackLocalStaticRTP(c, id, streamID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &TrackLocalStaticSample{
|
||||
rtpTrack: rtpTrack,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// ID is the unique identifier for this Track. This should be unique for the
|
||||
// stream, but doesn't have to globally unique. A common example would be 'audio' or 'video'
|
||||
// and StreamID would be 'desktop' or 'webcam'
|
||||
func (s *TrackLocalStaticSample) ID() string { return s.rtpTrack.ID() }
|
||||
|
||||
// StreamID is the group this track belongs too. This must be unique
|
||||
func (s *TrackLocalStaticSample) StreamID() string { return s.rtpTrack.StreamID() }
|
||||
|
||||
// Kind controls if this TrackLocal is audio or video
|
||||
func (s *TrackLocalStaticSample) Kind() webrtc.RTPCodecType { return s.rtpTrack.Kind() }
|
||||
|
||||
// Codec gets the Codec of the track
|
||||
func (s *TrackLocalStaticSample) Codec() webrtc.RTPCodecCapability {
|
||||
return s.rtpTrack.Codec()
|
||||
}
|
||||
|
||||
// Bind is called by the PeerConnection after negotiation is complete
|
||||
// This asserts that the code requested is supported by the remote peer.
|
||||
// If so it setups all the state (SSRC and PayloadType) to have a call
|
||||
func (s *TrackLocalStaticSample) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters, error) {
|
||||
codec, err := s.rtpTrack.Bind(t)
|
||||
if err != nil {
|
||||
return codec, err
|
||||
}
|
||||
|
||||
s.rtpTrack.mu.Lock()
|
||||
defer s.rtpTrack.mu.Unlock()
|
||||
|
||||
// We only need one packetizer
|
||||
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
|
||||
}
|
||||
|
||||
// Unbind implements the teardown logic when the track is no longer needed. This happens
|
||||
// because a track has been stopped.
|
||||
func (s *TrackLocalStaticSample) Unbind(t webrtc.TrackLocalContext) error {
|
||||
return s.rtpTrack.Unbind(t)
|
||||
}
|
||||
|
||||
// WriteSample writes a Sample to the TrackLocalStaticSample
|
||||
// If one PeerConnection fails the packets will still be sent to
|
||||
// all PeerConnections. The error message will contain the ID of the failed
|
||||
// PeerConnections so you can remove them
|
||||
func (s *TrackLocalStaticSample) WriteSample(sample media.Sample) error {
|
||||
s.rtpTrack.mu.RLock()
|
||||
p := s.packetizer
|
||||
clockRate := s.clockRate
|
||||
s.rtpTrack.mu.RUnlock()
|
||||
|
||||
if p == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
samples := sample.Duration.Seconds() * clockRate
|
||||
packets := p.(rtp.Packetizer).Packetize(sample.Data, uint32(samples))
|
||||
|
||||
writeErrs := []error{}
|
||||
for _, p := range packets {
|
||||
if s.OnBeforeWritePacket != nil {
|
||||
s.OnBeforeWritePacket(p)
|
||||
}
|
||||
|
||||
if err := s.rtpTrack.WriteRTP(p); err != nil {
|
||||
writeErrs = append(writeErrs, err)
|
||||
}
|
||||
}
|
||||
|
||||
return FlattenErrs(writeErrs)
|
||||
}
|
||||
@ -1,10 +0,0 @@
|
||||
package rtc
|
||||
|
||||
import "fmt"
|
||||
|
||||
func FlattenErrs(errors []error) error {
|
||||
if len(errors) == 0 {
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("%v", errors)
|
||||
}
|
||||
@ -0,0 +1,285 @@
|
||||
// The MIT License (MIT)
|
||||
//
|
||||
// Copyright (c) 2021 srs-bench(ossrs)
|
||||
//
|
||||
// Permission is hereby granted, free of charge, to any person obtaining a copy of
|
||||
// this software and associated documentation files (the "Software"), to deal in
|
||||
// the Software without restriction, including without limitation the rights to
|
||||
// use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
|
||||
// the Software, and to permit persons to whom the Software is furnished to do so,
|
||||
// subject to the following conditions:
|
||||
//
|
||||
// The above copyright notice and this permission notice shall be included in all
|
||||
// copies or substantial portions of the Software.
|
||||
//
|
||||
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
|
||||
// FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
|
||||
// COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
|
||||
// IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
|
||||
// CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
package srs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/ossrs/go-oryx-lib/errors"
|
||||
"github.com/ossrs/go-oryx-lib/logger"
|
||||
"github.com/pion/interceptor"
|
||||
"github.com/pion/rtp"
|
||||
"github.com/pion/sdp/v3"
|
||||
"github.com/pion/webrtc/v3"
|
||||
"github.com/pion/webrtc/v3/pkg/media"
|
||||
"github.com/pion/webrtc/v3/pkg/media/h264reader"
|
||||
"github.com/pion/webrtc/v3/pkg/media/oggreader"
|
||||
"io"
|
||||
"os"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
type videoIngester struct {
|
||||
sourceVideo string
|
||||
fps int
|
||||
markerInterceptor *RTPInterceptor
|
||||
sVideoTrack *webrtc.TrackLocalStaticSample
|
||||
sVideoSender *webrtc.RTPSender
|
||||
}
|
||||
|
||||
func NewVideoIngester(sourceVideo string) *videoIngester {
|
||||
return &videoIngester{markerInterceptor: &RTPInterceptor{}, sourceVideo: sourceVideo}
|
||||
}
|
||||
|
||||
func (v *videoIngester) Close() error {
|
||||
if v.sVideoSender != nil {
|
||||
v.sVideoSender.Stop()
|
||||
v.sVideoSender = nil
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (v *videoIngester) AddTrack(pc *webrtc.PeerConnection, fps int) error {
|
||||
v.fps = fps
|
||||
|
||||
mimeType, trackID := "video/H264", "video"
|
||||
if strings.HasSuffix(v.sourceVideo, ".ivf") {
|
||||
mimeType = "video/VP8"
|
||||
}
|
||||
|
||||
var err error
|
||||
v.sVideoTrack, err = webrtc.NewTrackLocalStaticSample(
|
||||
webrtc.RTPCodecCapability{MimeType: mimeType, ClockRate: 90000}, trackID, "pion",
|
||||
)
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "Create video track")
|
||||
}
|
||||
|
||||
v.sVideoSender, err = pc.AddTrack(v.sVideoTrack)
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "Add video track")
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func (v *videoIngester) Ingest(ctx context.Context) error {
|
||||
source, sender, track, fps := v.sourceVideo, v.sVideoSender, v.sVideoTrack, v.fps
|
||||
|
||||
f, err := os.Open(source)
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "Open file %v", source)
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
// TODO: FIXME: Support ivf for vp8.
|
||||
h264, err := h264reader.NewReader(f)
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "Open h264 %v", source)
|
||||
}
|
||||
|
||||
enc := sender.GetParameters().Encodings[0]
|
||||
codec := sender.GetParameters().Codecs[0]
|
||||
headers := sender.GetParameters().HeaderExtensions
|
||||
logger.Tf(ctx, "Video %v, tbn=%v, fps=%v, ssrc=%v, pt=%v, header=%v",
|
||||
codec.MimeType, codec.ClockRate, fps, enc.SSRC, codec.PayloadType, headers)
|
||||
|
||||
clock := newWallClock()
|
||||
sampleDuration := time.Duration(uint64(time.Millisecond) * 1000 / uint64(fps))
|
||||
for ctx.Err() == nil {
|
||||
var sps, pps *h264reader.NAL
|
||||
var oFrames []*h264reader.NAL
|
||||
for ctx.Err() == nil {
|
||||
frame, err := h264.NextNAL()
|
||||
if err == io.EOF {
|
||||
return io.EOF
|
||||
}
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "Read h264")
|
||||
}
|
||||
|
||||
oFrames = append(oFrames, frame)
|
||||
logger.If(ctx, "NALU %v PictureOrderCount=%v, ForbiddenZeroBit=%v, RefIdc=%v, %v bytes",
|
||||
frame.UnitType.String(), frame.PictureOrderCount, frame.ForbiddenZeroBit, frame.RefIdc, len(frame.Data))
|
||||
|
||||
if frame.UnitType == h264reader.NalUnitTypeSPS {
|
||||
sps = frame
|
||||
} else if frame.UnitType == h264reader.NalUnitTypePPS {
|
||||
pps = frame
|
||||
} else {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
var frames []*h264reader.NAL
|
||||
// Package SPS/PPS to STAP-A
|
||||
if sps != nil && pps != nil {
|
||||
stapA := packageAsSTAPA(sps, pps)
|
||||
frames = append(frames, stapA)
|
||||
}
|
||||
// Append other original frames.
|
||||
for _, frame := range oFrames {
|
||||
if frame.UnitType != h264reader.NalUnitTypeSPS && frame.UnitType != h264reader.NalUnitTypePPS {
|
||||
frames = append(frames, frame)
|
||||
}
|
||||
}
|
||||
|
||||
// Covert frames to sample(buffers).
|
||||
for i, frame := range frames {
|
||||
sample := media.Sample{Data: frame.Data, Duration: sampleDuration}
|
||||
// Use the sample timestamp for frames.
|
||||
if i != len(frames)-1 {
|
||||
sample.Duration = 0
|
||||
}
|
||||
|
||||
// For STAP-A, set marker to false, to make Chrome happy.
|
||||
if ri := v.markerInterceptor; ri.rtpWriter == nil {
|
||||
ri.rtpWriter = func(header *rtp.Header, payload []byte, attributes interceptor.Attributes) (int, error) {
|
||||
// TODO: Should we decode to check whether SPS/PPS?
|
||||
if len(payload) > 0 && payload[0]&0x1f == 24 {
|
||||
header.Marker = false // 24, STAP-A
|
||||
}
|
||||
return ri.nextRTPWriter.Write(header, payload, attributes)
|
||||
}
|
||||
}
|
||||
|
||||
if err = track.WriteSample(sample); err != nil {
|
||||
return errors.Wrapf(err, "Write sample")
|
||||
}
|
||||
}
|
||||
|
||||
if d := clock.Tick(sampleDuration); d > 0 {
|
||||
time.Sleep(d)
|
||||
}
|
||||
}
|
||||
|
||||
return ctx.Err()
|
||||
}
|
||||
|
||||
type audioIngester struct {
|
||||
sourceAudio string
|
||||
audioLevelInterceptor *RTPInterceptor
|
||||
sAudioTrack *webrtc.TrackLocalStaticSample
|
||||
sAudioSender *webrtc.RTPSender
|
||||
}
|
||||
|
||||
func NewAudioIngester(sourceAudio string) *audioIngester {
|
||||
return &audioIngester{audioLevelInterceptor: &RTPInterceptor{}, sourceAudio: sourceAudio}
|
||||
}
|
||||
|
||||
func (v *audioIngester) Close() error {
|
||||
if v.sAudioSender != nil {
|
||||
v.sAudioSender.Stop()
|
||||
v.sAudioSender = nil
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (v *audioIngester) AddTrack(pc *webrtc.PeerConnection) error {
|
||||
var err error
|
||||
|
||||
mimeType, trackID := "audio/opus", "audio"
|
||||
v.sAudioTrack, err = webrtc.NewTrackLocalStaticSample(
|
||||
webrtc.RTPCodecCapability{MimeType: mimeType, ClockRate: 48000, Channels: 2}, trackID, "pion",
|
||||
)
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "Create audio track")
|
||||
}
|
||||
|
||||
v.sAudioSender, err = pc.AddTrack(v.sAudioTrack)
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "Add audio track")
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (v *audioIngester) Ingest(ctx context.Context) error {
|
||||
source, sender, track := v.sourceAudio, v.sAudioSender, v.sAudioTrack
|
||||
|
||||
f, err := os.Open(source)
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "Open file %v", source)
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
ogg, _, err := oggreader.NewWith(f)
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "Open ogg %v", source)
|
||||
}
|
||||
|
||||
enc := sender.GetParameters().Encodings[0]
|
||||
codec := sender.GetParameters().Codecs[0]
|
||||
headers := sender.GetParameters().HeaderExtensions
|
||||
logger.Tf(ctx, "Audio %v, tbn=%v, channels=%v, ssrc=%v, pt=%v, header=%v",
|
||||
codec.MimeType, codec.ClockRate, codec.Channels, enc.SSRC, codec.PayloadType, headers)
|
||||
|
||||
// Whether should encode the audio-level in RTP header.
|
||||
var audioLevel *webrtc.RTPHeaderExtensionParameter
|
||||
for _, h := range headers {
|
||||
if h.URI == sdp.AudioLevelURI {
|
||||
audioLevel = &h
|
||||
}
|
||||
}
|
||||
|
||||
clock := newWallClock()
|
||||
var lastGranule uint64
|
||||
|
||||
for ctx.Err() == nil {
|
||||
pageData, pageHeader, err := ogg.ParseNextPage()
|
||||
if err == io.EOF {
|
||||
return io.EOF
|
||||
}
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "Read ogg")
|
||||
}
|
||||
|
||||
// The amount of samples is the difference between the last and current timestamp
|
||||
sampleCount := uint64(pageHeader.GranulePosition - lastGranule)
|
||||
lastGranule = pageHeader.GranulePosition
|
||||
sampleDuration := time.Duration(uint64(time.Millisecond) * 1000 * sampleCount / uint64(codec.ClockRate))
|
||||
|
||||
// For audio-level, set the extensions if negotiated.
|
||||
if ri := v.audioLevelInterceptor; ri.rtpWriter == nil {
|
||||
ri.rtpWriter = func(header *rtp.Header, payload []byte, attributes interceptor.Attributes) (int, error) {
|
||||
if audioLevel != nil {
|
||||
audioLevelPayload, err := new(rtp.AudioLevelExtension).Marshal()
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
header.SetExtension(uint8(audioLevel.ID), audioLevelPayload)
|
||||
}
|
||||
|
||||
return ri.nextRTPWriter.Write(header, payload, attributes)
|
||||
}
|
||||
}
|
||||
|
||||
if err = track.WriteSample(media.Sample{Data: pageData, Duration: sampleDuration}); err != nil {
|
||||
return errors.Wrapf(err, "Write sample")
|
||||
}
|
||||
|
||||
if d := clock.Tick(sampleDuration); d > 0 {
|
||||
time.Sleep(d)
|
||||
}
|
||||
}
|
||||
|
||||
return ctx.Err()
|
||||
}
|
||||
@ -0,0 +1,175 @@
|
||||
// The MIT License (MIT)
|
||||
//
|
||||
// Copyright (c) 2021 srs-bench(ossrs)
|
||||
//
|
||||
// Permission is hereby granted, free of charge, to any person obtaining a copy of
|
||||
// this software and associated documentation files (the "Software"), to deal in
|
||||
// the Software without restriction, including without limitation the rights to
|
||||
// use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
|
||||
// the Software, and to permit persons to whom the Software is furnished to do so,
|
||||
// subject to the following conditions:
|
||||
//
|
||||
// The above copyright notice and this permission notice shall be included in all
|
||||
// copies or substantial portions of the Software.
|
||||
//
|
||||
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
|
||||
// FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
|
||||
// COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
|
||||
// IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
|
||||
// CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
package srs
|
||||
|
||||
import (
|
||||
"github.com/pion/interceptor"
|
||||
"github.com/pion/rtcp"
|
||||
"github.com/pion/rtp"
|
||||
)
|
||||
|
||||
type RTPInterceptorOptionFunc func(i *RTPInterceptor)
|
||||
|
||||
// Common RTP packet interceptor for benchmark.
|
||||
// @remark Should never merge with RTCPInterceptor, because they has the same Write interface.
|
||||
type RTPInterceptor struct {
|
||||
localInfo *interceptor.StreamInfo
|
||||
remoteInfo *interceptor.StreamInfo
|
||||
// If rtpReader is nil, use the default next one to read.
|
||||
rtpReader interceptor.RTPReaderFunc
|
||||
nextRTPReader interceptor.RTPReader
|
||||
// If rtpWriter is nil, use the default next one to write.
|
||||
rtpWriter interceptor.RTPWriterFunc
|
||||
nextRTPWriter interceptor.RTPWriter
|
||||
BypassInterceptor
|
||||
}
|
||||
|
||||
func NewRTPInterceptor(options ...RTPInterceptorOptionFunc) *RTPInterceptor {
|
||||
v := &RTPInterceptor{}
|
||||
for _, opt := range options {
|
||||
opt(v)
|
||||
}
|
||||
return v
|
||||
}
|
||||
|
||||
func (v *RTPInterceptor) BindLocalStream(info *interceptor.StreamInfo, writer interceptor.RTPWriter) interceptor.RTPWriter {
|
||||
if v.localInfo != nil {
|
||||
return writer // Only handle one stream.
|
||||
}
|
||||
|
||||
v.localInfo = info
|
||||
v.nextRTPWriter = writer
|
||||
return v // Handle all RTP
|
||||
}
|
||||
|
||||
func (v *RTPInterceptor) Write(header *rtp.Header, payload []byte, attributes interceptor.Attributes) (int, error) {
|
||||
if v.rtpWriter != nil {
|
||||
return v.rtpWriter(header, payload, attributes)
|
||||
}
|
||||
return v.nextRTPWriter.Write(header, payload, attributes)
|
||||
}
|
||||
|
||||
func (v *RTPInterceptor) UnbindLocalStream(info *interceptor.StreamInfo) {
|
||||
if v.localInfo == nil || v.localInfo.ID != info.ID {
|
||||
return
|
||||
}
|
||||
v.localInfo = nil // Reset the interceptor.
|
||||
}
|
||||
|
||||
func (v *RTPInterceptor) BindRemoteStream(info *interceptor.StreamInfo, reader interceptor.RTPReader) interceptor.RTPReader {
|
||||
if v.remoteInfo != nil {
|
||||
return reader // Only handle one stream.
|
||||
}
|
||||
|
||||
v.nextRTPReader = reader
|
||||
return v // Handle all RTP
|
||||
}
|
||||
|
||||
func (v *RTPInterceptor) Read(b []byte, a interceptor.Attributes) (int, interceptor.Attributes, error) {
|
||||
if v.rtpReader != nil {
|
||||
return v.rtpReader(b, a)
|
||||
}
|
||||
return v.nextRTPReader.Read(b, a)
|
||||
}
|
||||
|
||||
func (v *RTPInterceptor) UnbindRemoteStream(info *interceptor.StreamInfo) {
|
||||
if v.remoteInfo == nil || v.remoteInfo.ID != info.ID {
|
||||
return
|
||||
}
|
||||
v.remoteInfo = nil
|
||||
}
|
||||
|
||||
type RTCPInterceptorOptionFunc func(i *RTCPInterceptor)
|
||||
|
||||
// Common RTCP packet interceptor for benchmark.
|
||||
// @remark Should never merge with RTPInterceptor, because they has the same Write interface.
|
||||
type RTCPInterceptor struct {
|
||||
// If rtcpReader is nil, use the default next one to read.
|
||||
rtcpReader interceptor.RTCPReaderFunc
|
||||
nextRTCPReader interceptor.RTCPReader
|
||||
// If rtcpWriter is nil, use the default next one to write.
|
||||
rtcpWriter interceptor.RTCPWriterFunc
|
||||
nextRTCPWriter interceptor.RTCPWriter
|
||||
BypassInterceptor
|
||||
}
|
||||
|
||||
func NewRTCPInterceptor(options ...RTCPInterceptorOptionFunc) *RTCPInterceptor {
|
||||
v := &RTCPInterceptor{}
|
||||
for _, opt := range options {
|
||||
opt(v)
|
||||
}
|
||||
return v
|
||||
}
|
||||
|
||||
func (v *RTCPInterceptor) BindRTCPReader(reader interceptor.RTCPReader) interceptor.RTCPReader {
|
||||
v.nextRTCPReader = reader
|
||||
return v // Handle all RTCP
|
||||
}
|
||||
|
||||
func (v *RTCPInterceptor) Read(b []byte, a interceptor.Attributes) (int, interceptor.Attributes, error) {
|
||||
if v.rtcpReader != nil {
|
||||
return v.rtcpReader(b, a)
|
||||
}
|
||||
return v.nextRTCPReader.Read(b, a)
|
||||
}
|
||||
|
||||
func (v *RTCPInterceptor) BindRTCPWriter(writer interceptor.RTCPWriter) interceptor.RTCPWriter {
|
||||
v.nextRTCPWriter = writer
|
||||
return v // Handle all RTCP
|
||||
}
|
||||
|
||||
func (v *RTCPInterceptor) Write(pkts []rtcp.Packet, attributes interceptor.Attributes) (int, error) {
|
||||
if v.rtcpWriter != nil {
|
||||
return v.rtcpWriter(pkts, attributes)
|
||||
}
|
||||
return v.nextRTCPWriter.Write(pkts, attributes)
|
||||
}
|
||||
|
||||
// Do nothing.
|
||||
type BypassInterceptor struct {
|
||||
interceptor.Interceptor
|
||||
}
|
||||
|
||||
func (v *BypassInterceptor) BindRTCPReader(reader interceptor.RTCPReader) interceptor.RTCPReader {
|
||||
return reader
|
||||
}
|
||||
|
||||
func (v *BypassInterceptor) BindRTCPWriter(writer interceptor.RTCPWriter) interceptor.RTCPWriter {
|
||||
return writer
|
||||
}
|
||||
|
||||
func (v *BypassInterceptor) BindLocalStream(info *interceptor.StreamInfo, writer interceptor.RTPWriter) interceptor.RTPWriter {
|
||||
return writer
|
||||
}
|
||||
|
||||
func (v *BypassInterceptor) UnbindLocalStream(info *interceptor.StreamInfo) {
|
||||
}
|
||||
|
||||
func (v *BypassInterceptor) BindRemoteStream(info *interceptor.StreamInfo, reader interceptor.RTPReader) interceptor.RTPReader {
|
||||
return reader
|
||||
}
|
||||
|
||||
func (v *BypassInterceptor) UnbindRemoteStream(info *interceptor.StreamInfo) {
|
||||
}
|
||||
|
||||
func (v *BypassInterceptor) Close() error {
|
||||
return nil
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,723 @@
|
||||
// The MIT License (MIT)
|
||||
//
|
||||
// Copyright (c) 2021 srs-bench(ossrs)
|
||||
//
|
||||
// Permission is hereby granted, free of charge, to any person obtaining a copy of
|
||||
// this software and associated documentation files (the "Software"), to deal in
|
||||
// the Software without restriction, including without limitation the rights to
|
||||
// use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
|
||||
// the Software, and to permit persons to whom the Software is furnished to do so,
|
||||
// subject to the following conditions:
|
||||
//
|
||||
// The above copyright notice and this permission notice shall be included in all
|
||||
// copies or substantial portions of the Software.
|
||||
//
|
||||
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
|
||||
// FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
|
||||
// COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
|
||||
// IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
|
||||
// CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
package srs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"flag"
|
||||
"fmt"
|
||||
"github.com/ossrs/go-oryx-lib/errors"
|
||||
"github.com/ossrs/go-oryx-lib/logger"
|
||||
vnet_proxy "github.com/ossrs/srs-bench/vnet"
|
||||
"github.com/pion/interceptor"
|
||||
"github.com/pion/logging"
|
||||
"github.com/pion/rtcp"
|
||||
"github.com/pion/transport/vnet"
|
||||
"github.com/pion/webrtc/v3"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"net/http"
|
||||
"os"
|
||||
"path"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
var srsSchema = "http"
|
||||
var srsHttps = flag.Bool("srs-https", false, "Whther connect to HTTPS-API")
|
||||
var srsServer = flag.String("srs-server", "127.0.0.1", "The RTC server to connect to")
|
||||
var srsStream = flag.String("srs-stream", "/rtc/regression", "The RTC stream to play")
|
||||
var srsLog = flag.Bool("srs-log", false, "Whether enable the detail log")
|
||||
var srsTimeout = flag.Int("srs-timeout", 5000, "For each case, the timeout in ms")
|
||||
var srsPlayPLI = flag.Int("srs-play-pli", 5000, "The PLI interval in seconds for player.")
|
||||
var srsPlayOKPackets = flag.Int("srs-play-ok-packets", 10, "If got N packets, it's ok, or fail")
|
||||
var srsPublishOKPackets = flag.Int("srs-publish-ok-packets", 10, "If send N packets, it's ok, or fail")
|
||||
var srsPublishAudio = flag.String("srs-publish-audio", "avatar.ogg", "The audio file for publisher.")
|
||||
var srsPublishVideo = flag.String("srs-publish-video", "avatar.h264", "The video file for publisher.")
|
||||
var srsPublishVideoFps = flag.Int("srs-publish-video-fps", 25, "The video fps for publisher.")
|
||||
var srsVnetClientIP = flag.String("srs-vnet-client-ip", "192.168.168.168", "The client ip in pion/vnet.")
|
||||
var srsDTLSDropPackets = flag.Int("srs-dtls-drop-packets", 5, "If dropped N packets, it's ok, or fail")
|
||||
|
||||
func prepareTest() error {
|
||||
var err error
|
||||
|
||||
// Should parse it first.
|
||||
flag.Parse()
|
||||
|
||||
// The stream should starts with /, for example, /rtc/regression
|
||||
if !strings.HasPrefix(*srsStream, "/") {
|
||||
*srsStream = "/" + *srsStream
|
||||
}
|
||||
|
||||
// Generate srs protocol from whether use HTTPS.
|
||||
if *srsHttps {
|
||||
srsSchema = "https"
|
||||
}
|
||||
|
||||
// Check file.
|
||||
tryOpenFile := func(filename string) (string, error) {
|
||||
if filename == "" {
|
||||
return filename, nil
|
||||
}
|
||||
|
||||
f, err := os.Open(filename)
|
||||
if err != nil {
|
||||
nfilename := path.Join("../", filename)
|
||||
f2, err := os.Open(nfilename)
|
||||
if err != nil {
|
||||
return filename, errors.Wrapf(err, "No video file at %v or %v", filename, nfilename)
|
||||
}
|
||||
defer f2.Close()
|
||||
|
||||
return nfilename, nil
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
return filename, nil
|
||||
}
|
||||
|
||||
if *srsPublishVideo, err = tryOpenFile(*srsPublishVideo); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if *srsPublishAudio, err = tryOpenFile(*srsPublishAudio); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func TestMain(m *testing.M) {
|
||||
if err := prepareTest(); err != nil {
|
||||
logger.Ef(nil, "Prepare test fail, err %+v", err)
|
||||
os.Exit(-1)
|
||||
}
|
||||
|
||||
// Disable the logger during all tests.
|
||||
if *srsLog == false {
|
||||
olw := logger.Switch(ioutil.Discard)
|
||||
defer func() {
|
||||
logger.Switch(olw)
|
||||
}()
|
||||
}
|
||||
|
||||
os.Exit(m.Run())
|
||||
}
|
||||
|
||||
type TestWebRTCAPIOptionFunc func(api *TestWebRTCAPI)
|
||||
|
||||
type TestWebRTCAPI struct {
|
||||
// The options to setup the api.
|
||||
options []TestWebRTCAPIOptionFunc
|
||||
// The api and settings.
|
||||
api *webrtc.API
|
||||
mediaEngine *webrtc.MediaEngine
|
||||
registry *interceptor.Registry
|
||||
settingEngine *webrtc.SettingEngine
|
||||
// The vnet router, can be shared by different apis, but we do not share it.
|
||||
router *vnet.Router
|
||||
// The network for api.
|
||||
network *vnet.Net
|
||||
// The vnet UDP proxy bind to the router.
|
||||
proxy *vnet_proxy.UDPProxy
|
||||
}
|
||||
|
||||
func NewTestWebRTCAPI(options ...TestWebRTCAPIOptionFunc) (*TestWebRTCAPI, error) {
|
||||
v := &TestWebRTCAPI{}
|
||||
|
||||
v.mediaEngine = &webrtc.MediaEngine{}
|
||||
if err := v.mediaEngine.RegisterDefaultCodecs(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
v.registry = &interceptor.Registry{}
|
||||
if err := webrtc.RegisterDefaultInterceptors(v.mediaEngine, v.registry); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
for _, setup := range options {
|
||||
setup(v)
|
||||
}
|
||||
|
||||
v.settingEngine = &webrtc.SettingEngine{}
|
||||
|
||||
return v, nil
|
||||
}
|
||||
|
||||
func (v *TestWebRTCAPI) Close() error {
|
||||
if v.proxy != nil {
|
||||
v.proxy.Close()
|
||||
v.proxy = nil
|
||||
}
|
||||
|
||||
if v.router != nil {
|
||||
v.router.Stop()
|
||||
v.router = nil
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (v *TestWebRTCAPI) Setup(vnetClientIP string, options ...TestWebRTCAPIOptionFunc) error {
|
||||
// Setting engine for https://github.com/pion/transport/tree/master/vnet
|
||||
setupVnet := func(vnetClientIP string) (err error) {
|
||||
// We create a private router for a api, however, it's possible to share the
|
||||
// same router between apis.
|
||||
if v.router, err = vnet.NewRouter(&vnet.RouterConfig{
|
||||
CIDR: "0.0.0.0/0", // Accept all ip, no sub router.
|
||||
LoggerFactory: logging.NewDefaultLoggerFactory(),
|
||||
}); err != nil {
|
||||
return errors.Wrapf(err, "create router for api")
|
||||
}
|
||||
|
||||
// Each api should bind to a network, however, it's possible to share it
|
||||
// for different apis.
|
||||
v.network = vnet.NewNet(&vnet.NetConfig{
|
||||
StaticIP: vnetClientIP,
|
||||
})
|
||||
|
||||
if err = v.router.AddNet(v.network); err != nil {
|
||||
return errors.Wrapf(err, "create network for api")
|
||||
}
|
||||
|
||||
v.settingEngine.SetVNet(v.network)
|
||||
|
||||
// Create a proxy bind to the router.
|
||||
if v.proxy, err = vnet_proxy.NewProxy(v.router); err != nil {
|
||||
return errors.Wrapf(err, "create proxy for router")
|
||||
}
|
||||
|
||||
return v.router.Start()
|
||||
}
|
||||
if err := setupVnet(vnetClientIP); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, setup := range options {
|
||||
setup(v)
|
||||
}
|
||||
|
||||
for _, setup := range v.options {
|
||||
setup(v)
|
||||
}
|
||||
|
||||
v.api = webrtc.NewAPI(
|
||||
webrtc.WithMediaEngine(v.mediaEngine),
|
||||
webrtc.WithInterceptorRegistry(v.registry),
|
||||
webrtc.WithSettingEngine(*v.settingEngine),
|
||||
)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (v *TestWebRTCAPI) NewPeerConnection(configuration webrtc.Configuration) (*webrtc.PeerConnection, error) {
|
||||
return v.api.NewPeerConnection(configuration)
|
||||
}
|
||||
|
||||
type TestPlayerOptionFunc func(p *TestPlayer)
|
||||
|
||||
type TestPlayer struct {
|
||||
pc *webrtc.PeerConnection
|
||||
receivers []*webrtc.RTPReceiver
|
||||
// root api object
|
||||
api *TestWebRTCAPI
|
||||
// Optional suffix for stream url.
|
||||
streamSuffix string
|
||||
}
|
||||
|
||||
func NewTestPlayer(api *TestWebRTCAPI, options ...TestPlayerOptionFunc) *TestPlayer {
|
||||
v := &TestPlayer{api: api}
|
||||
|
||||
for _, opt := range options {
|
||||
opt(v)
|
||||
}
|
||||
|
||||
return v
|
||||
}
|
||||
|
||||
func (v *TestPlayer) Close() error {
|
||||
if v.pc != nil {
|
||||
v.pc.Close()
|
||||
v.pc = nil
|
||||
}
|
||||
|
||||
for _, receiver := range v.receivers {
|
||||
receiver.Stop()
|
||||
}
|
||||
v.receivers = nil
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (v *TestPlayer) Run(ctx context.Context, cancel context.CancelFunc) error {
|
||||
r := fmt.Sprintf("%v://%v%v", srsSchema, *srsServer, *srsStream)
|
||||
if v.streamSuffix != "" {
|
||||
r = fmt.Sprintf("%v-%v", r, v.streamSuffix)
|
||||
}
|
||||
pli := time.Duration(*srsPlayPLI) * time.Millisecond
|
||||
logger.Tf(ctx, "Start play url=%v", r)
|
||||
|
||||
pc, err := v.api.NewPeerConnection(webrtc.Configuration{})
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "Create PC")
|
||||
}
|
||||
v.pc = pc
|
||||
|
||||
pc.AddTransceiverFromKind(webrtc.RTPCodecTypeAudio, webrtc.RTPTransceiverInit{
|
||||
Direction: webrtc.RTPTransceiverDirectionRecvonly,
|
||||
})
|
||||
pc.AddTransceiverFromKind(webrtc.RTPCodecTypeVideo, webrtc.RTPTransceiverInit{
|
||||
Direction: webrtc.RTPTransceiverDirectionRecvonly,
|
||||
})
|
||||
|
||||
offer, err := pc.CreateOffer(nil)
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "Create Offer")
|
||||
}
|
||||
|
||||
if err := pc.SetLocalDescription(offer); err != nil {
|
||||
return errors.Wrapf(err, "Set offer %v", offer)
|
||||
}
|
||||
|
||||
answer, err := apiRtcRequest(ctx, "/rtc/v1/play", r, offer.SDP)
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "Api request offer=%v", offer.SDP)
|
||||
}
|
||||
|
||||
// Start a proxy for real server and vnet.
|
||||
if address, err := parseAddressOfCandidate(answer); err != nil {
|
||||
return errors.Wrapf(err, "parse address of %v", answer)
|
||||
} else if err := v.api.proxy.Proxy(v.api.network, address); err != nil {
|
||||
return errors.Wrapf(err, "proxy %v to %v", v.api.network, address)
|
||||
}
|
||||
|
||||
if err := pc.SetRemoteDescription(webrtc.SessionDescription{
|
||||
Type: webrtc.SDPTypeAnswer, SDP: answer,
|
||||
}); err != nil {
|
||||
return errors.Wrapf(err, "Set answer %v", answer)
|
||||
}
|
||||
|
||||
handleTrack := func(ctx context.Context, track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) error {
|
||||
// Send a PLI on an interval so that the publisher is pushing a keyframe
|
||||
go func() {
|
||||
if track.Kind() == webrtc.RTPCodecTypeAudio {
|
||||
return
|
||||
}
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(pli):
|
||||
_ = pc.WriteRTCP([]rtcp.Packet{&rtcp.PictureLossIndication{
|
||||
MediaSSRC: uint32(track.SSRC()),
|
||||
}})
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
v.receivers = append(v.receivers, receiver)
|
||||
|
||||
for ctx.Err() == nil {
|
||||
_, _, err := track.ReadRTP()
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "Read RTP")
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
pc.OnTrack(func(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) {
|
||||
err = handleTrack(ctx, track, receiver)
|
||||
if err != nil {
|
||||
codec := track.Codec()
|
||||
err = errors.Wrapf(err, "Handle track %v, pt=%v", codec.MimeType, codec.PayloadType)
|
||||
cancel()
|
||||
}
|
||||
})
|
||||
|
||||
pc.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) {
|
||||
if state == webrtc.ICEConnectionStateFailed || state == webrtc.ICEConnectionStateClosed {
|
||||
err = errors.Errorf("Close for ICE state %v", state)
|
||||
cancel()
|
||||
}
|
||||
})
|
||||
|
||||
<-ctx.Done()
|
||||
return err
|
||||
}
|
||||
|
||||
type TestPublisherOptionFunc func(p *TestPublisher)
|
||||
|
||||
type TestPublisher struct {
|
||||
onOffer func(s *webrtc.SessionDescription) error
|
||||
onAnswer func(s *webrtc.SessionDescription) error
|
||||
iceReadyCancel context.CancelFunc
|
||||
// internal objects
|
||||
aIngester *audioIngester
|
||||
vIngester *videoIngester
|
||||
pc *webrtc.PeerConnection
|
||||
// root api object
|
||||
api *TestWebRTCAPI
|
||||
// Optional suffix for stream url.
|
||||
streamSuffix string
|
||||
}
|
||||
|
||||
func NewTestPublisher(api *TestWebRTCAPI, options ...TestPublisherOptionFunc) *TestPublisher {
|
||||
sourceVideo, sourceAudio := *srsPublishVideo, *srsPublishAudio
|
||||
|
||||
v := &TestPublisher{api: api}
|
||||
|
||||
for _, opt := range options {
|
||||
opt(v)
|
||||
}
|
||||
|
||||
// Create ingesters.
|
||||
if sourceAudio != "" {
|
||||
v.aIngester = NewAudioIngester(sourceAudio)
|
||||
}
|
||||
if sourceVideo != "" {
|
||||
v.vIngester = NewVideoIngester(sourceVideo)
|
||||
}
|
||||
|
||||
// Setup the interceptors for packets.
|
||||
api.options = append(api.options, func(api *TestWebRTCAPI) {
|
||||
// Filter for RTCP packets.
|
||||
rtcpInterceptor := &RTCPInterceptor{}
|
||||
rtcpInterceptor.rtcpReader = func(buf []byte, attributes interceptor.Attributes) (int, interceptor.Attributes, error) {
|
||||
return rtcpInterceptor.nextRTCPReader.Read(buf, attributes)
|
||||
}
|
||||
rtcpInterceptor.rtcpWriter = func(pkts []rtcp.Packet, attributes interceptor.Attributes) (int, error) {
|
||||
return rtcpInterceptor.nextRTCPWriter.Write(pkts, attributes)
|
||||
}
|
||||
api.registry.Add(rtcpInterceptor)
|
||||
|
||||
// Filter for ingesters.
|
||||
if sourceAudio != "" {
|
||||
api.registry.Add(v.aIngester.audioLevelInterceptor)
|
||||
}
|
||||
if sourceVideo != "" {
|
||||
api.registry.Add(v.vIngester.markerInterceptor)
|
||||
}
|
||||
})
|
||||
|
||||
return v
|
||||
}
|
||||
|
||||
func (v *TestPublisher) Close() error {
|
||||
if v.vIngester != nil {
|
||||
v.vIngester.Close()
|
||||
}
|
||||
|
||||
if v.aIngester != nil {
|
||||
v.aIngester.Close()
|
||||
}
|
||||
|
||||
if v.pc != nil {
|
||||
v.pc.Close()
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (v *TestPublisher) SetStreamSuffix(suffix string) *TestPublisher {
|
||||
v.streamSuffix = suffix
|
||||
return v
|
||||
}
|
||||
|
||||
func (v *TestPublisher) Run(ctx context.Context, cancel context.CancelFunc) error {
|
||||
r := fmt.Sprintf("%v://%v%v", srsSchema, *srsServer, *srsStream)
|
||||
if v.streamSuffix != "" {
|
||||
r = fmt.Sprintf("%v-%v", r, v.streamSuffix)
|
||||
}
|
||||
sourceVideo, sourceAudio, fps := *srsPublishVideo, *srsPublishAudio, *srsPublishVideoFps
|
||||
|
||||
logger.Tf(ctx, "Start publish url=%v, audio=%v, video=%v, fps=%v",
|
||||
r, sourceAudio, sourceVideo, fps)
|
||||
|
||||
pc, err := v.api.NewPeerConnection(webrtc.Configuration{})
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "Create PC")
|
||||
}
|
||||
v.pc = pc
|
||||
|
||||
if v.vIngester != nil {
|
||||
if err := v.vIngester.AddTrack(pc, fps); err != nil {
|
||||
return errors.Wrapf(err, "Add track")
|
||||
}
|
||||
defer v.vIngester.Close()
|
||||
}
|
||||
|
||||
if v.aIngester != nil {
|
||||
if err := v.aIngester.AddTrack(pc); err != nil {
|
||||
return errors.Wrapf(err, "Add track")
|
||||
}
|
||||
defer v.aIngester.Close()
|
||||
}
|
||||
|
||||
offer, err := pc.CreateOffer(nil)
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "Create Offer")
|
||||
}
|
||||
|
||||
if err := pc.SetLocalDescription(offer); err != nil {
|
||||
return errors.Wrapf(err, "Set offer %v", offer)
|
||||
}
|
||||
|
||||
if v.onOffer != nil {
|
||||
if err := v.onOffer(&offer); err != nil {
|
||||
return errors.Wrapf(err, "sdp %v %v", offer.Type, offer.SDP)
|
||||
}
|
||||
}
|
||||
|
||||
answerSDP, err := apiRtcRequest(ctx, "/rtc/v1/publish", r, offer.SDP)
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "Api request offer=%v", offer.SDP)
|
||||
}
|
||||
|
||||
// Start a proxy for real server and vnet.
|
||||
if address, err := parseAddressOfCandidate(answerSDP); err != nil {
|
||||
return errors.Wrapf(err, "parse address of %v", answerSDP)
|
||||
} else if err := v.api.proxy.Proxy(v.api.network, address); err != nil {
|
||||
return errors.Wrapf(err, "proxy %v to %v", v.api.network, address)
|
||||
}
|
||||
|
||||
answer := &webrtc.SessionDescription{
|
||||
Type: webrtc.SDPTypeAnswer, SDP: answerSDP,
|
||||
}
|
||||
if v.onAnswer != nil {
|
||||
if err := v.onAnswer(answer); err != nil {
|
||||
return errors.Wrapf(err, "on answerSDP")
|
||||
}
|
||||
}
|
||||
|
||||
if err := pc.SetRemoteDescription(*answer); err != nil {
|
||||
return errors.Wrapf(err, "Set answerSDP %v", answerSDP)
|
||||
}
|
||||
|
||||
logger.Tf(ctx, "State signaling=%v, ice=%v, conn=%v", pc.SignalingState(), pc.ICEConnectionState(), pc.ConnectionState())
|
||||
|
||||
// ICE state management.
|
||||
pc.OnICEGatheringStateChange(func(state webrtc.ICEGathererState) {
|
||||
logger.Tf(ctx, "ICE gather state %v", state)
|
||||
})
|
||||
pc.OnICECandidate(func(candidate *webrtc.ICECandidate) {
|
||||
logger.Tf(ctx, "ICE candidate %v %v:%v", candidate.Protocol, candidate.Address, candidate.Port)
|
||||
|
||||
})
|
||||
pc.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) {
|
||||
logger.Tf(ctx, "ICE state %v", state)
|
||||
})
|
||||
|
||||
pc.OnSignalingStateChange(func(state webrtc.SignalingState) {
|
||||
logger.Tf(ctx, "Signaling state %v", state)
|
||||
})
|
||||
|
||||
if v.aIngester != nil {
|
||||
v.aIngester.sAudioSender.Transport().OnStateChange(func(state webrtc.DTLSTransportState) {
|
||||
logger.Tf(ctx, "DTLS state %v", state)
|
||||
})
|
||||
}
|
||||
|
||||
pcDone, pcDoneCancel := context.WithCancel(context.Background())
|
||||
pc.OnConnectionStateChange(func(state webrtc.PeerConnectionState) {
|
||||
logger.Tf(ctx, "PC state %v", state)
|
||||
|
||||
if state == webrtc.PeerConnectionStateConnected {
|
||||
pcDoneCancel()
|
||||
if v.iceReadyCancel != nil {
|
||||
v.iceReadyCancel()
|
||||
}
|
||||
}
|
||||
|
||||
if state == webrtc.PeerConnectionStateFailed || state == webrtc.PeerConnectionStateClosed {
|
||||
err = errors.Errorf("Close for PC state %v", state)
|
||||
cancel()
|
||||
}
|
||||
})
|
||||
|
||||
// Wait for event from context or tracks.
|
||||
var wg sync.WaitGroup
|
||||
var finalErr error
|
||||
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer logger.Tf(ctx, "ingest notify done")
|
||||
|
||||
<-ctx.Done()
|
||||
|
||||
if v.aIngester != nil && v.aIngester.sAudioSender != nil {
|
||||
v.aIngester.sAudioSender.Stop()
|
||||
}
|
||||
|
||||
if v.vIngester != nil && v.vIngester.sVideoSender != nil {
|
||||
v.vIngester.sVideoSender.Stop()
|
||||
}
|
||||
}()
|
||||
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer cancel()
|
||||
|
||||
if v.aIngester == nil {
|
||||
return
|
||||
}
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-pcDone.Done():
|
||||
}
|
||||
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer logger.Tf(ctx, "aingester sender read done")
|
||||
|
||||
buf := make([]byte, 1500)
|
||||
for ctx.Err() == nil {
|
||||
if _, _, err := v.aIngester.sAudioSender.Read(buf); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
for {
|
||||
if err := v.aIngester.Ingest(ctx); err != nil {
|
||||
if err == io.EOF {
|
||||
logger.Tf(ctx, "aingester retry for %v", err)
|
||||
continue
|
||||
}
|
||||
if err != context.Canceled {
|
||||
finalErr = errors.Wrapf(err, "audio")
|
||||
}
|
||||
|
||||
logger.Tf(ctx, "aingester err=%v, final=%v", err, finalErr)
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer cancel()
|
||||
|
||||
if v.vIngester == nil {
|
||||
return
|
||||
}
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-pcDone.Done():
|
||||
logger.Tf(ctx, "PC(ICE+DTLS+SRTP) done, start ingest video %v", sourceVideo)
|
||||
}
|
||||
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer logger.Tf(ctx, "vingester sender read done")
|
||||
|
||||
buf := make([]byte, 1500)
|
||||
for ctx.Err() == nil {
|
||||
// The Read() might block in r.rtcpInterceptor.Read(b, a),
|
||||
// so that the Stop() can not stop it.
|
||||
if _, _, err := v.vIngester.sVideoSender.Read(buf); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
for {
|
||||
if err := v.vIngester.Ingest(ctx); err != nil {
|
||||
if err == io.EOF {
|
||||
logger.Tf(ctx, "vingester retry for %v", err)
|
||||
continue
|
||||
}
|
||||
if err != context.Canceled {
|
||||
finalErr = errors.Wrapf(err, "video")
|
||||
}
|
||||
|
||||
logger.Tf(ctx, "vingester err=%v, final=%v", err, finalErr)
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
wg.Wait()
|
||||
|
||||
logger.Tf(ctx, "ingester done ctx=%v, final=%v", ctx.Err(), finalErr)
|
||||
if finalErr != nil {
|
||||
return finalErr
|
||||
}
|
||||
return ctx.Err()
|
||||
}
|
||||
|
||||
func TestRTCServerVersion(t *testing.T) {
|
||||
api := fmt.Sprintf("http://%v:1985/api/v1/versions", *srsServer)
|
||||
req, err := http.NewRequest("POST", api, nil)
|
||||
if err != nil {
|
||||
t.Errorf("Request %v", api)
|
||||
return
|
||||
}
|
||||
|
||||
res, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
t.Errorf("Do request %v", api)
|
||||
return
|
||||
}
|
||||
|
||||
b, err := ioutil.ReadAll(res.Body)
|
||||
if err != nil {
|
||||
t.Errorf("Read body of %v", api)
|
||||
return
|
||||
}
|
||||
|
||||
obj := struct {
|
||||
Code int `json:"code"`
|
||||
Server string `json:"server"`
|
||||
Data struct {
|
||||
Major int `json:"major"`
|
||||
Minor int `json:"minor"`
|
||||
Revision int `json:"revision"`
|
||||
Version string `json:"version"`
|
||||
} `json:"data"`
|
||||
}{}
|
||||
if err := json.Unmarshal(b, &obj); err != nil {
|
||||
t.Errorf("Parse %v", string(b))
|
||||
return
|
||||
}
|
||||
if obj.Code != 0 {
|
||||
t.Errorf("Server err code=%v, server=%v", obj.Code, obj.Server)
|
||||
return
|
||||
}
|
||||
if obj.Data.Major == 0 && obj.Data.Minor == 0 {
|
||||
t.Errorf("Invalid version %v", obj.Data)
|
||||
return
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,278 @@
|
||||
// The MIT License (MIT)
|
||||
//
|
||||
// Copyright (c) 2021 srs-bench(ossrs)
|
||||
//
|
||||
// Permission is hereby granted, free of charge, to any person obtaining a copy of
|
||||
// this software and associated documentation files (the "Software"), to deal in
|
||||
// the Software without restriction, including without limitation the rights to
|
||||
// use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
|
||||
// the Software, and to permit persons to whom the Software is furnished to do so,
|
||||
// subject to the following conditions:
|
||||
//
|
||||
// The above copyright notice and this permission notice shall be included in all
|
||||
// copies or substantial portions of the Software.
|
||||
//
|
||||
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
|
||||
// FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
|
||||
// COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
|
||||
// IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
|
||||
// CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
package vnet_test
|
||||
|
||||
import (
|
||||
"net"
|
||||
|
||||
vnet_proxy "github.com/ossrs/srs-bench/vnet"
|
||||
"github.com/pion/logging"
|
||||
"github.com/pion/transport/vnet"
|
||||
)
|
||||
|
||||
// Proxy many vnet endpoint to one real server endpoint.
|
||||
// For example:
|
||||
// vnet(10.0.0.11:5787) => proxy => 192.168.1.10:8000
|
||||
// vnet(10.0.0.11:5788) => proxy => 192.168.1.10:8000
|
||||
// vnet(10.0.0.11:5789) => proxy => 192.168.1.10:8000
|
||||
func ExampleUDPProxyManyToOne() { // nolint:govet
|
||||
var clientNetwork *vnet.Net
|
||||
|
||||
var serverAddr *net.UDPAddr
|
||||
if addr, err := net.ResolveUDPAddr("udp4", "192.168.1.10:8000"); err != nil {
|
||||
// handle error
|
||||
} else {
|
||||
serverAddr = addr
|
||||
}
|
||||
|
||||
// Setup the network and proxy.
|
||||
if true {
|
||||
// Create vnet WAN with one endpoint, please read from
|
||||
// https://github.com/pion/transport/tree/master/vnet#example-wan-with-one-endpoint-vnet
|
||||
router, err := vnet.NewRouter(&vnet.RouterConfig{
|
||||
CIDR: "0.0.0.0/0",
|
||||
LoggerFactory: logging.NewDefaultLoggerFactory(),
|
||||
})
|
||||
if err != nil {
|
||||
// handle error
|
||||
}
|
||||
|
||||
// Create a network and add to router, for example, for client.
|
||||
clientNetwork = vnet.NewNet(&vnet.NetConfig{
|
||||
StaticIP: "10.0.0.11",
|
||||
})
|
||||
if err = router.AddNet(clientNetwork); err != nil {
|
||||
// handle error
|
||||
}
|
||||
|
||||
// Start the router.
|
||||
if err = router.Start(); err != nil {
|
||||
// handle error
|
||||
}
|
||||
defer router.Stop() // nolint:errcheck
|
||||
|
||||
// Create a proxy, bind to the router.
|
||||
proxy, err := vnet_proxy.NewProxy(router)
|
||||
if err != nil {
|
||||
// handle error
|
||||
}
|
||||
defer proxy.Close() // nolint:errcheck
|
||||
|
||||
// Start to proxy some addresses, clientNetwork is a hit for proxy,
|
||||
// that the client in vnet is from this network.
|
||||
if err := proxy.Proxy(clientNetwork, serverAddr); err != nil {
|
||||
// handle error
|
||||
}
|
||||
}
|
||||
|
||||
// Now, all packets from client, will be proxy to real server, vice versa.
|
||||
client0, err := clientNetwork.ListenPacket("udp4", "10.0.0.11:5787")
|
||||
if err != nil {
|
||||
// handle error
|
||||
}
|
||||
_, _ = client0.WriteTo([]byte("Hello"), serverAddr)
|
||||
|
||||
client1, err := clientNetwork.ListenPacket("udp4", "10.0.0.11:5788")
|
||||
if err != nil {
|
||||
// handle error
|
||||
}
|
||||
_, _ = client1.WriteTo([]byte("Hello"), serverAddr)
|
||||
|
||||
client2, err := clientNetwork.ListenPacket("udp4", "10.0.0.11:5789")
|
||||
if err != nil {
|
||||
// handle error
|
||||
}
|
||||
_, _ = client2.WriteTo([]byte("Hello"), serverAddr)
|
||||
}
|
||||
|
||||
// Proxy many vnet endpoint to one real server endpoint.
|
||||
// For example:
|
||||
// vnet(10.0.0.11:5787) => proxy => 192.168.1.10:8000
|
||||
// vnet(10.0.0.11:5788) => proxy => 192.168.1.10:8000
|
||||
func ExampleUDPProxyMultileTimes() { // nolint:govet
|
||||
var clientNetwork *vnet.Net
|
||||
|
||||
var serverAddr *net.UDPAddr
|
||||
if addr, err := net.ResolveUDPAddr("udp4", "192.168.1.10:8000"); err != nil {
|
||||
// handle error
|
||||
} else {
|
||||
serverAddr = addr
|
||||
}
|
||||
|
||||
// Setup the network and proxy.
|
||||
var proxy *vnet_proxy.UDPProxy
|
||||
if true {
|
||||
// Create vnet WAN with one endpoint, please read from
|
||||
// https://github.com/pion/transport/tree/master/vnet#example-wan-with-one-endpoint-vnet
|
||||
router, err := vnet.NewRouter(&vnet.RouterConfig{
|
||||
CIDR: "0.0.0.0/0",
|
||||
LoggerFactory: logging.NewDefaultLoggerFactory(),
|
||||
})
|
||||
if err != nil {
|
||||
// handle error
|
||||
}
|
||||
|
||||
// Create a network and add to router, for example, for client.
|
||||
clientNetwork = vnet.NewNet(&vnet.NetConfig{
|
||||
StaticIP: "10.0.0.11",
|
||||
})
|
||||
if err = router.AddNet(clientNetwork); err != nil {
|
||||
// handle error
|
||||
}
|
||||
|
||||
// Start the router.
|
||||
if err = router.Start(); err != nil {
|
||||
// handle error
|
||||
}
|
||||
defer router.Stop() // nolint:errcheck
|
||||
|
||||
// Create a proxy, bind to the router.
|
||||
proxy, err = vnet_proxy.NewProxy(router)
|
||||
if err != nil {
|
||||
// handle error
|
||||
}
|
||||
defer proxy.Close() // nolint:errcheck
|
||||
}
|
||||
|
||||
if true {
|
||||
// Start to proxy some addresses, clientNetwork is a hit for proxy,
|
||||
// that the client in vnet is from this network.
|
||||
if err := proxy.Proxy(clientNetwork, serverAddr); err != nil {
|
||||
// handle error
|
||||
}
|
||||
|
||||
// Now, all packets from client, will be proxy to real server, vice versa.
|
||||
client0, err := clientNetwork.ListenPacket("udp4", "10.0.0.11:5787")
|
||||
if err != nil {
|
||||
// handle error
|
||||
}
|
||||
_, _ = client0.WriteTo([]byte("Hello"), serverAddr)
|
||||
}
|
||||
|
||||
if true {
|
||||
// It's ok to proxy multiple times, for example, the publisher and player
|
||||
// might need to proxy when got answer.
|
||||
if err := proxy.Proxy(clientNetwork, serverAddr); err != nil {
|
||||
// handle error
|
||||
}
|
||||
|
||||
client1, err := clientNetwork.ListenPacket("udp4", "10.0.0.11:5788")
|
||||
if err != nil {
|
||||
// handle error
|
||||
}
|
||||
_, _ = client1.WriteTo([]byte("Hello"), serverAddr)
|
||||
}
|
||||
}
|
||||
|
||||
// Proxy one vnet endpoint to one real server endpoint.
|
||||
// For example:
|
||||
// vnet(10.0.0.11:5787) => proxy0 => 192.168.1.10:8000
|
||||
// vnet(10.0.0.11:5788) => proxy1 => 192.168.1.10:8001
|
||||
// vnet(10.0.0.11:5789) => proxy2 => 192.168.1.10:8002
|
||||
func ExampleUDPProxyOneToOne() { // nolint:govet
|
||||
var clientNetwork *vnet.Net
|
||||
|
||||
var serverAddr0 *net.UDPAddr
|
||||
if addr, err := net.ResolveUDPAddr("udp4", "192.168.1.10:8000"); err != nil {
|
||||
// handle error
|
||||
} else {
|
||||
serverAddr0 = addr
|
||||
}
|
||||
|
||||
var serverAddr1 *net.UDPAddr
|
||||
if addr, err := net.ResolveUDPAddr("udp4", "192.168.1.10:8001"); err != nil {
|
||||
// handle error
|
||||
} else {
|
||||
serverAddr1 = addr
|
||||
}
|
||||
|
||||
var serverAddr2 *net.UDPAddr
|
||||
if addr, err := net.ResolveUDPAddr("udp4", "192.168.1.10:8002"); err != nil {
|
||||
// handle error
|
||||
} else {
|
||||
serverAddr2 = addr
|
||||
}
|
||||
|
||||
// Setup the network and proxy.
|
||||
if true {
|
||||
// Create vnet WAN with one endpoint, please read from
|
||||
// https://github.com/pion/transport/tree/master/vnet#example-wan-with-one-endpoint-vnet
|
||||
router, err := vnet.NewRouter(&vnet.RouterConfig{
|
||||
CIDR: "0.0.0.0/0",
|
||||
LoggerFactory: logging.NewDefaultLoggerFactory(),
|
||||
})
|
||||
if err != nil {
|
||||
// handle error
|
||||
}
|
||||
|
||||
// Create a network and add to router, for example, for client.
|
||||
clientNetwork = vnet.NewNet(&vnet.NetConfig{
|
||||
StaticIP: "10.0.0.11",
|
||||
})
|
||||
if err = router.AddNet(clientNetwork); err != nil {
|
||||
// handle error
|
||||
}
|
||||
|
||||
// Start the router.
|
||||
if err = router.Start(); err != nil {
|
||||
// handle error
|
||||
}
|
||||
defer router.Stop() // nolint:errcheck
|
||||
|
||||
// Create a proxy, bind to the router.
|
||||
proxy, err := vnet_proxy.NewProxy(router)
|
||||
if err != nil {
|
||||
// handle error
|
||||
}
|
||||
defer proxy.Close() // nolint:errcheck
|
||||
|
||||
// Start to proxy some addresses, clientNetwork is a hit for proxy,
|
||||
// that the client in vnet is from this network.
|
||||
if err := proxy.Proxy(clientNetwork, serverAddr0); err != nil {
|
||||
// handle error
|
||||
}
|
||||
if err := proxy.Proxy(clientNetwork, serverAddr1); err != nil {
|
||||
// handle error
|
||||
}
|
||||
if err := proxy.Proxy(clientNetwork, serverAddr2); err != nil {
|
||||
// handle error
|
||||
}
|
||||
}
|
||||
|
||||
// Now, all packets from client, will be proxy to real server, vice versa.
|
||||
client0, err := clientNetwork.ListenPacket("udp4", "10.0.0.11:5787")
|
||||
if err != nil {
|
||||
// handle error
|
||||
}
|
||||
_, _ = client0.WriteTo([]byte("Hello"), serverAddr0)
|
||||
|
||||
client1, err := clientNetwork.ListenPacket("udp4", "10.0.0.11:5788")
|
||||
if err != nil {
|
||||
// handle error
|
||||
}
|
||||
_, _ = client1.WriteTo([]byte("Hello"), serverAddr1)
|
||||
|
||||
client2, err := clientNetwork.ListenPacket("udp4", "10.0.0.11:5789")
|
||||
if err != nil {
|
||||
// handle error
|
||||
}
|
||||
_, _ = client2.WriteTo([]byte("Hello"), serverAddr2)
|
||||
}
|
||||
@ -0,0 +1,222 @@
|
||||
// The MIT License (MIT)
|
||||
//
|
||||
// Copyright (c) 2021 srs-bench(ossrs)
|
||||
//
|
||||
// Permission is hereby granted, free of charge, to any person obtaining a copy of
|
||||
// this software and associated documentation files (the "Software"), to deal in
|
||||
// the Software without restriction, including without limitation the rights to
|
||||
// use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
|
||||
// the Software, and to permit persons to whom the Software is furnished to do so,
|
||||
// subject to the following conditions:
|
||||
//
|
||||
// The above copyright notice and this permission notice shall be included in all
|
||||
// copies or substantial portions of the Software.
|
||||
//
|
||||
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
|
||||
// FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
|
||||
// COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
|
||||
// IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
|
||||
// CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
package vnet
|
||||
|
||||
import (
|
||||
"net"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/pion/transport/vnet"
|
||||
)
|
||||
|
||||
// A UDP proxy between real server(net.UDPConn) and vnet.UDPConn.
|
||||
//
|
||||
// High level design:
|
||||
// ..............................................
|
||||
// : Virtual Network (vnet) :
|
||||
// : :
|
||||
// +-------+ * 1 +----+ +--------+ :
|
||||
// | :App |------------>|:Net|--o<-----|:Router | .............................
|
||||
// +-------+ +----+ | | : UDPProxy :
|
||||
// : | | +----+ +---------+ +---------+ +--------+
|
||||
// : | |--->o--|:Net|-->o-| vnet. |-->o-| net. |--->-| :Real |
|
||||
// : | | +----+ | UDPConn | | UDPConn | | Server |
|
||||
// : | | : +---------+ +---------+ +--------+
|
||||
// : | | ............................:
|
||||
// : +--------+ :
|
||||
// ...............................................
|
||||
//
|
||||
// The whole big picture:
|
||||
// ......................................
|
||||
// : Virtual Network (vnet) :
|
||||
// : :
|
||||
// +-------+ * 1 +----+ +--------+ :
|
||||
// | :App |------------>|:Net|--o<-----|:Router | .............................
|
||||
// +-------+ +----+ | | : UDPProxy :
|
||||
// +-----------+ * 1 +----+ | | +----+ +---------+ +---------+ +--------+
|
||||
// |:STUNServer|-------->|:Net|--o<-----| |--->o--|:Net|-->o-| vnet. |-->o-| net. |--->-| :Real |
|
||||
// +-----------+ +----+ | | +----+ | UDPConn | | UDPConn | | Server |
|
||||
// +-----------+ * 1 +----+ | | : +---------+ +---------+ +--------+
|
||||
// |:TURNServer|-------->|:Net|--o<-----| | ............................:
|
||||
// +-----------+ +----+ [1] | | :
|
||||
// : 1 | | 1 <<has>> :
|
||||
// : +---<>| |<>----+ [2] :
|
||||
// : | +--------+ | :
|
||||
// To form | *| v 0..1 :
|
||||
// a subnet tree | o [3] +-----+ :
|
||||
// : | ^ |:NAT | :
|
||||
// : | | +-----+ :
|
||||
// : +-------+ :
|
||||
// ......................................
|
||||
type UDPProxy struct {
|
||||
// The router bind to.
|
||||
router *vnet.Router
|
||||
|
||||
// Each vnet source, bind to a real socket to server.
|
||||
// key is real server addr, which is net.Addr
|
||||
// value is *aUDPProxyWorker
|
||||
workers sync.Map
|
||||
|
||||
// For each endpoint, we never know when to start and stop proxy,
|
||||
// so we stop the endpoint when timeout.
|
||||
timeout time.Duration
|
||||
|
||||
// For utest, to mock the target real server.
|
||||
// Optional, use the address of received client packet.
|
||||
mockRealServerAddr *net.UDPAddr
|
||||
}
|
||||
|
||||
// NewProxy create a proxy, the router for this proxy belongs/bind to. If need to proxy for
|
||||
// please create a new proxy for each router. For all addresses we proxy, we will create a
|
||||
// vnet.Net in this router and proxy all packets.
|
||||
func NewProxy(router *vnet.Router) (*UDPProxy, error) {
|
||||
v := &UDPProxy{router: router, timeout: 2 * time.Minute}
|
||||
return v, nil
|
||||
}
|
||||
|
||||
// Close the proxy, stop all workers.
|
||||
func (v *UDPProxy) Close() error {
|
||||
// nolint:godox // TODO: FIXME: Do cleanup.
|
||||
return nil
|
||||
}
|
||||
|
||||
// Proxy starts a worker for server, ignore if already started.
|
||||
func (v *UDPProxy) Proxy(client *vnet.Net, server *net.UDPAddr) error {
|
||||
// Note that even if the worker exists, it's also ok to create a same worker,
|
||||
// because the router will use the last one, and the real server will see a address
|
||||
// change event after we switch to the next worker.
|
||||
if _, ok := v.workers.Load(server.String()); ok {
|
||||
// nolint:godox // TODO: Need to restart the stopped worker?
|
||||
return nil
|
||||
}
|
||||
|
||||
// Not exists, create a new one.
|
||||
worker := &aUDPProxyWorker{
|
||||
router: v.router, mockRealServerAddr: v.mockRealServerAddr,
|
||||
}
|
||||
v.workers.Store(server.String(), worker)
|
||||
|
||||
return worker.Proxy(client, server)
|
||||
}
|
||||
|
||||
// A proxy worker for a specified proxy server.
|
||||
type aUDPProxyWorker struct {
|
||||
router *vnet.Router
|
||||
mockRealServerAddr *net.UDPAddr
|
||||
|
||||
// Each vnet source, bind to a real socket to server.
|
||||
// key is vnet client addr, which is net.Addr
|
||||
// value is *net.UDPConn
|
||||
endpoints sync.Map
|
||||
}
|
||||
|
||||
func (v *aUDPProxyWorker) Proxy(client *vnet.Net, serverAddr *net.UDPAddr) error { // nolint:gocognit
|
||||
// Create vnet for real server by serverAddr.
|
||||
nw := vnet.NewNet(&vnet.NetConfig{
|
||||
StaticIP: serverAddr.IP.String(),
|
||||
})
|
||||
if err := v.router.AddNet(nw); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// We must create a "same" vnet.UDPConn as the net.UDPConn,
|
||||
// which has the same ip:port, to copy packets between them.
|
||||
vnetSocket, err := nw.ListenUDP("udp4", serverAddr)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Start a proxy goroutine.
|
||||
var findEndpointBy func(addr net.Addr) (*net.UDPConn, error)
|
||||
// nolint:godox // TODO: FIXME: Do cleanup.
|
||||
go func() {
|
||||
buf := make([]byte, 1500)
|
||||
|
||||
for {
|
||||
n, addr, err := vnetSocket.ReadFrom(buf)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
if n <= 0 || addr == nil {
|
||||
continue // Drop packet
|
||||
}
|
||||
|
||||
realSocket, err := findEndpointBy(addr)
|
||||
if err != nil {
|
||||
continue // Drop packet.
|
||||
}
|
||||
|
||||
if _, err := realSocket.Write(buf[:n]); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
// Got new vnet client, start a new endpoint.
|
||||
findEndpointBy = func(addr net.Addr) (*net.UDPConn, error) {
|
||||
// Exists binding.
|
||||
if value, ok := v.endpoints.Load(addr.String()); ok {
|
||||
// Exists endpoint, reuse it.
|
||||
return value.(*net.UDPConn), nil
|
||||
}
|
||||
|
||||
// The real server we proxy to, for utest to mock it.
|
||||
realAddr := serverAddr
|
||||
if v.mockRealServerAddr != nil {
|
||||
realAddr = v.mockRealServerAddr
|
||||
}
|
||||
|
||||
// Got new vnet client, create new endpoint.
|
||||
realSocket, err := net.DialUDP("udp4", nil, realAddr)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Bind address.
|
||||
v.endpoints.Store(addr.String(), realSocket)
|
||||
|
||||
// Got packet from real serverAddr, we should proxy it to vnet.
|
||||
// nolint:godox // TODO: FIXME: Do cleanup.
|
||||
go func(vnetClientAddr net.Addr) {
|
||||
buf := make([]byte, 1500)
|
||||
for {
|
||||
n, _, err := realSocket.ReadFrom(buf)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
if n <= 0 {
|
||||
continue // Drop packet
|
||||
}
|
||||
|
||||
if _, err := vnetSocket.WriteTo(buf[:n], vnetClientAddr); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}(addr)
|
||||
|
||||
return realSocket, nil
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@ -0,0 +1,61 @@
|
||||
// The MIT License (MIT)
|
||||
//
|
||||
// Copyright (c) 2021 srs-bench(ossrs)
|
||||
//
|
||||
// Permission is hereby granted, free of charge, to any person obtaining a copy of
|
||||
// this software and associated documentation files (the "Software"), to deal in
|
||||
// the Software without restriction, including without limitation the rights to
|
||||
// use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
|
||||
// the Software, and to permit persons to whom the Software is furnished to do so,
|
||||
// subject to the following conditions:
|
||||
//
|
||||
// The above copyright notice and this permission notice shall be included in all
|
||||
// copies or substantial portions of the Software.
|
||||
//
|
||||
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
|
||||
// FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
|
||||
// COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
|
||||
// IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
|
||||
// CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
package vnet
|
||||
|
||||
import (
|
||||
"net"
|
||||
)
|
||||
|
||||
func (v *UDPProxy) Deliver(sourceAddr, destAddr net.Addr, b []byte) (nn int, err error) {
|
||||
v.workers.Range(func(key, value interface{}) bool {
|
||||
if nn, err := value.(*aUDPProxyWorker).Deliver(sourceAddr, destAddr, b); err != nil {
|
||||
return false // Fail, abort.
|
||||
} else if nn == len(b) {
|
||||
return false // Done.
|
||||
}
|
||||
|
||||
return true // Deliver by next worker.
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
func (v *aUDPProxyWorker) Deliver(sourceAddr, destAddr net.Addr, b []byte) (nn int, err error) {
|
||||
addr, ok := sourceAddr.(*net.UDPAddr)
|
||||
if !ok {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
// TODO: Support deliver packet from real server to vnet.
|
||||
// If packet is from vent, proxy to real server.
|
||||
var realSocket *net.UDPConn
|
||||
if value, ok := v.endpoints.Load(addr.String()); !ok {
|
||||
return 0, nil
|
||||
} else {
|
||||
realSocket = value.(*net.UDPConn)
|
||||
}
|
||||
|
||||
// Send to real server.
|
||||
if _, err := realSocket.Write(b); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
return len(b), nil
|
||||
}
|
||||
@ -0,0 +1,184 @@
|
||||
// The MIT License (MIT)
|
||||
//
|
||||
// Copyright (c) 2021 srs-bench(ossrs)
|
||||
//
|
||||
// Permission is hereby granted, free of charge, to any person obtaining a copy of
|
||||
// this software and associated documentation files (the "Software"), to deal in
|
||||
// the Software without restriction, including without limitation the rights to
|
||||
// use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
|
||||
// the Software, and to permit persons to whom the Software is furnished to do so,
|
||||
// subject to the following conditions:
|
||||
//
|
||||
// The above copyright notice and this permission notice shall be included in all
|
||||
// copies or substantial portions of the Software.
|
||||
//
|
||||
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
|
||||
// FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
|
||||
// COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
|
||||
// IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
|
||||
// CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
package vnet
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"github.com/pion/logging"
|
||||
"github.com/pion/transport/vnet"
|
||||
"net"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// vnet client:
|
||||
// 10.0.0.11:5787
|
||||
// proxy to real server:
|
||||
// 192.168.1.10:8000
|
||||
func TestUDPProxyDirectDeliver(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
var r0, r1, r2 error
|
||||
defer func() {
|
||||
if r0 != nil || r1 != nil || r2 != nil {
|
||||
t.Errorf("fail for ctx=%v, r0=%v, r1=%v, r2=%v", ctx.Err(), r0, r1, r2)
|
||||
}
|
||||
}()
|
||||
|
||||
var wg sync.WaitGroup
|
||||
defer wg.Wait()
|
||||
|
||||
// Timeout, fail
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer cancel()
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
case <-time.After(time.Duration(*testTimeout) * time.Millisecond):
|
||||
r2 = fmt.Errorf("timeout")
|
||||
}
|
||||
}()
|
||||
|
||||
// For utest, we always proxy vnet packets to the random port we listen to.
|
||||
mockServer := NewMockUDPEchoServer()
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer cancel()
|
||||
if err := mockServer.doMockUDPServer(ctx); err != nil {
|
||||
r0 = err
|
||||
}
|
||||
}()
|
||||
|
||||
// Create a vent and proxy.
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer cancel()
|
||||
|
||||
// When real server is ready, start the vnet test.
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-mockServer.realServerReady.Done():
|
||||
}
|
||||
|
||||
doVnetProxy := func() error {
|
||||
router, err := vnet.NewRouter(&vnet.RouterConfig{
|
||||
CIDR: "0.0.0.0/0",
|
||||
LoggerFactory: logging.NewDefaultLoggerFactory(),
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
clientNetwork := vnet.NewNet(&vnet.NetConfig{
|
||||
StaticIP: "10.0.0.11",
|
||||
})
|
||||
if err = router.AddNet(clientNetwork); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := router.Start(); err != nil {
|
||||
return err
|
||||
}
|
||||
defer router.Stop()
|
||||
|
||||
proxy, err := NewProxy(router)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer proxy.Close()
|
||||
|
||||
// For utest, mock the target real server.
|
||||
proxy.mockRealServerAddr = mockServer.realServerAddr
|
||||
|
||||
// The real server address to proxy to.
|
||||
// Note that for utest, we will proxy to a local address.
|
||||
serverAddr, err := net.ResolveUDPAddr("udp4", "192.168.1.10:8000")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := proxy.Proxy(clientNetwork, serverAddr); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Now, all packets from client, will be proxy to real server, vice versa.
|
||||
client, err := clientNetwork.ListenPacket("udp4", "10.0.0.11:5787")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// When system quit, interrupt client.
|
||||
selfKill, selfKillCancel := context.WithCancel(context.Background())
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
selfKillCancel()
|
||||
client.Close()
|
||||
}()
|
||||
|
||||
// Write by vnet client.
|
||||
if _, err := client.WriteTo([]byte("Hello"), serverAddr); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
buf := make([]byte, 1500)
|
||||
if n, addr, err := client.ReadFrom(buf); err != nil {
|
||||
if selfKill.Err() == context.Canceled {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
} else if n != 5 || addr == nil {
|
||||
return fmt.Errorf("n=%v, addr=%v", n, addr)
|
||||
} else if string(buf[:n]) != "Hello" {
|
||||
return fmt.Errorf("data %v", buf[:n])
|
||||
}
|
||||
|
||||
// Directly write, simulate the ARQ packet.
|
||||
// We should got the echo packet also.
|
||||
if _, err := proxy.Deliver(client.LocalAddr(), serverAddr, []byte("Hello")); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if n, addr, err := client.ReadFrom(buf); err != nil {
|
||||
if selfKill.Err() == context.Canceled {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
} else if n != 5 || addr == nil {
|
||||
return fmt.Errorf("n=%v, addr=%v", n, addr)
|
||||
} else if string(buf[:n]) != "Hello" {
|
||||
return fmt.Errorf("data %v", buf[:n])
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
if err := doVnetProxy(); err != nil {
|
||||
r1 = err
|
||||
}
|
||||
}()
|
||||
}
|
||||
@ -0,0 +1,615 @@
|
||||
// The MIT License (MIT)
|
||||
//
|
||||
// Copyright (c) 2021 srs-bench(ossrs)
|
||||
//
|
||||
// Permission is hereby granted, free of charge, to any person obtaining a copy of
|
||||
// this software and associated documentation files (the "Software"), to deal in
|
||||
// the Software without restriction, including without limitation the rights to
|
||||
// use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
|
||||
// the Software, and to permit persons to whom the Software is furnished to do so,
|
||||
// subject to the following conditions:
|
||||
//
|
||||
// The above copyright notice and this permission notice shall be included in all
|
||||
// copies or substantial portions of the Software.
|
||||
//
|
||||
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
|
||||
// FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
|
||||
// COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
|
||||
// IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
|
||||
// CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
package vnet
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"flag"
|
||||
"fmt"
|
||||
"net"
|
||||
"os"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/pion/logging"
|
||||
"github.com/pion/transport/vnet"
|
||||
)
|
||||
|
||||
type MockUDPEchoServer struct {
|
||||
realServerAddr *net.UDPAddr
|
||||
realServerReady context.Context
|
||||
realServerReadyCancel context.CancelFunc
|
||||
}
|
||||
|
||||
func NewMockUDPEchoServer() *MockUDPEchoServer {
|
||||
v := &MockUDPEchoServer{}
|
||||
v.realServerReady, v.realServerReadyCancel = context.WithCancel(context.Background())
|
||||
return v
|
||||
}
|
||||
|
||||
func (v *MockUDPEchoServer) doMockUDPServer(ctx context.Context) error {
|
||||
// Listen to a random port.
|
||||
laddr, err := net.ResolveUDPAddr("udp4", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
conn, err := net.ListenUDP("udp4", laddr)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
v.realServerAddr = conn.LocalAddr().(*net.UDPAddr)
|
||||
v.realServerReadyCancel()
|
||||
|
||||
// When system quit, interrupt client.
|
||||
selfKill, selfKillCancel := context.WithCancel(context.Background())
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
selfKillCancel()
|
||||
_ = conn.Close()
|
||||
}()
|
||||
|
||||
// Note that if they has the same ID, the address should not changed.
|
||||
addrs := make(map[string]net.Addr)
|
||||
|
||||
// Start an echo UDP server.
|
||||
buf := make([]byte, 1500)
|
||||
for ctx.Err() == nil {
|
||||
n, addr, err := conn.ReadFrom(buf)
|
||||
if err != nil {
|
||||
if errors.Is(selfKill.Err(), context.Canceled) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
} else if n == 0 || addr == nil {
|
||||
return fmt.Errorf("n=%v, addr=%v", n, addr) // nolint:goerr113
|
||||
} else if nn, err := conn.WriteTo(buf[:n], addr); err != nil {
|
||||
return err
|
||||
} else if nn != n {
|
||||
return fmt.Errorf("nn=%v, n=%v", nn, n) // nolint:goerr113
|
||||
}
|
||||
|
||||
// Check the address, shold not change, use content as ID.
|
||||
clientID := string(buf[:n])
|
||||
if oldAddr, ok := addrs[clientID]; ok && oldAddr.String() != addr.String() {
|
||||
return fmt.Errorf("address change %v to %v", oldAddr.String(), addr.String()) // nolint:goerr113
|
||||
}
|
||||
addrs[clientID] = addr
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
var testTimeout = flag.Int("timeout", 5000, "For each case, the timeout in ms") // nolint:gochecknoglobals
|
||||
|
||||
func TestMain(m *testing.M) {
|
||||
flag.Parse()
|
||||
os.Exit(m.Run())
|
||||
}
|
||||
|
||||
// vnet client:
|
||||
// 10.0.0.11:5787
|
||||
// proxy to real server:
|
||||
// 192.168.1.10:8000
|
||||
func TestUDPProxyOne2One(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
var r0, r1, r2 error
|
||||
defer func() {
|
||||
if r0 != nil || r1 != nil || r2 != nil {
|
||||
t.Errorf("fail for ctx=%v, r0=%v, r1=%v, r2=%v", ctx.Err(), r0, r1, r2)
|
||||
}
|
||||
}()
|
||||
|
||||
var wg sync.WaitGroup
|
||||
defer wg.Wait()
|
||||
|
||||
// Timeout, fail
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer cancel()
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
case <-time.After(time.Duration(*testTimeout) * time.Millisecond):
|
||||
r2 = fmt.Errorf("timeout") // nolint:goerr113
|
||||
}
|
||||
}()
|
||||
|
||||
// For utest, we always proxy vnet packets to the random port we listen to.
|
||||
mockServer := NewMockUDPEchoServer()
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer cancel()
|
||||
if err := mockServer.doMockUDPServer(ctx); err != nil {
|
||||
r0 = err
|
||||
}
|
||||
}()
|
||||
|
||||
// Create a vent and proxy.
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer cancel()
|
||||
|
||||
// When real server is ready, start the vnet test.
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-mockServer.realServerReady.Done():
|
||||
}
|
||||
|
||||
doVnetProxy := func() error {
|
||||
router, err := vnet.NewRouter(&vnet.RouterConfig{
|
||||
CIDR: "0.0.0.0/0",
|
||||
LoggerFactory: logging.NewDefaultLoggerFactory(),
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
clientNetwork := vnet.NewNet(&vnet.NetConfig{
|
||||
StaticIP: "10.0.0.11",
|
||||
})
|
||||
if err = router.AddNet(clientNetwork); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err = router.Start(); err != nil {
|
||||
return err
|
||||
}
|
||||
defer router.Stop() // nolint:errcheck
|
||||
|
||||
proxy, err := NewProxy(router)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer proxy.Close() // nolint:errcheck
|
||||
|
||||
// For utest, mock the target real server.
|
||||
proxy.mockRealServerAddr = mockServer.realServerAddr
|
||||
|
||||
// The real server address to proxy to.
|
||||
// Note that for utest, we will proxy to a local address.
|
||||
serverAddr, err := net.ResolveUDPAddr("udp4", "192.168.1.10:8000")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err = proxy.Proxy(clientNetwork, serverAddr); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Now, all packets from client, will be proxy to real server, vice versa.
|
||||
client, err := clientNetwork.ListenPacket("udp4", "10.0.0.11:5787")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// When system quit, interrupt client.
|
||||
selfKill, selfKillCancel := context.WithCancel(context.Background())
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
selfKillCancel()
|
||||
_ = client.Close() // nolint:errcheck
|
||||
}()
|
||||
|
||||
for i := 0; i < 10; i++ {
|
||||
if _, err = client.WriteTo([]byte("Hello"), serverAddr); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var n int
|
||||
var addr net.Addr
|
||||
buf := make([]byte, 1500)
|
||||
if n, addr, err = client.ReadFrom(buf); err != nil { // nolint:gocritic
|
||||
if errors.Is(selfKill.Err(), context.Canceled) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
} else if n != 5 || addr == nil {
|
||||
return fmt.Errorf("n=%v, addr=%v", n, addr) // nolint:goerr113
|
||||
} else if string(buf[:n]) != "Hello" {
|
||||
return fmt.Errorf("data %v", buf[:n]) // nolint:goerr113
|
||||
}
|
||||
|
||||
// Wait for awhile for each UDP packet, to simulate real network.
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil
|
||||
case <-time.After(30 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
if err := doVnetProxy(); err != nil {
|
||||
r1 = err
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// vnet client:
|
||||
// 10.0.0.11:5787
|
||||
// 10.0.0.11:5788
|
||||
// proxy to real server:
|
||||
// 192.168.1.10:8000
|
||||
func TestUDPProxyTwo2One(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
var r0, r1, r2, r3 error
|
||||
defer func() {
|
||||
if r0 != nil || r1 != nil || r2 != nil || r3 != nil {
|
||||
t.Errorf("fail for ctx=%v, r0=%v, r1=%v, r2=%v, r3=%v", ctx.Err(), r0, r1, r2, r3)
|
||||
}
|
||||
}()
|
||||
|
||||
var wg sync.WaitGroup
|
||||
defer wg.Wait()
|
||||
|
||||
// Timeout, fail
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer cancel()
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
case <-time.After(time.Duration(*testTimeout) * time.Millisecond):
|
||||
r2 = fmt.Errorf("timeout") // nolint:goerr113
|
||||
}
|
||||
}()
|
||||
|
||||
// For utest, we always proxy vnet packets to the random port we listen to.
|
||||
mockServer := NewMockUDPEchoServer()
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer cancel()
|
||||
if err := mockServer.doMockUDPServer(ctx); err != nil {
|
||||
r0 = err
|
||||
}
|
||||
}()
|
||||
|
||||
// Create a vent and proxy.
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer cancel()
|
||||
|
||||
// When real server is ready, start the vnet test.
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-mockServer.realServerReady.Done():
|
||||
}
|
||||
|
||||
doVnetProxy := func() error {
|
||||
router, err := vnet.NewRouter(&vnet.RouterConfig{
|
||||
CIDR: "0.0.0.0/0",
|
||||
LoggerFactory: logging.NewDefaultLoggerFactory(),
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
clientNetwork := vnet.NewNet(&vnet.NetConfig{
|
||||
StaticIP: "10.0.0.11",
|
||||
})
|
||||
if err = router.AddNet(clientNetwork); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err = router.Start(); err != nil {
|
||||
return err
|
||||
}
|
||||
defer router.Stop() // nolint:errcheck
|
||||
|
||||
proxy, err := NewProxy(router)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer proxy.Close() // nolint:errcheck
|
||||
|
||||
// For utest, mock the target real server.
|
||||
proxy.mockRealServerAddr = mockServer.realServerAddr
|
||||
|
||||
// The real server address to proxy to.
|
||||
// Note that for utest, we will proxy to a local address.
|
||||
serverAddr, err := net.ResolveUDPAddr("udp4", "192.168.1.10:8000")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err = proxy.Proxy(clientNetwork, serverAddr); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
handClient := func(address, echoData string) error {
|
||||
// Now, all packets from client, will be proxy to real server, vice versa.
|
||||
client, err := clientNetwork.ListenPacket("udp4", address) // nolint:govet
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// When system quit, interrupt client.
|
||||
selfKill, selfKillCancel := context.WithCancel(context.Background())
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
selfKillCancel()
|
||||
_ = client.Close()
|
||||
}()
|
||||
|
||||
for i := 0; i < 10; i++ {
|
||||
if _, err := client.WriteTo([]byte(echoData), serverAddr); err != nil { // nolint:govet
|
||||
return err
|
||||
}
|
||||
|
||||
var n int
|
||||
var addr net.Addr
|
||||
buf := make([]byte, 1400)
|
||||
if n, addr, err = client.ReadFrom(buf); err != nil { // nolint:gocritic
|
||||
if errors.Is(selfKill.Err(), context.Canceled) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
} else if n != len(echoData) || addr == nil {
|
||||
return fmt.Errorf("n=%v, addr=%v", n, addr) // nolint:goerr113
|
||||
} else if string(buf[:n]) != echoData {
|
||||
return fmt.Errorf("check data %v", buf[:n]) // nolint:goerr113
|
||||
}
|
||||
|
||||
// Wait for awhile for each UDP packet, to simulate real network.
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil
|
||||
case <-time.After(30 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
client0, client0Cancel := context.WithCancel(context.Background())
|
||||
go func() {
|
||||
defer client0Cancel()
|
||||
address := "10.0.0.11:5787"
|
||||
if err := handClient(address, "Hello"); err != nil { // nolint:govet
|
||||
r3 = fmt.Errorf("client %v err %v", address, err) // nolint:goerr113
|
||||
}
|
||||
}()
|
||||
|
||||
client1, client1Cancel := context.WithCancel(context.Background())
|
||||
go func() {
|
||||
defer client1Cancel()
|
||||
address := "10.0.0.11:5788"
|
||||
if err := handClient(address, "World"); err != nil { // nolint:govet
|
||||
r3 = fmt.Errorf("client %v err %v", address, err) // nolint:goerr113
|
||||
}
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
case <-client0.Done():
|
||||
case <-client1.Done():
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
if err := doVnetProxy(); err != nil {
|
||||
r1 = err
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// vnet client:
|
||||
// 10.0.0.11:5787
|
||||
// proxy to real server:
|
||||
// 192.168.1.10:8000
|
||||
//
|
||||
// vnet client:
|
||||
// 10.0.0.11:5788
|
||||
// proxy to real server:
|
||||
// 192.168.1.10:8000
|
||||
func TestUDPProxyProxyTwice(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
var r0, r1, r2, r3 error
|
||||
defer func() {
|
||||
if r0 != nil || r1 != nil || r2 != nil || r3 != nil {
|
||||
t.Errorf("fail for ctx=%v, r0=%v, r1=%v, r2=%v, r3=%v", ctx.Err(), r0, r1, r2, r3)
|
||||
}
|
||||
}()
|
||||
|
||||
var wg sync.WaitGroup
|
||||
defer wg.Wait()
|
||||
|
||||
// Timeout, fail
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer cancel()
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
case <-time.After(time.Duration(*testTimeout) * time.Millisecond):
|
||||
r2 = fmt.Errorf("timeout") // nolint:goerr113
|
||||
}
|
||||
}()
|
||||
|
||||
// For utest, we always proxy vnet packets to the random port we listen to.
|
||||
mockServer := NewMockUDPEchoServer()
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer cancel()
|
||||
if err := mockServer.doMockUDPServer(ctx); err != nil {
|
||||
r0 = err
|
||||
}
|
||||
}()
|
||||
|
||||
// Create a vent and proxy.
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer cancel()
|
||||
|
||||
// When real server is ready, start the vnet test.
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-mockServer.realServerReady.Done():
|
||||
}
|
||||
|
||||
doVnetProxy := func() error {
|
||||
router, err := vnet.NewRouter(&vnet.RouterConfig{
|
||||
CIDR: "0.0.0.0/0",
|
||||
LoggerFactory: logging.NewDefaultLoggerFactory(),
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
clientNetwork := vnet.NewNet(&vnet.NetConfig{
|
||||
StaticIP: "10.0.0.11",
|
||||
})
|
||||
if err = router.AddNet(clientNetwork); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err = router.Start(); err != nil {
|
||||
return err
|
||||
}
|
||||
defer router.Stop() // nolint:errcheck
|
||||
|
||||
proxy, err := NewProxy(router)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer proxy.Close() // nolint:errcheck
|
||||
|
||||
// For utest, mock the target real server.
|
||||
proxy.mockRealServerAddr = mockServer.realServerAddr
|
||||
|
||||
// The real server address to proxy to.
|
||||
// Note that for utest, we will proxy to a local address.
|
||||
serverAddr, err := net.ResolveUDPAddr("udp4", "192.168.1.10:8000")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
handClient := func(address, echoData string) error {
|
||||
// We proxy multiple times, for example, in publisher and player, both call
|
||||
// the proxy when got answer.
|
||||
if err := proxy.Proxy(clientNetwork, serverAddr); err != nil { // nolint:govet
|
||||
return err
|
||||
}
|
||||
|
||||
// Now, all packets from client, will be proxy to real server, vice versa.
|
||||
client, err := clientNetwork.ListenPacket("udp4", address) // nolint:govet
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// When system quit, interrupt client.
|
||||
selfKill, selfKillCancel := context.WithCancel(context.Background())
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
selfKillCancel()
|
||||
_ = client.Close() // nolint:errcheck
|
||||
}()
|
||||
|
||||
for i := 0; i < 10; i++ {
|
||||
if _, err = client.WriteTo([]byte(echoData), serverAddr); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
buf := make([]byte, 1500)
|
||||
if n, addr, err := client.ReadFrom(buf); err != nil { // nolint:gocritic,govet
|
||||
if errors.Is(selfKill.Err(), context.Canceled) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
} else if n != len(echoData) || addr == nil {
|
||||
return fmt.Errorf("n=%v, addr=%v", n, addr) // nolint:goerr113
|
||||
} else if string(buf[:n]) != echoData {
|
||||
return fmt.Errorf("verify data %v", buf[:n]) // nolint:goerr113
|
||||
}
|
||||
|
||||
// Wait for awhile for each UDP packet, to simulate real network.
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil
|
||||
case <-time.After(30 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
client0, client0Cancel := context.WithCancel(context.Background())
|
||||
go func() {
|
||||
defer client0Cancel()
|
||||
address := "10.0.0.11:5787"
|
||||
if err = handClient(address, "Hello"); err != nil {
|
||||
r3 = fmt.Errorf("client %v err %v", address, err) // nolint:goerr113
|
||||
}
|
||||
}()
|
||||
|
||||
client1, client1Cancel := context.WithCancel(context.Background())
|
||||
go func() {
|
||||
defer client1Cancel()
|
||||
|
||||
// Slower than client0, 60ms.
|
||||
// To simulate the real player or publisher, might not start at the same time.
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(150 * time.Millisecond):
|
||||
}
|
||||
|
||||
address := "10.0.0.11:5788"
|
||||
if err = handClient(address, "World"); err != nil {
|
||||
r3 = fmt.Errorf("client %v err %v", address, err) // nolint:goerr113
|
||||
}
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
case <-client0.Done():
|
||||
case <-client1.Done():
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
if err := doVnetProxy(); err != nil {
|
||||
r1 = err
|
||||
}
|
||||
}()
|
||||
}
|
||||
Loading…
Reference in New Issue