package/gortsplib
1
0
Fork 0
mirror of https://github.com/aler9/gortsplib synced 2025-10-13 02:43:53 +08:00
Code Issues Packages Projects Releases Wiki Activity

server: support reading with multicast

Browse Source
This commit is contained in:
aler9
2021-06-14 19:39:50 +02:00
committed by Alessandro Ros
parent f80faf5ac5
commit 3f3226b53d
16 changed files with 749 additions and 275 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@@ -30,7 +30,7 @@ Features:
* Handle requests from clients
* Sessions and connections are independent
* Read streams from clients with UDP, TCP or TLS/RTSPS
* Write streams to clients with UDP, TCP or TLS/RTSPS
* Write streams to clients with UDP, UDP-multicast, TCP or TLS/RTSPS
* Generate RTCP receiver reports automatically
* Utilities
* Encode and decode RTSP primitives, RTP/H264, RTP/AAC, SDP

44
examples/server-tls/main.go
View File

@@ -17,9 +17,8 @@ import (
type serverHandler struct {
mutex sync.Mutex
stream *gortsplib.ServerStream
publisher *gortsplib.ServerSession
readers map[*gortsplib.ServerSession]struct{}
sdp []byte
}
// called after a connection is opened.
@@ -44,23 +43,22 @@ func (sh *serverHandler) OnSessionClose(ctx *gortsplib.ServerHandlerOnSessionClo
sh.mutex.Lock()
defer sh.mutex.Unlock()
if ctx.Session == sh.publisher {
sh.publisher = nil
sh.sdp = nil
} else {
delete(sh.readers, ctx.Session)
// close stream-related listeners and disconnect every reader
if sh.stream != nil && ctx.Session == sh.publisher {
sh.stream.Close()
sh.stream = nil
}
}
// called after receiving a DESCRIBE request.
func (sh *serverHandler) OnDescribe(ctx *gortsplib.ServerHandlerOnDescribeCtx) (*base.Response, []byte, error) {
func (sh *serverHandler) OnDescribe(ctx *gortsplib.ServerHandlerOnDescribeCtx) (*base.Response, *gortsplib.ServerStream, error) {
log.Printf("describe request")
sh.mutex.Lock()
defer sh.mutex.Unlock()
// no one is publishing yet
if sh.publisher == nil {
if sh.stream == nil {
return &base.Response{
StatusCode: base.StatusNotFound,
}, nil, nil
@@ -68,7 +66,7 @@ func (sh *serverHandler) OnDescribe(ctx *gortsplib.ServerHandlerOnDescribeCtx) (
return &base.Response{
StatusCode: base.StatusOK,
}, sh.sdp, nil
}, sh.stream, nil
}
// called after receiving an ANNOUNCE request.
@@ -78,14 +76,14 @@ func (sh *serverHandler) OnAnnounce(ctx *gortsplib.ServerHandlerOnAnnounceCtx) (
sh.mutex.Lock()
defer sh.mutex.Unlock()
if sh.publisher != nil {
if sh.stream != nil {
return &base.Response{
StatusCode: base.StatusBadRequest,
}, fmt.Errorf("someone is already publishing")
}
sh.stream = gortsplib.NewServerStream(ctx.Tracks)
sh.publisher = ctx.Session
sh.sdp = ctx.Tracks.Write()
return &base.Response{
StatusCode: base.StatusOK,
@@ -93,23 +91,25 @@ func (sh *serverHandler) OnAnnounce(ctx *gortsplib.ServerHandlerOnAnnounceCtx) (
}
// called after receiving a SETUP request.
func (sh *serverHandler) OnSetup(ctx *gortsplib.ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
func (sh *serverHandler) OnSetup(ctx *gortsplib.ServerHandlerOnSetupCtx) (*base.Response, *gortsplib.ServerStream, *uint32, error) {
log.Printf("setup request")
// no one is publishing yet
if sh.stream == nil {
return &base.Response{
StatusCode: base.StatusNotFound,
}, nil, nil, nil
}
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, sh.stream, nil, nil
}
// called after receiving a PLAY request.
func (sh *serverHandler) OnPlay(ctx *gortsplib.ServerHandlerOnPlayCtx) (*base.Response, error) {
log.Printf("play request")
sh.mutex.Lock()
defer sh.mutex.Unlock()
sh.readers[ctx.Session] = struct{}{}
return &base.Response{
StatusCode: base.StatusOK,
}, nil
@@ -131,9 +131,7 @@ func (sh *serverHandler) OnFrame(ctx *gortsplib.ServerHandlerOnFrameCtx) {
// if we are the publisher, route frames to readers
if ctx.Session == sh.publisher {
for r := range sh.readers {
r.WriteFrame(ctx.TrackID, ctx.StreamType, ctx.Payload)
}
sh.stream.WriteFrame(ctx.TrackID, ctx.StreamType, ctx.Payload)
}
}
@@ -148,7 +146,7 @@ func main() {
// configure server
s := &gortsplib.Server{
Handler: &serverHandler{readers: make(map[*gortsplib.ServerSession]struct{})},
Handler: &serverHandler{},
TLSConfig: &tls.Config{Certificates: []tls.Certificate{cert}},
}

44
examples/server/main.go
View File

@@ -16,9 +16,8 @@ import (
type serverHandler struct {
mutex sync.Mutex
stream *gortsplib.ServerStream
publisher *gortsplib.ServerSession
readers map[*gortsplib.ServerSession]struct{}
sdp []byte
}
// called after a connection is opened.
@@ -43,23 +42,22 @@ func (sh *serverHandler) OnSessionClose(ctx *gortsplib.ServerHandlerOnSessionClo
sh.mutex.Lock()
defer sh.mutex.Unlock()
if ctx.Session == sh.publisher {
sh.publisher = nil
sh.sdp = nil
} else {
delete(sh.readers, ctx.Session)
// close stream-related listeners and disconnect every reader
if sh.stream != nil && ctx.Session == sh.publisher {
sh.stream.Close()
sh.stream = nil
}
}
// called after receiving a DESCRIBE request.
func (sh *serverHandler) OnDescribe(ctx *gortsplib.ServerHandlerOnDescribeCtx) (*base.Response, []byte, error) {
func (sh *serverHandler) OnDescribe(ctx *gortsplib.ServerHandlerOnDescribeCtx) (*base.Response, *gortsplib.ServerStream, error) {
log.Printf("describe request")
sh.mutex.Lock()
defer sh.mutex.Unlock()
// no one is publishing yet
if sh.publisher == nil {
if sh.stream == nil {
return &base.Response{
StatusCode: base.StatusNotFound,
}, nil, nil
@@ -67,7 +65,7 @@ func (sh *serverHandler) OnDescribe(ctx *gortsplib.ServerHandlerOnDescribeCtx) (
return &base.Response{
StatusCode: base.StatusOK,
}, sh.sdp, nil
}, sh.stream, nil
}
// called after receiving an ANNOUNCE request.
@@ -77,14 +75,14 @@ func (sh *serverHandler) OnAnnounce(ctx *gortsplib.ServerHandlerOnAnnounceCtx) (
sh.mutex.Lock()
defer sh.mutex.Unlock()
if sh.publisher != nil {
if sh.stream != nil {
return &base.Response{
StatusCode: base.StatusBadRequest,
}, fmt.Errorf("someone is already publishing")
}
sh.stream = gortsplib.NewServerStream(ctx.Tracks)
sh.publisher = ctx.Session
sh.sdp = ctx.Tracks.Write()
return &base.Response{
StatusCode: base.StatusOK,
@@ -92,23 +90,25 @@ func (sh *serverHandler) OnAnnounce(ctx *gortsplib.ServerHandlerOnAnnounceCtx) (
}
// called after receiving a SETUP request.
func (sh *serverHandler) OnSetup(ctx *gortsplib.ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
func (sh *serverHandler) OnSetup(ctx *gortsplib.ServerHandlerOnSetupCtx) (*base.Response, *gortsplib.ServerStream, *uint32, error) {
log.Printf("setup request")
// no one is publishing yet
if sh.stream == nil {
return &base.Response{
StatusCode: base.StatusNotFound,
}, nil, nil, nil
}
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, sh.stream, nil, nil
}
// called after receiving a PLAY request.
func (sh *serverHandler) OnPlay(ctx *gortsplib.ServerHandlerOnPlayCtx) (*base.Response, error) {
log.Printf("play request")
sh.mutex.Lock()
defer sh.mutex.Unlock()
sh.readers[ctx.Session] = struct{}{}
return &base.Response{
StatusCode: base.StatusOK,
}, nil
@@ -130,16 +130,14 @@ func (sh *serverHandler) OnFrame(ctx *gortsplib.ServerHandlerOnFrameCtx) {
// if we are the publisher, route frames to readers
if ctx.Session == sh.publisher {
for r := range sh.readers {
r.WriteFrame(ctx.TrackID, ctx.StreamType, ctx.Payload)
}
sh.stream.WriteFrame(ctx.TrackID, ctx.StreamType, ctx.Payload)
}
}
func main() {
// configure server
s := &gortsplib.Server{
Handler: &serverHandler{readers: make(map[*gortsplib.ServerSession]struct{})},
Handler: &serverHandler{},
UDPRTPAddress: ":8000",
UDPRTCPAddress: ":8001",
}

1
go.mod
View File

@@ -8,4 +8,5 @@ require (
github.com/pion/rtp v1.6.1
github.com/pion/sdp/v3 v3.0.2
github.com/stretchr/testify v1.6.1
golang.org/x/net v0.0.0-20210610132358-84b48f89b13b
)

8
go.sum
View File

@@ -17,6 +17,14 @@ github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZN
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0=
github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
golang.org/x/net v0.0.0-20210610132358-84b48f89b13b h1:k+E048sYJHyVnsr1GDrRZWQ32D2C7lWs9JRc0bel53A=
golang.org/x/net v0.0.0-20210610132358-84b48f89b13b/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210423082822-04245dca01da h1:b3NXsE2LusjYGGjL5bxEVZZORm/YEFFrWFjR8eFrw/c=
golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=

24
pkg/headers/transport.go
View File

@@ -230,19 +230,31 @@ func (h Transport) Write() base.HeaderValue {
}
}
if h.Destination != nil {
rets = append(rets, "destination="+*h.Destination)
}
if h.InterleavedIDs != nil {
ports := *h.InterleavedIDs
rets = append(rets, "interleaved="+strconv.FormatInt(int64(ports[0]), 10)+"-"+strconv.FormatInt(int64(ports[1]), 10))
rets = append(rets, "interleaved="+strconv.FormatInt(int64(h.InterleavedIDs[0]), 10)+"-"+strconv.FormatInt(int64(h.InterleavedIDs[1]), 10))
}
if h.Ports != nil {
rets = append(rets, "port="+strconv.FormatInt(int64(h.Ports[0]), 10)+
"-"+strconv.FormatInt(int64(h.Ports[1]), 10))
}
if h.TTL != nil {
rets = append(rets, "ttl="+strconv.FormatUint(uint64(*h.TTL), 10))
}
if h.ClientPorts != nil {
ports := *h.ClientPorts
rets = append(rets, "client_port="+strconv.FormatInt(int64(ports[0]), 10)+"-"+strconv.FormatInt(int64(ports[1]), 10))
rets = append(rets, "client_port="+strconv.FormatInt(int64(h.ClientPorts[0]), 10)+
"-"+strconv.FormatInt(int64(h.ClientPorts[1]), 10))
}
if h.ServerPorts != nil {
ports := *h.ServerPorts
rets = append(rets, "server_port="+strconv.FormatInt(int64(ports[0]), 10)+"-"+strconv.FormatInt(int64(ports[1]), 10))
rets = append(rets, "server_port="+strconv.FormatInt(int64(h.ServerPorts[0]), 10)+
"-"+strconv.FormatInt(int64(h.ServerPorts[1]), 10))
}
if h.SSRC != nil {

2
pkg/headers/transport_test.go
View File

@@ -48,7 +48,7 @@ var casesTransport = []struct {
{
"udp multicast play request / response",
base.HeaderValue{`RTP/AVP;multicast;destination=225.219.201.15;port=7000-7001;ttl=127`},
base.HeaderValue{`RTP/AVP;multicast`},
base.HeaderValue{`RTP/AVP;multicast;destination=225.219.201.15;port=7000-7001;ttl=127`},
Transport{
Protocol: base.StreamProtocolUDP,
Delivery: func() *base.StreamDelivery {

20
server.go
View File

@@ -126,11 +126,14 @@ type Server struct {
sessions map[string]*ServerSession
conns map[*ServerConn]struct{}
exitError error
streams map[*ServerStream]struct{}
// in
connClose chan *ServerConn
sessionRequest chan request
sessionClose chan *ServerSession
streamAdd chan *ServerStream
streamRemove chan *ServerStream
}
// Start starts listening on the given address.
@@ -195,12 +198,12 @@ func (s *Server) Start(address string) error {
return fmt.Errorf("RTCP and RTP ports must be consecutive")
}
s.udpRTPListener, err = newServerUDPListener(s, s.UDPRTPAddress, StreamTypeRTP)
s.udpRTPListener, err = newServerUDPListener(s, false, s.UDPRTPAddress, StreamTypeRTP)
if err != nil {
return err
}
s.udpRTCPListener, err = newServerUDPListener(s, s.UDPRTCPAddress, StreamTypeRTCP)
s.udpRTCPListener, err = newServerUDPListener(s, false, s.UDPRTCPAddress, StreamTypeRTCP)
if err != nil {
s.udpRTPListener.close()
return err
@@ -241,9 +244,12 @@ func (s *Server) run() {
s.sessions = make(map[string]*ServerSession)
s.conns = make(map[*ServerConn]struct{})
s.streams = make(map[*ServerStream]struct{})
s.connClose = make(chan *ServerConn)
s.sessionRequest = make(chan request)
s.sessionClose = make(chan *ServerSession)
s.streamAdd = make(chan *ServerStream)
s.streamRemove = make(chan *ServerStream)
s.wg.Add(1)
connNew := make(chan net.Conn)
@@ -336,6 +342,12 @@ outer:
delete(s.sessions, ss.id)
ss.Close()
case st := <-s.streamAdd:
s.streams[st] = struct{}{}
case st := <-s.streamRemove:
delete(s.streams, st)
case <-s.ctx.Done():
break outer
}
@@ -352,6 +364,10 @@ outer:
}
s.tcpListener.Close()
for st := range s.streams {
st.Close()
}
}
// StartAndWait starts the server and waits until a fatal error.

40
server_publish_test.go
View File

@@ -301,12 +301,12 @@ func TestServerPublishSetupPath(t *testing.T) {
StatusCode: base.StatusOK,
}, nil
},
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
require.Equal(t, ca.path, ctx.Path)
require.Equal(t, ca.trackID, ctx.TrackID)
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, nil, nil, nil
},
},
}
@@ -398,10 +398,10 @@ func TestServerPublishErrorSetupDifferentPaths(t *testing.T) {
StatusCode: base.StatusOK,
}, nil
},
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, nil, nil, nil
},
},
}
@@ -480,10 +480,10 @@ func TestServerPublishErrorSetupTrackTwice(t *testing.T) {
StatusCode: base.StatusOK,
}, nil
},
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, nil, nil, nil
},
},
}
@@ -574,10 +574,10 @@ func TestServerPublishErrorRecordPartialTracks(t *testing.T) {
StatusCode: base.StatusOK,
}, nil
},
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, nil, nil, nil
},
onRecord: func(ctx *ServerHandlerOnRecordCtx) (*base.Response, error) {
return &base.Response{
@@ -694,10 +694,10 @@ func TestServerPublish(t *testing.T) {
StatusCode: base.StatusOK,
}, nil
},
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, nil, nil, nil
},
onRecord: func(ctx *ServerHandlerOnRecordCtx) (*base.Response, error) {
return &base.Response{
@@ -912,10 +912,10 @@ func TestServerPublishErrorWrongProtocol(t *testing.T) {
StatusCode: base.StatusOK,
}, nil
},
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, nil, nil, nil
},
onRecord: func(ctx *ServerHandlerOnRecordCtx) (*base.Response, error) {
return &base.Response{
@@ -1018,10 +1018,10 @@ func TestServerPublishRTCPReport(t *testing.T) {
StatusCode: base.StatusOK,
}, nil
},
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, nil, nil, nil
},
onRecord: func(ctx *ServerHandlerOnRecordCtx) (*base.Response, error) {
return &base.Response{
@@ -1174,10 +1174,10 @@ func TestServerPublishTimeout(t *testing.T) {
StatusCode: base.StatusOK,
}, nil
},
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, nil, nil, nil
},
onRecord: func(ctx *ServerHandlerOnRecordCtx) (*base.Response, error) {
return &base.Response{
@@ -1302,10 +1302,10 @@ func TestServerPublishWithoutTeardown(t *testing.T) {
StatusCode: base.StatusOK,
}, nil
},
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, nil, nil, nil
},
onRecord: func(ctx *ServerHandlerOnRecordCtx) (*base.Response, error) {
return &base.Response{
@@ -1414,10 +1414,10 @@ func TestServerPublishUDPChangeConn(t *testing.T) {
StatusCode: base.StatusOK,
}, nil
},
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, nil, nil, nil
},
onRecord: func(ctx *ServerHandlerOnRecordCtx) (*base.Response, error) {
return &base.Response{

235
server_read_test.go
View File

@@ -4,10 +4,12 @@ import (
"bufio"
"crypto/tls"
"net"
"strconv"
"testing"
"time"
"github.com/stretchr/testify/require"
"golang.org/x/net/ipv4"
"github.com/aler9/gortsplib/pkg/base"
"github.com/aler9/gortsplib/pkg/headers"
@@ -59,19 +61,24 @@ func TestServerReadSetupPath(t *testing.T) {
},
} {
t.Run(ca.name, func(t *testing.T) {
track, err := NewTrackH264(96, []byte("123456"), []byte("123456"))
require.NoError(t, err)
stream := NewServerStream(Tracks{track})
s := &Server{
Handler: &testServerHandler{
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
require.Equal(t, ca.path, ctx.Path)
require.Equal(t, ca.trackID, ctx.TrackID)
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, stream, nil, nil
},
},
}
err := s.Start("localhost:8554")
err = s.Start("localhost:8554")
require.NoError(t, err)
defer s.Close()
@@ -110,21 +117,26 @@ func TestServerReadSetupPath(t *testing.T) {
func TestServerReadErrorSetupDifferentPaths(t *testing.T) {
connClosed := make(chan struct{})
track, err := NewTrackH264(96, []byte("123456"), []byte("123456"))
require.NoError(t, err)
stream := NewServerStream(Tracks{track})
s := &Server{
Handler: &testServerHandler{
onConnClose: func(ctx *ServerHandlerOnConnCloseCtx) {
require.Equal(t, "can't setup tracks with different paths", ctx.Error.Error())
close(connClosed)
},
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, stream, nil, nil
},
},
}
err := s.Start("localhost:8554")
err = s.Start("localhost:8554")
require.NoError(t, err)
defer s.Close()
@@ -177,21 +189,26 @@ func TestServerReadErrorSetupDifferentPaths(t *testing.T) {
func TestServerReadErrorSetupTrackTwice(t *testing.T) {
connClosed := make(chan struct{})
track, err := NewTrackH264(96, []byte("123456"), []byte("123456"))
require.NoError(t, err)
stream := NewServerStream(Tracks{track})
s := &Server{
Handler: &testServerHandler{
onConnClose: func(ctx *ServerHandlerOnConnCloseCtx) {
require.Equal(t, "track 0 has already been setup", ctx.Error.Error())
close(connClosed)
},
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, stream, nil, nil
},
},
}
err := s.Start("localhost:8554")
err = s.Start("localhost:8554")
require.NoError(t, err)
defer s.Close()
@@ -246,6 +263,7 @@ func TestServerRead(t *testing.T) {
"udp",
"tcp",
"tls",
"multicast",
} {
t.Run(proto, func(t *testing.T) {
connOpened := make(chan struct{})
@@ -254,6 +272,11 @@ func TestServerRead(t *testing.T) {
sessionClosed := make(chan struct{})
framesReceived := make(chan struct{})
track, err := NewTrackH264(96, []byte("123456"), []byte("123456"))
require.NoError(t, err)
stream := NewServerStream(Tracks{track})
s := &Server{
Handler: &testServerHandler{
onConnOpen: func(ctx *ServerHandlerOnConnOpenCtx) {
@@ -268,15 +291,19 @@ func TestServerRead(t *testing.T) {
onSessionClose: func(ctx *ServerHandlerOnSessionCloseCtx) {
close(sessionClosed)
},
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
v := uint32(123456)
return &base.Response{
StatusCode: base.StatusOK,
}, &v, nil
}, stream, &v, nil
},
onPlay: func(ctx *ServerHandlerOnPlayCtx) (*base.Response, error) {
ctx.Session.WriteFrame(0, StreamTypeRTP, []byte{0x01, 0x02, 0x03, 0x04})
ctx.Session.WriteFrame(0, StreamTypeRTCP, []byte{0x05, 0x06, 0x07, 0x08})
go func() {
time.Sleep(1 * time.Second)
stream.WriteFrame(0, StreamTypeRTP, []byte{0x01, 0x02, 0x03, 0x04})
stream.WriteFrame(0, StreamTypeRTCP, []byte{0x05, 0x06, 0x07, 0x08})
}()
return &base.Response{
StatusCode: base.StatusOK,
@@ -297,7 +324,7 @@ func TestServerRead(t *testing.T) {
}
switch proto {
case "udp":
case "udp", "multicast":
s.UDPRTPAddress = "127.0.0.1:8000"
s.UDPRTCPAddress = "127.0.0.1:8001"
@@ -307,7 +334,7 @@ func TestServerRead(t *testing.T) {
s.TLSConfig = &tls.Config{Certificates: []tls.Certificate{cert}}
}
err := s.Start("localhost:8554")
err = s.Start("localhost:8554")
require.NoError(t, err)
defer s.Close()
@@ -325,20 +352,27 @@ func TestServerRead(t *testing.T) {
<-connOpened
inTH := &headers.Transport{
Delivery: func() *base.StreamDelivery {
v := base.StreamDeliveryUnicast
return &v
}(),
Mode: func() *headers.TransportMode {
v := headers.TransportModePlay
return &v
}(),
}
if proto == "udp" {
switch proto {
case "udp":
v := base.StreamDeliveryUnicast
inTH.Delivery = &v
inTH.Protocol = base.StreamProtocolUDP
inTH.ClientPorts = &[2]int{35466, 35467}
} else {
case "multicast":
v := base.StreamDeliveryMulticast
inTH.Delivery = &v
inTH.Protocol = base.StreamProtocolUDP
default:
v := base.StreamDeliveryUnicast
inTH.Delivery = &v
inTH.Protocol = base.StreamProtocolTCP
inTH.InterleavedIDs = &[2]int{0, 1}
}
@@ -363,7 +397,8 @@ func TestServerRead(t *testing.T) {
var l1 net.PacketConn
var l2 net.PacketConn
if proto == "udp" {
switch proto {
case "udp":
l1, err = net.ListenPacket("udp", "localhost:35466")
require.NoError(t, err)
defer l1.Close()
@@ -371,6 +406,35 @@ func TestServerRead(t *testing.T) {
l2, err = net.ListenPacket("udp", "localhost:35467")
require.NoError(t, err)
defer l2.Close()
case "multicast":
l1, err = net.ListenPacket("udp4", "224.0.0.0:"+strconv.FormatInt(int64(th.Ports[0]), 10))
require.NoError(t, err)
defer l1.Close()
p := ipv4.NewPacketConn(l1)
intfs, err := net.Interfaces()
require.NoError(t, err)
for _, intf := range intfs {
err := p.JoinGroup(&intf, &net.UDPAddr{IP: multicastIP})
require.NoError(t, err)
}
l2, err = net.ListenPacket("udp4", "224.0.0.0:"+strconv.FormatInt(int64(th.Ports[1]), 10))
require.NoError(t, err)
defer l2.Close()
p = ipv4.NewPacketConn(l2)
intfs, err = net.Interfaces()
require.NoError(t, err)
for _, intf := range intfs {
err := p.JoinGroup(&intf, &net.UDPAddr{IP: multicastIP})
require.NoError(t, err)
}
}
res, err = writeReqReadRes(bconn, base.Request{
@@ -385,17 +449,23 @@ func TestServerRead(t *testing.T) {
require.Equal(t, base.StatusOK, res.StatusCode)
// server -> client
if proto == "udp" {
if proto == "udp" || proto == "multicast" {
buf := make([]byte, 2048)
n, _, err := l1.ReadFrom(buf)
require.NoError(t, err)
require.Equal(t, []byte{0x01, 0x02, 0x03, 0x04}, buf[:n])
buf = make([]byte, 2048)
// skip firewall opening
if proto == "udp" {
_, _, err = l2.ReadFrom(buf)
require.NoError(t, err)
}
n, _, err = l2.ReadFrom(buf)
require.NoError(t, err)
require.Equal(t, []byte{0x05, 0x06, 0x07, 0x08}, buf[:n])
} else {
var f base.InterleavedFrame
f.Payload = make([]byte, 2048)
@@ -414,23 +484,29 @@ func TestServerRead(t *testing.T) {
}
// client -> server (RTCP)
if proto == "udp" {
switch proto {
case "udp":
l2.WriteTo([]byte{0x01, 0x02, 0x03, 0x04}, &net.UDPAddr{
IP: net.ParseIP("127.0.0.1"),
Port: th.ServerPorts[1],
})
} else {
<-framesReceived
case "multicast":
// sending RTCP with multicast is currently not supported
// since the source IP cannot be verified correctly
default:
err = base.InterleavedFrame{
TrackID: 0,
StreamType: StreamTypeRTCP,
Payload: []byte{0x01, 0x02, 0x03, 0x04},
}.Write(bconn.Writer)
require.NoError(t, err)
<-framesReceived
}
<-framesReceived
if proto == "udp" {
if proto == "udp" || proto == "multicast" {
// ping with OPTIONS
res, err = writeReqReadRes(bconn, base.Request{
Method: base.Options,
@@ -479,22 +555,27 @@ func TestServerReadTCPResponseBeforeFrames(t *testing.T) {
writerDone := make(chan struct{})
writerTerminate := make(chan struct{})
track, err := NewTrackH264(96, []byte("123456"), []byte("123456"))
require.NoError(t, err)
stream := NewServerStream(Tracks{track})
s := &Server{
Handler: &testServerHandler{
onConnClose: func(ctx *ServerHandlerOnConnCloseCtx) {
close(writerTerminate)
<-writerDone
},
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, stream, nil, nil
},
onPlay: func(ctx *ServerHandlerOnPlayCtx) (*base.Response, error) {
go func() {
defer close(writerDone)
ctx.Session.WriteFrame(0, StreamTypeRTP, []byte("\x00\x00\x00\x00"))
stream.WriteFrame(0, StreamTypeRTP, []byte("\x00\x00\x00\x00"))
t := time.NewTicker(50 * time.Millisecond)
defer t.Stop()
@@ -502,7 +583,7 @@ func TestServerReadTCPResponseBeforeFrames(t *testing.T) {
for {
select {
case <-t.C:
ctx.Session.WriteFrame(0, StreamTypeRTP, []byte("\x00\x00\x00\x00"))
stream.WriteFrame(0, StreamTypeRTP, []byte("\x00\x00\x00\x00"))
case <-writerTerminate:
return
}
@@ -518,7 +599,7 @@ func TestServerReadTCPResponseBeforeFrames(t *testing.T) {
},
}
err := s.Start("localhost:8554")
err = s.Start("localhost:8554")
require.NoError(t, err)
defer s.Close()
@@ -567,12 +648,17 @@ func TestServerReadTCPResponseBeforeFrames(t *testing.T) {
}
func TestServerReadPlayPlay(t *testing.T) {
track, err := NewTrackH264(96, []byte("123456"), []byte("123456"))
require.NoError(t, err)
stream := NewServerStream(Tracks{track})
s := &Server{
Handler: &testServerHandler{
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, stream, nil, nil
},
onPlay: func(ctx *ServerHandlerOnPlayCtx) (*base.Response, error) {
return &base.Response{
@@ -584,7 +670,7 @@ func TestServerReadPlayPlay(t *testing.T) {
UDPRTCPAddress: "127.0.0.1:8001",
}
err := s.Start("localhost:8554")
err = s.Start("localhost:8554")
require.NoError(t, err)
defer s.Close()
@@ -643,16 +729,21 @@ func TestServerReadPlayPausePlay(t *testing.T) {
writerDone := make(chan struct{})
writerTerminate := make(chan struct{})
track, err := NewTrackH264(96, []byte("123456"), []byte("123456"))
require.NoError(t, err)
stream := NewServerStream(Tracks{track})
s := &Server{
Handler: &testServerHandler{
onConnClose: func(ctx *ServerHandlerOnConnCloseCtx) {
close(writerTerminate)
<-writerDone
},
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, stream, nil, nil
},
onPlay: func(ctx *ServerHandlerOnPlayCtx) (*base.Response, error) {
if !writerStarted {
@@ -666,7 +757,7 @@ func TestServerReadPlayPausePlay(t *testing.T) {
for {
select {
case <-t.C:
ctx.Session.WriteFrame(0, StreamTypeRTP, []byte("\x00\x00\x00\x00"))
stream.WriteFrame(0, StreamTypeRTP, []byte("\x00\x00\x00\x00"))
case <-writerTerminate:
return
}
@@ -686,7 +777,7 @@ func TestServerReadPlayPausePlay(t *testing.T) {
},
}
err := s.Start("localhost:8554")
err = s.Start("localhost:8554")
require.NoError(t, err)
defer s.Close()
@@ -755,16 +846,21 @@ func TestServerReadPlayPausePause(t *testing.T) {
writerDone := make(chan struct{})
writerTerminate := make(chan struct{})
track, err := NewTrackH264(96, []byte("123456"), []byte("123456"))
require.NoError(t, err)
stream := NewServerStream(Tracks{track})
s := &Server{
Handler: &testServerHandler{
onConnClose: func(ctx *ServerHandlerOnConnCloseCtx) {
close(writerTerminate)
<-writerDone
},
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, stream, nil, nil
},
onPlay: func(ctx *ServerHandlerOnPlayCtx) (*base.Response, error) {
go func() {
@@ -776,7 +872,7 @@ func TestServerReadPlayPausePause(t *testing.T) {
for {
select {
case <-t.C:
ctx.Session.WriteFrame(0, StreamTypeRTP, []byte("\x00\x00\x00\x00"))
stream.WriteFrame(0, StreamTypeRTP, []byte("\x00\x00\x00\x00"))
case <-writerTerminate:
return
}
@@ -795,7 +891,7 @@ func TestServerReadPlayPausePause(t *testing.T) {
},
}
err := s.Start("localhost:8554")
err = s.Start("localhost:8554")
require.NoError(t, err)
defer s.Close()
@@ -874,6 +970,11 @@ func TestServerReadTimeout(t *testing.T) {
t.Run(proto, func(t *testing.T) {
sessionClosed := make(chan struct{})
track, err := NewTrackH264(96, []byte("123456"), []byte("123456"))
require.NoError(t, err)
stream := NewServerStream(Tracks{track})
s := &Server{
Handler: &testServerHandler{
onSessionClose: func(ctx *ServerHandlerOnSessionCloseCtx) {
@@ -884,10 +985,10 @@ func TestServerReadTimeout(t *testing.T) {
StatusCode: base.StatusOK,
}, nil
},
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, stream, nil, nil
},
onPlay: func(ctx *ServerHandlerOnPlayCtx) (*base.Response, error) {
return &base.Response{
@@ -902,7 +1003,7 @@ func TestServerReadTimeout(t *testing.T) {
s.UDPRTPAddress = "127.0.0.1:8000"
s.UDPRTCPAddress = "127.0.0.1:8001"
err := s.Start("localhost:8554")
err = s.Start("localhost:8554")
require.NoError(t, err)
defer s.Close()
@@ -961,6 +1062,11 @@ func TestServerReadWithoutTeardown(t *testing.T) {
connClosed := make(chan struct{})
sessionClosed := make(chan struct{})
track, err := NewTrackH264(96, []byte("123456"), []byte("123456"))
require.NoError(t, err)
stream := NewServerStream(Tracks{track})
s := &Server{
Handler: &testServerHandler{
onConnClose: func(ctx *ServerHandlerOnConnCloseCtx) {
@@ -974,10 +1080,10 @@ func TestServerReadWithoutTeardown(t *testing.T) {
StatusCode: base.StatusOK,
}, nil
},
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, stream, nil, nil
},
onPlay: func(ctx *ServerHandlerOnPlayCtx) (*base.Response, error) {
return &base.Response{
@@ -994,7 +1100,7 @@ func TestServerReadWithoutTeardown(t *testing.T) {
s.UDPRTCPAddress = "127.0.0.1:8001"
}
err := s.Start("localhost:8554")
err = s.Start("localhost:8554")
require.NoError(t, err)
defer s.Close()
@@ -1053,12 +1159,17 @@ func TestServerReadWithoutTeardown(t *testing.T) {
}
func TestServerReadUDPChangeConn(t *testing.T) {
track, err := NewTrackH264(96, []byte("123456"), []byte("123456"))
require.NoError(t, err)
stream := NewServerStream(Tracks{track})
s := &Server{
Handler: &testServerHandler{
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, stream, nil, nil
},
onPlay: func(ctx *ServerHandlerOnPlayCtx) (*base.Response, error) {
return &base.Response{
@@ -1077,7 +1188,7 @@ func TestServerReadUDPChangeConn(t *testing.T) {
UDPRTCPAddress: "127.0.0.1:8001",
}
err := s.Start("localhost:8554")
err = s.Start("localhost:8554")
require.NoError(t, err)
defer s.Close()
@@ -1146,19 +1257,25 @@ func TestServerReadUDPChangeConn(t *testing.T) {
}
func TestServerReadNonSetuppedPath(t *testing.T) {
track, err := NewTrackH264(96, []byte("123456"), []byte("123456"))
require.NoError(t, err)
stream := NewServerStream(Tracks{track})
s := &Server{
Handler: &testServerHandler{
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, stream, nil, nil
},
onPlay: func(ctx *ServerHandlerOnPlayCtx) (*base.Response, error) {
go func() {
time.Sleep(1 * time.Second)
ctx.Session.WriteFrame(1, base.StreamTypeRTP, []byte{0x01, 0x02, 0x03, 0x04})
ctx.Session.WriteFrame(0, base.StreamTypeRTP, []byte{0x05, 0x06, 0x07, 0x08})
stream.WriteFrame(1, base.StreamTypeRTP, []byte{0x01, 0x02, 0x03, 0x04})
stream.WriteFrame(0, base.StreamTypeRTP, []byte{0x05, 0x06, 0x07, 0x08})
}()
return &base.Response{
StatusCode: base.StatusOK,
}, nil
@@ -1166,7 +1283,7 @@ func TestServerReadNonSetuppedPath(t *testing.T) {
},
}
err := s.Start("localhost:8554")
err = s.Start("localhost:8554")
require.NoError(t, err)
defer s.Close()

153
server_test.go
View File

@@ -42,9 +42,9 @@ type testServerHandler struct {
onConnClose func(*ServerHandlerOnConnCloseCtx)
onSessionOpen func(*ServerHandlerOnSessionOpenCtx)
onSessionClose func(*ServerHandlerOnSessionCloseCtx)
onDescribe func(*ServerHandlerOnDescribeCtx) (*base.Response, []byte, error)
onDescribe func(*ServerHandlerOnDescribeCtx) (*base.Response, *ServerStream, error)
onAnnounce func(*ServerHandlerOnAnnounceCtx) (*base.Response, error)
onSetup func(*ServerHandlerOnSetupCtx) (*base.Response, *uint32, error)
onSetup func(*ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error)
onPlay func(*ServerHandlerOnPlayCtx) (*base.Response, error)
onRecord func(*ServerHandlerOnRecordCtx) (*base.Response, error)
onPause func(*ServerHandlerOnPauseCtx) (*base.Response, error)
@@ -77,7 +77,7 @@ func (sh *testServerHandler) OnSessionClose(ctx *ServerHandlerOnSessionCloseCtx)
}
}
func (sh *testServerHandler) OnDescribe(ctx *ServerHandlerOnDescribeCtx) (*base.Response, []byte, error) {
func (sh *testServerHandler) OnDescribe(ctx *ServerHandlerOnDescribeCtx) (*base.Response, *ServerStream, error) {
if sh.onDescribe != nil {
return sh.onDescribe(ctx)
}
@@ -91,11 +91,11 @@ func (sh *testServerHandler) OnAnnounce(ctx *ServerHandlerOnAnnounceCtx) (*base.
return nil, fmt.Errorf("unimplemented")
}
func (sh *testServerHandler) OnSetup(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
func (sh *testServerHandler) OnSetup(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
if sh.onSetup != nil {
return sh.onSetup(ctx)
}
return nil, nil, fmt.Errorf("unimplemented")
return nil, nil, nil, fmt.Errorf("unimplemented")
}
func (sh *testServerHandler) OnPlay(ctx *ServerHandlerOnPlayCtx) (*base.Response, error) {
@@ -262,13 +262,15 @@ func TestServerHighLevelPublishRead(t *testing.T) {
{"ffmpeg", "tls", "gstreamer", "tls"},
{"gstreamer", "tls", "ffmpeg", "tls"},
{"gstreamer", "tls", "gstreamer", "tls"},
{"ffmpeg", "udp", "ffmpeg", "multicast"},
{"ffmpeg", "udp", "gstreamer", "multicast"},
} {
t.Run(ca.publisherSoft+"_"+ca.publisherProto+"_"+
ca.readerSoft+"_"+ca.readerProto, func(t *testing.T) {
var mutex sync.Mutex
var stream *ServerStream
var publisher *ServerSession
var sdp []byte
readers := make(map[*ServerSession]struct{})
s := &Server{
Handler: &testServerHandler{
@@ -276,14 +278,14 @@ func TestServerHighLevelPublishRead(t *testing.T) {
mutex.Lock()
defer mutex.Unlock()
if stream != nil {
if ctx.Session == publisher {
publisher = nil
sdp = nil
} else {
delete(readers, ctx.Session)
stream.Close()
stream = nil
}
}
},
onDescribe: func(ctx *ServerHandlerOnDescribeCtx) (*base.Response, []byte, error) {
onDescribe: func(ctx *ServerHandlerOnDescribeCtx) (*base.Response, *ServerStream, error) {
if ctx.Path != "teststream" {
return &base.Response{
StatusCode: base.StatusBadRequest,
@@ -293,7 +295,7 @@ func TestServerHighLevelPublishRead(t *testing.T) {
mutex.Lock()
defer mutex.Unlock()
if publisher == nil {
if stream == nil {
return &base.Response{
StatusCode: base.StatusNotFound,
}, nil, nil
@@ -301,7 +303,7 @@ func TestServerHighLevelPublishRead(t *testing.T) {
return &base.Response{
StatusCode: base.StatusOK,
}, sdp, nil
}, stream, nil
},
onAnnounce: func(ctx *ServerHandlerOnAnnounceCtx) (*base.Response, error) {
if ctx.Path != "teststream" {
@@ -313,29 +315,35 @@ func TestServerHighLevelPublishRead(t *testing.T) {
mutex.Lock()
defer mutex.Unlock()
if publisher != nil {
if stream != nil {
return &base.Response{
StatusCode: base.StatusBadRequest,
}, fmt.Errorf("someone is already publishing")
}
stream = NewServerStream(ctx.Tracks)
publisher = ctx.Session
sdp = ctx.Tracks.Write()
return &base.Response{
StatusCode: base.StatusOK,
}, nil
},
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
if ctx.Path != "teststream" {
return &base.Response{
StatusCode: base.StatusBadRequest,
}, nil, fmt.Errorf("invalid path (%s)", ctx.Req.URL)
}, nil, nil, fmt.Errorf("invalid path (%s)", ctx.Req.URL)
}
if stream == nil {
return &base.Response{
StatusCode: base.StatusNotFound,
}, nil, nil, nil
}
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, stream, nil, nil
},
onPlay: func(ctx *ServerHandlerOnPlayCtx) (*base.Response, error) {
if ctx.Path != "teststream" {
@@ -344,11 +352,6 @@ func TestServerHighLevelPublishRead(t *testing.T) {
}, fmt.Errorf("invalid path (%s)", ctx.Req.URL)
}
mutex.Lock()
defer mutex.Unlock()
readers[ctx.Session] = struct{}{}
return &base.Response{
StatusCode: base.StatusOK,
}, nil
@@ -369,29 +372,21 @@ func TestServerHighLevelPublishRead(t *testing.T) {
defer mutex.Unlock()
if ctx.Session == publisher {
for r := range readers {
r.WriteFrame(ctx.TrackID, ctx.StreamType, ctx.Payload)
}
stream.WriteFrame(ctx.TrackID, ctx.StreamType, ctx.Payload)
}
},
},
}
var proto string
var publisherSubProto string
var readerSubProto string
if ca.publisherProto == "tls" {
proto = "rtsps"
publisherSubProto = "tcp"
readerSubProto = "tcp"
cert, err := tls.X509KeyPair(serverCert, serverKey)
require.NoError(t, err)
s.TLSConfig = &tls.Config{Certificates: []tls.Certificate{cert}}
} else {
proto = "rtsp"
publisherSubProto = ca.publisherProto
readerSubProto = ca.readerProto
s.UDPRTPAddress = "127.0.0.1:8000"
s.UDPRTCPAddress = "127.0.0.1:8001"
}
@@ -402,22 +397,38 @@ func TestServerHighLevelPublishRead(t *testing.T) {
switch ca.publisherSoft {
case "ffmpeg":
ts := func() string {
switch ca.publisherProto {
case "udp", "tcp":
return ca.publisherProto
}
return "tcp"
}()
cnt1, err := newContainer("ffmpeg", "publish", []string{
"-re",
"-stream_loop", "-1",
"-i", "emptyvideo.mkv",
"-c", "copy",
"-f", "rtsp",
"-rtsp_transport", publisherSubProto,
"-rtsp_transport", ts,
proto + "://localhost:8554/teststream",
})
require.NoError(t, err)
defer cnt1.close()
case "gstreamer":
ts := func() string {
switch ca.publisherProto {
case "udp", "tcp":
return ca.publisherProto
}
return "tcp"
}()
cnt1, err := newContainer("gstreamer", "publish", []string{
"filesrc location=emptyvideo.mkv ! matroskademux ! video/x-h264 ! rtspclientsink " +
"location=" + proto + "://127.0.0.1:8554/teststream protocols=" + publisherSubProto + " tls-validation-flags=0 latency=0 timeout=0 rtx-time=0",
"location=" + proto + "://127.0.0.1:8554/teststream protocols=" + ts + " tls-validation-flags=0 latency=0 timeout=0 rtx-time=0",
})
require.NoError(t, err)
defer cnt1.close()
@@ -429,8 +440,18 @@ func TestServerHighLevelPublishRead(t *testing.T) {
switch ca.readerSoft {
case "ffmpeg":
ts := func() string {
switch ca.readerProto {
case "udp", "tcp":
return ca.readerProto
case "multicast":
return "udp_multicast"
}
return "tcp"
}()
cnt2, err := newContainer("ffmpeg", "read", []string{
"-rtsp_transport", readerSubProto,
"-rtsp_transport", ts,
"-i", proto + "://localhost:8554/teststream",
"-vframes", "1",
"-f", "image2",
@@ -441,8 +462,18 @@ func TestServerHighLevelPublishRead(t *testing.T) {
require.Equal(t, 0, cnt2.wait())
case "gstreamer":
ts := func() string {
switch ca.readerProto {
case "udp", "tcp":
return ca.readerProto
case "multicast":
return "udp-mcast"
}
return "tcp"
}()
cnt2, err := newContainer("gstreamer", "read", []string{
"rtspsrc location=" + proto + "://127.0.0.1:8554/teststream protocols=" + readerSubProto + " tls-validation-flags=0 latency=0 " +
"rtspsrc location=" + proto + "://127.0.0.1:8554/teststream protocols=" + ts + " tls-validation-flags=0 latency=0 " +
"! application/x-rtp,media=video ! decodebin ! exitafterframe ! fakesink",
})
require.NoError(t, err)
@@ -599,12 +630,17 @@ func TestServerErrorInvalidMethod(t *testing.T) {
}
func TestServerErrorTCPTwoConnOneSession(t *testing.T) {
track, err := NewTrackH264(96, []byte("123456"), []byte("123456"))
require.NoError(t, err)
stream := NewServerStream(Tracks{track})
s := &Server{
Handler: &testServerHandler{
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, stream, nil, nil
},
onPlay: func(ctx *ServerHandlerOnPlayCtx) (*base.Response, error) {
return &base.Response{
@@ -619,7 +655,7 @@ func TestServerErrorTCPTwoConnOneSession(t *testing.T) {
},
}
err := s.Start("localhost:8554")
err = s.Start("localhost:8554")
require.NoError(t, err)
defer s.Close()
@@ -691,12 +727,17 @@ func TestServerErrorTCPTwoConnOneSession(t *testing.T) {
}
func TestServerErrorTCPOneConnTwoSessions(t *testing.T) {
track, err := NewTrackH264(96, []byte("123456"), []byte("123456"))
require.NoError(t, err)
stream := NewServerStream(Tracks{track})
s := &Server{
Handler: &testServerHandler{
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, stream, nil, nil
},
onPlay: func(ctx *ServerHandlerOnPlayCtx) (*base.Response, error) {
return &base.Response{
@@ -711,7 +752,7 @@ func TestServerErrorTCPOneConnTwoSessions(t *testing.T) {
},
}
err := s.Start("localhost:8554")
err = s.Start("localhost:8554")
require.NoError(t, err)
defer s.Close()
@@ -901,10 +942,10 @@ func TestServerSessionClose(t *testing.T) {
onSessionClose: func(ctx *ServerHandlerOnSessionCloseCtx) {
close(sessionClosed)
},
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, nil, nil, nil
},
},
}
@@ -945,20 +986,25 @@ func TestServerSessionClose(t *testing.T) {
func TestServerSessionAutoClose(t *testing.T) {
sessionClosed := make(chan struct{})
track, err := NewTrackH264(96, []byte("123456"), []byte("123456"))
require.NoError(t, err)
stream := NewServerStream(Tracks{track})
s := &Server{
Handler: &testServerHandler{
onSessionClose: func(ctx *ServerHandlerOnSessionCloseCtx) {
close(sessionClosed)
},
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, stream, nil, nil
},
},
}
err := s.Start("localhost:8554")
err = s.Start("localhost:8554")
require.NoError(t, err)
defer s.Close()
@@ -1006,6 +1052,11 @@ func TestServerErrorInvalidPath(t *testing.T) {
t.Run(string(method), func(t *testing.T) {
connClosed := make(chan struct{})
track, err := NewTrackH264(96, []byte("123456"), []byte("123456"))
require.NoError(t, err)
stream := NewServerStream(Tracks{track})
s := &Server{
Handler: &testServerHandler{
onConnClose: func(ctx *ServerHandlerOnConnCloseCtx) {
@@ -1017,10 +1068,10 @@ func TestServerErrorInvalidPath(t *testing.T) {
StatusCode: base.StatusOK,
}, nil
},
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *uint32, error) {
onSetup: func(ctx *ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error) {
return &base.Response{
StatusCode: base.StatusOK,
}, nil, nil
}, stream, nil, nil
},
onPlay: func(ctx *ServerHandlerOnPlayCtx) (*base.Response, error) {
return &base.Response{
@@ -1030,7 +1081,7 @@ func TestServerErrorInvalidPath(t *testing.T) {
},
}
err := s.Start("localhost:8554")
err = s.Start("localhost:8554")
require.NoError(t, err)
defer s.Close()

6
serverconn.go
View File

@@ -336,21 +336,21 @@ func (sc *ServerConn) handleRequest(req *base.Request) (*base.Response, error) {
path, query := base.PathSplitQuery(pathAndQuery)
res, sdp, err := h.OnDescribe(&ServerHandlerOnDescribeCtx{
res, stream, err := h.OnDescribe(&ServerHandlerOnDescribeCtx{
Conn: sc,
Req: req,
Path: path,
Query: query,
})
if res.StatusCode == base.StatusOK && sdp != nil {
if res.StatusCode == base.StatusOK {
if res.Header == nil {
res.Header = make(base.Header)
}
res.Header["Content-Base"] = base.HeaderValue{req.URL.String() + "/"}
res.Header["Content-Type"] = base.HeaderValue{"application/sdp"}
res.Body = sdp
res.Body = stream.Tracks().Write()
}
return res, err

6
serverhandler.go
View File

@@ -71,11 +71,12 @@ type ServerHandlerOnDescribeCtx struct {
// ServerHandlerOnDescribe can be implemented by a ServerHandler.
type ServerHandlerOnDescribe interface {
OnDescribe(*ServerHandlerOnDescribeCtx) (*base.Response, []byte, error)
OnDescribe(*ServerHandlerOnDescribeCtx) (*base.Response, *ServerStream, error)
}
// ServerHandlerOnAnnounceCtx is the context of an ANNOUNCE request.
type ServerHandlerOnAnnounceCtx struct {
Server *Server
Session *ServerSession
Conn *ServerConn
Req *base.Request
@@ -91,6 +92,7 @@ type ServerHandlerOnAnnounce interface {
// ServerHandlerOnSetupCtx is the context of a OPTIONS request.
type ServerHandlerOnSetupCtx struct {
Server *Server
Session *ServerSession
Conn *ServerConn
Req *base.Request
@@ -104,7 +106,7 @@ type ServerHandlerOnSetupCtx struct {
type ServerHandlerOnSetup interface {
// returns a Response and an optional SSRC that is
// inserted into the Transport header.
OnSetup(*ServerHandlerOnSetupCtx) (*base.Response, *uint32, error)
OnSetup(*ServerHandlerOnSetupCtx) (*base.Response, *ServerStream, *uint32, error)
}
// ServerHandlerOnPlayCtx is the context of a PLAY request.

173
serversession.go
View File

@@ -23,7 +23,7 @@ func setupGetTrackIDPathQuery(
url *base.URL,
thMode *headers.TransportMode,
announcedTracks []ServerSessionAnnouncedTrack,
setupPath *string, setupQuery *string) (int, string, string, error) {
setuppedPath *string, setuppedQuery *string) (int, string, string, error) {
pathAndQuery, ok := url.RTSPPathAndQuery()
if !ok {
return 0, "", "", liberrors.ErrServerInvalidPath{}
@@ -54,7 +54,7 @@ func setupGetTrackIDPathQuery(
path, query := base.PathSplitQuery(pathAndQuery)
if setupPath != nil && (path != *setupPath || query != *setupQuery) {
if setuppedPath != nil && (path != *setuppedPath || query != *setuppedQuery) {
return 0, "", "", fmt.Errorf("can't setup tracks with different paths")
}
@@ -64,7 +64,7 @@ func setupGetTrackIDPathQuery(
for trackID, track := range announcedTracks {
u, _ := track.track.URL()
if u.String() == url.String() {
return trackID, *setupPath, *setupQuery, nil
return trackID, *setuppedPath, *setuppedQuery, nil
}
}
@@ -123,9 +123,11 @@ type ServerSession struct {
conns map[*ServerConn]struct{}
state ServerSessionState
setuppedTracks map[int]ServerSessionSetuppedTrack
setupProtocol *base.StreamProtocol
setupPath *string
setupQuery *string
setuppedProtocol *base.StreamProtocol
setuppedDelivery *base.StreamDelivery
setuppedStream *ServerStream // read
setuppedPath *string
setuppedQuery *string
lastRequestTime time.Time
tcpConn *ServerConn // tcp
udpIP net.IP // udp
@@ -179,16 +181,21 @@ func (ss *ServerSession) State() ServerSessionState {
return ss.state
}
// StreamProtocol returns the stream protocol of the setupped tracks.
func (ss *ServerSession) StreamProtocol() *base.StreamProtocol {
return ss.setupProtocol
}
// SetuppedTracks returns the setupped tracks.
func (ss *ServerSession) SetuppedTracks() map[int]ServerSessionSetuppedTrack {
return ss.setuppedTracks
}
// SetuppedProtocol returns the stream protocol of the setupped tracks.
func (ss *ServerSession) SetuppedProtocol() *base.StreamProtocol {
return ss.setuppedProtocol
}
// SetuppedDelivery returns the delivery method of the setupped tracks.
func (ss *ServerSession) SetuppedDelivery() *base.StreamDelivery {
return ss.setuppedDelivery
}
// AnnouncedTracks returns the announced tracks.
func (ss *ServerSession) AnnouncedTracks() []ServerSessionAnnouncedTrack {
return ss.announcedTracks
@@ -268,7 +275,7 @@ func (ss *ServerSession) run() {
// if session is not in state RECORD or PLAY, or protocol is TCP
if (ss.state != ServerSessionStateRecord &&
ss.state != ServerSessionStatePlay) ||
*ss.setupProtocol == base.StreamProtocolTCP {
*ss.setuppedProtocol == base.StreamProtocolTCP {
// close if there are no active connections
if len(ss.conns) == 0 {
@@ -279,7 +286,7 @@ func (ss *ServerSession) run() {
case <-checkTimeoutTicker.C:
switch {
// in case of RECORD and UDP, timeout happens when no frames are being received
case ss.state == ServerSessionStateRecord && *ss.setupProtocol == base.StreamProtocolUDP:
case ss.state == ServerSessionStateRecord && *ss.setuppedProtocol == base.StreamProtocolUDP:
now := time.Now()
lft := atomic.LoadInt64(ss.udpLastFrameTime)
if now.Sub(time.Unix(lft, 0)) >= ss.s.ReadTimeout {
@@ -287,7 +294,7 @@ func (ss *ServerSession) run() {
}
// in case of PLAY and UDP, timeout happens when no request arrives
case ss.state == ServerSessionStatePlay && *ss.setupProtocol == base.StreamProtocolUDP:
case ss.state == ServerSessionStatePlay && *ss.setuppedProtocol == base.StreamProtocolUDP:
now := time.Now()
if now.Sub(ss.lastRequestTime) >= ss.s.closeSessionAfterNoRequestsFor {
return liberrors.ErrServerSessionTimedOut{}
@@ -317,17 +324,23 @@ func (ss *ServerSession) run() {
switch ss.state {
case ServerSessionStatePlay:
if *ss.setupProtocol == base.StreamProtocolUDP {
ss.setuppedStream.readerSetInactive(ss)
if *ss.setuppedProtocol == base.StreamProtocolUDP {
ss.s.udpRTCPListener.removeClient(ss)
}
case ServerSessionStateRecord:
if *ss.setupProtocol == base.StreamProtocolUDP {
if *ss.setuppedProtocol == base.StreamProtocolUDP {
ss.s.udpRTPListener.removeClient(ss)
ss.s.udpRTCPListener.removeClient(ss)
}
}
if ss.setuppedStream != nil {
ss.setuppedStream.readerRemove(ss)
}
for sc := range ss.conns {
if sc == ss.tcpConn {
sc.Close()
@@ -465,6 +478,7 @@ func (ss *ServerSession) handleRequest(sc *ServerConn, req *base.Request) (*base
}
res, err := ss.s.Handler.(ServerHandlerOnAnnounce).OnAnnounce(&ServerHandlerOnAnnounceCtx{
Server: ss.s,
Session: ss,
Conn: sc,
Req: req,
@@ -475,8 +489,8 @@ func (ss *ServerSession) handleRequest(sc *ServerConn, req *base.Request) (*base
if res.StatusCode == base.StatusOK {
ss.state = ServerSessionStatePreRecord
ss.setupPath = &path
ss.setupQuery = &query
ss.setuppedPath = &path
ss.setuppedQuery = &query
ss.announcedTracks = make([]ServerSessionAnnouncedTrack, len(tracks))
for trackID, track := range tracks {
@@ -513,14 +527,8 @@ func (ss *ServerSession) handleRequest(sc *ServerConn, req *base.Request) (*base
}, liberrors.ErrServerTransportHeaderInvalid{Err: err}
}
if inTH.Delivery != nil && *inTH.Delivery == base.StreamDeliveryMulticast {
return &base.Response{
StatusCode: base.StatusUnsupportedTransport,
}, nil
}
trackID, path, query, err := setupGetTrackIDPathQuery(req.URL, inTH.Mode,
ss.announcedTracks, ss.setupPath, ss.setupQuery)
ss.announcedTracks, ss.setuppedPath, ss.setuppedQuery)
if err != nil {
return &base.Response{
StatusCode: base.StatusBadRequest,
@@ -533,6 +541,13 @@ func (ss *ServerSession) handleRequest(sc *ServerConn, req *base.Request) (*base
}, liberrors.ErrServerTrackAlreadySetup{TrackID: trackID}
}
delivery := func() base.StreamDelivery {
if inTH.Delivery != nil {
return *inTH.Delivery
}
return base.StreamDeliveryUnicast
}()
switch ss.state {
case ServerSessionStateInitial, ServerSessionStatePrePlay: // play
if inTH.Mode != nil && *inTH.Mode != headers.TransportModePlay {
@@ -542,6 +557,12 @@ func (ss *ServerSession) handleRequest(sc *ServerConn, req *base.Request) (*base
}
default: // record
if delivery == base.StreamDeliveryMulticast {
return &base.Response{
StatusCode: base.StatusUnsupportedTransport,
}, nil
}
if inTH.Mode == nil || *inTH.Mode != headers.TransportModeRecord {
return &base.Response{
StatusCode: base.StatusBadRequest,
@@ -556,13 +577,19 @@ func (ss *ServerSession) handleRequest(sc *ServerConn, req *base.Request) (*base
}, nil
}
if inTH.ClientPorts == nil {
if delivery == base.StreamDeliveryUnicast && inTH.ClientPorts == nil {
return &base.Response{
StatusCode: base.StatusBadRequest,
}, liberrors.ErrServerTransportHeaderNoClientPorts{}
}
} else {
if delivery == base.StreamDeliveryMulticast {
return &base.Response{
StatusCode: base.StatusUnsupportedTransport,
}, nil
}
if inTH.InterleavedIDs == nil {
return &base.Response{
StatusCode: base.StatusBadRequest,
@@ -579,13 +606,15 @@ func (ss *ServerSession) handleRequest(sc *ServerConn, req *base.Request) (*base
}
}
if ss.setupProtocol != nil && *ss.setupProtocol != inTH.Protocol {
if ss.setuppedProtocol != nil &&
(*ss.setuppedProtocol != inTH.Protocol || *ss.setuppedDelivery != delivery) {
return &base.Response{
StatusCode: base.StatusBadRequest,
}, liberrors.ErrServerTracksDifferentProtocols{}
}
res, ssrc, err := ss.s.Handler.(ServerHandlerOnSetup).OnSetup(&ServerHandlerOnSetupCtx{
res, stream, ssrc, err := ss.s.Handler.(ServerHandlerOnSetup).OnSetup(&ServerHandlerOnSetupCtx{
Server: ss.s,
Session: ss,
Conn: sc,
Req: req,
@@ -596,7 +625,16 @@ func (ss *ServerSession) handleRequest(sc *ServerConn, req *base.Request) (*base
})
if res.StatusCode == base.StatusOK {
ss.setupProtocol = &inTH.Protocol
if ss.state == ServerSessionStateInitial {
ss.state = ServerSessionStatePrePlay
ss.setuppedPath = &path
ss.setuppedQuery = &query
ss.setuppedStream = stream
stream.readerAdd(ss, delivery == base.StreamDeliveryMulticast)
}
ss.setuppedProtocol = &inTH.Protocol
ss.setuppedDelivery = &delivery
if ss.setuppedTracks == nil {
ss.setuppedTracks = make(map[int]ServerSessionSetuppedTrack)
@@ -606,27 +644,42 @@ func (ss *ServerSession) handleRequest(sc *ServerConn, req *base.Request) (*base
res.Header = make(base.Header)
}
th := headers.Transport{
Delivery: func() *base.StreamDelivery {
v := base.StreamDeliveryUnicast
return &v
}(),
th := headers.Transport{}
switch {
case delivery == base.StreamDeliveryMulticast:
ss.setuppedTracks[trackID] = ServerSessionSetuppedTrack{}
th.Protocol = base.StreamProtocolUDP
de := base.StreamDeliveryMulticast
th.Delivery = &de
v := uint(127)
th.TTL = &v
d := multicastIP.String()
th.Destination = &d
th.Ports = &[2]int{
stream.multicastListeners[trackID].rtpListener.port(),
stream.multicastListeners[trackID].rtcpListener.port(),
}
if inTH.Protocol == base.StreamProtocolUDP {
case inTH.Protocol == base.StreamProtocolUDP:
ss.setuppedTracks[trackID] = ServerSessionSetuppedTrack{
udpRTPPort: inTH.ClientPorts[0],
udpRTCPPort: inTH.ClientPorts[1],
}
th.Protocol = base.StreamProtocolUDP
de := base.StreamDeliveryUnicast
th.Delivery = &de
th.ClientPorts = inTH.ClientPorts
th.ServerPorts = &[2]int{sc.s.udpRTPListener.port(), sc.s.udpRTCPListener.port()}
} else {
default: // TCP
ss.setuppedTracks[trackID] = ServerSessionSetuppedTrack{}
th.Protocol = base.StreamProtocolTCP
de := base.StreamDeliveryUnicast
th.Delivery = &de
th.InterleavedIDs = inTH.InterleavedIDs
}
@@ -637,12 +690,6 @@ func (ss *ServerSession) handleRequest(sc *ServerConn, req *base.Request) (*base
res.Header["Transport"] = th.Write()
}
if ss.state == ServerSessionStateInitial {
ss.state = ServerSessionStatePrePlay
ss.setupPath = &path
ss.setupQuery = &query
}
// workaround to prevent a bug in rtspclientsink
// that makes impossible for the client to receive the response
// and send frames.
@@ -684,16 +731,6 @@ func (ss *ServerSession) handleRequest(sc *ServerConn, req *base.Request) (*base
path, query := base.PathSplitQuery(pathAndQuery)
// allow to use WriteFrame() before response
if ss.state != ServerSessionStatePlay {
if *ss.setupProtocol == base.StreamProtocolUDP {
ss.udpIP = sc.ip()
ss.udpZone = sc.zone()
} else {
ss.tcpConn = sc
}
}
res, err := sc.s.Handler.(ServerHandlerOnPlay).OnPlay(&ServerHandlerOnPlayCtx{
Session: ss,
Conn: sc,
@@ -706,7 +743,17 @@ func (ss *ServerSession) handleRequest(sc *ServerConn, req *base.Request) (*base
if res.StatusCode == base.StatusOK {
ss.state = ServerSessionStatePlay
if *ss.setupProtocol == base.StreamProtocolUDP {
if *ss.setuppedProtocol == base.StreamProtocolUDP {
ss.udpIP = sc.ip()
ss.udpZone = sc.zone()
} else {
ss.tcpConn = sc
}
ss.setuppedStream.readerSetActive(ss)
if *ss.setuppedProtocol == base.StreamProtocolUDP &&
*ss.setuppedDelivery == base.StreamDeliveryUnicast {
// readers can send RTCP frames, they cannot sent RTP frames
for trackID, track := range ss.setuppedTracks {
sc.s.udpRTCPListener.addClient(ss.udpIP, track.udpRTCPPort, ss, trackID, false)
@@ -721,10 +768,6 @@ func (ss *ServerSession) handleRequest(sc *ServerConn, req *base.Request) (*base
return res, liberrors.ErrServerTCPFramesEnable{}
}
ss.udpIP = nil
ss.udpZone = ""
ss.tcpConn = nil
}
return res, err
@@ -758,7 +801,7 @@ func (ss *ServerSession) handleRequest(sc *ServerConn, req *base.Request) (*base
path, query := base.PathSplitQuery(pathAndQuery)
// allow to use WriteFrame() before response
if *ss.setupProtocol == base.StreamProtocolUDP {
if *ss.setuppedProtocol == base.StreamProtocolUDP {
ss.udpIP = sc.ip()
ss.udpZone = sc.zone()
} else {
@@ -776,7 +819,7 @@ func (ss *ServerSession) handleRequest(sc *ServerConn, req *base.Request) (*base
if res.StatusCode == base.StatusOK {
ss.state = ServerSessionStateRecord
if *ss.setupProtocol == base.StreamProtocolUDP {
if *ss.setuppedProtocol == base.StreamProtocolUDP {
for trackID, track := range ss.setuppedTracks {
ss.s.udpRTPListener.addClient(ss.udpIP, track.udpRTPPort, ss, trackID, true)
ss.s.udpRTCPListener.addClient(ss.udpIP, track.udpRTCPPort, ss, trackID, true)
@@ -836,12 +879,14 @@ func (ss *ServerSession) handleRequest(sc *ServerConn, req *base.Request) (*base
if res.StatusCode == base.StatusOK {
switch ss.state {
case ServerSessionStatePlay:
ss.setuppedStream.readerSetInactive(ss)
ss.state = ServerSessionStatePrePlay
ss.udpIP = nil
ss.udpZone = ""
ss.tcpConn = nil
if *ss.setupProtocol == base.StreamProtocolUDP {
if *ss.setuppedProtocol == base.StreamProtocolUDP {
ss.s.udpRTCPListener.removeClient(ss)
} else {
return res, liberrors.ErrServerTCPFramesDisable{}
@@ -853,7 +898,7 @@ func (ss *ServerSession) handleRequest(sc *ServerConn, req *base.Request) (*base
ss.udpZone = ""
ss.tcpConn = nil
if *ss.setupProtocol == base.StreamProtocolUDP {
if *ss.setuppedProtocol == base.StreamProtocolUDP {
ss.s.udpRTPListener.removeClient(ss)
} else {
return res, liberrors.ErrServerTCPFramesDisable{}
@@ -904,13 +949,14 @@ func (ss *ServerSession) handleRequest(sc *ServerConn, req *base.Request) (*base
}, liberrors.ErrServerUnhandledRequest{Req: req}
}
// WriteFrame writes a frame.
// WriteFrame writes a frame to the session.
func (ss *ServerSession) WriteFrame(trackID int, streamType StreamType, payload []byte) {
if _, ok := ss.SetuppedTracks()[trackID]; !ok {
return
}
if *ss.setupProtocol == base.StreamProtocolUDP {
if *ss.setuppedProtocol == base.StreamProtocolUDP {
if *ss.setuppedDelivery == base.StreamDeliveryUnicast {
track := ss.setuppedTracks[trackID]
if streamType == StreamTypeRTP {
@@ -926,6 +972,7 @@ func (ss *ServerSession) WriteFrame(trackID int, streamType StreamType, payload
Port: track.udpRTCPPort,
})
}
}
} else {
ss.tcpConn.tcpFrameWriteBuffer.Push(&base.InterleavedFrame{
TrackID: trackID,

159
serverstream.go Normal file
View File

@@ -0,0 +1,159 @@
package gortsplib
import (
"net"
"sync"
"github.com/aler9/gortsplib/pkg/base"
)
type listenerPair struct {
rtpListener *serverUDPListener
rtcpListener *serverUDPListener
}
// ServerStream is an entity that allows to send frames to multiple readers.
type ServerStream struct {
s *Server
tracks Tracks
mutex sync.RWMutex
readersUnicast map[*ServerSession]struct{}
readers map[*ServerSession]struct{}
multicastListeners []*listenerPair
}
// NewServerStream allocates a ServerStream.
func NewServerStream(tracks Tracks) *ServerStream {
return &ServerStream{
tracks: tracks,
readersUnicast: make(map[*ServerSession]struct{}),
readers: make(map[*ServerSession]struct{}),
}
}
// Close closes a ServerStream.
func (st *ServerStream) Close() error {
st.mutex.Lock()
defer st.mutex.Unlock()
if st.s != nil {
select {
case st.s.streamRemove <- st:
case <-st.s.ctx.Done():
}
}
if st.multicastListeners != nil {
for _, l := range st.multicastListeners {
l.rtpListener.close()
l.rtcpListener.close()
}
st.multicastListeners = nil
}
for ss := range st.readers {
ss.Close()
}
st.readers = nil
return nil
}
// Tracks returns the tracks of the stream.
func (st *ServerStream) Tracks() Tracks {
return st.tracks
}
func (st *ServerStream) readerAdd(ss *ServerSession, isMulticast bool) {
st.mutex.Lock()
defer st.mutex.Unlock()
st.readers[ss] = struct{}{}
if !isMulticast {
return
}
if st.s == nil {
st.s = ss.s
select {
case st.s.streamAdd <- st:
case <-st.s.ctx.Done():
}
}
if st.multicastListeners != nil {
return
}
st.multicastListeners = make([]*listenerPair, len(st.tracks))
for i := range st.tracks {
rtpListener, rtcpListener := newServerUDPListenerMulticastPair(st.s)
st.multicastListeners[i] = &listenerPair{
rtpListener: rtpListener,
rtcpListener: rtcpListener,
}
}
}
func (st *ServerStream) readerRemove(ss *ServerSession) {
st.mutex.Lock()
defer st.mutex.Unlock()
delete(st.readers, ss)
if len(st.readers) == 0 && st.multicastListeners != nil {
for _, l := range st.multicastListeners {
l.rtpListener.close()
l.rtcpListener.close()
}
}
}
func (st *ServerStream) readerSetActive(ss *ServerSession) {
st.mutex.Lock()
defer st.mutex.Unlock()
if *ss.setuppedDelivery == base.StreamDeliveryUnicast {
st.readersUnicast[ss] = struct{}{}
}
}
func (st *ServerStream) readerSetInactive(ss *ServerSession) {
st.mutex.Lock()
defer st.mutex.Unlock()
if *ss.setuppedDelivery == base.StreamDeliveryUnicast {
delete(st.readersUnicast, ss)
}
}
// WriteFrame writes a frame to all the readers of the stream.
func (st *ServerStream) WriteFrame(trackID int, streamType StreamType, payload []byte) {
st.mutex.RLock()
defer st.mutex.RUnlock()
// send unicast
for r := range st.readersUnicast {
r.WriteFrame(trackID, streamType, payload)
}
// send multicast
if st.multicastListeners != nil {
if streamType == StreamTypeRTP {
st.multicastListeners[trackID].rtpListener.write(payload, &net.UDPAddr{
IP: multicastIP,
Zone: "",
Port: st.multicastListeners[trackID].rtpListener.port(),
})
} else {
st.multicastListeners[trackID].rtcpListener.write(payload, &net.UDPAddr{
IP: multicastIP,
Zone: "",
Port: st.multicastListeners[trackID].rtcpListener.port(),
})
}
}
}

69
serverudpl.go
View File

@@ -2,11 +2,15 @@ package gortsplib
import (
"context"
"math/rand"
"net"
"strconv"
"sync"
"sync/atomic"
"time"
"golang.org/x/net/ipv4"
"github.com/aler9/gortsplib/pkg/multibuffer"
"github.com/aler9/gortsplib/pkg/ringbuffer"
)
@@ -15,6 +19,8 @@ const (
serverConnUDPListenerKernelReadBufferSize = 0x80000 // same as gstreamer's rtspsrc
)
var multicastIP = net.ParseIP("239.0.0.0")
type bufAddrPair struct {
buf []byte
addr *net.UDPAddr
@@ -57,17 +63,76 @@ type serverUDPListener struct {
ringBuffer *ringbuffer.RingBuffer
}
func newServerUDPListenerMulticastPair(s *Server) (*serverUDPListener, *serverUDPListener) {
// choose two consecutive ports in range 65535-10000
// rtp must be even and rtcp odd
for {
rtpPort := (rand.Intn((65535-10000)/2) * 2) + 10000
rtpListener, err := newServerUDPListener(s, true, multicastIP.String()+":"+strconv.FormatInt(int64(rtpPort), 10), StreamTypeRTP)
if err != nil {
continue
}
rtcpPort := rtpPort + 1
rtcpListener, err := newServerUDPListener(s, true, multicastIP.String()+":"+strconv.FormatInt(int64(rtcpPort), 10), StreamTypeRTCP)
if err != nil {
rtpListener.close()
continue
}
return rtpListener, rtcpListener
}
}
func newServerUDPListener(
s *Server,
multicast bool,
address string,
streamType StreamType) (*serverUDPListener, error) {
var pc *net.UDPConn
if multicast {
addr, err := net.ResolveUDPAddr("udp4", address)
if err != nil {
return nil, err
}
tmp, err := net.ListenPacket("udp4", (&net.UDPAddr{
IP: net.ParseIP("224.0.0.0"),
Port: addr.Port,
}).String())
if err != nil {
return nil, err
}
p := ipv4.NewPacketConn(tmp)
err = p.SetTTL(127)
if err != nil {
return nil, err
}
intfs, err := net.Interfaces()
if err != nil {
return nil, err
}
for _, intf := range intfs {
err := p.JoinGroup(&intf, &net.UDPAddr{IP: addr.IP})
if err != nil {
return nil, err
}
}
pc = tmp.(*net.UDPConn)
} else {
tmp, err := s.ListenPacket("udp", address)
if err != nil {
return nil, err
}
pc := tmp.(*net.UDPConn)
pc = tmp.(*net.UDPConn)
}
err = pc.SetReadBuffer(serverConnUDPListenerKernelReadBufferSize)
err := pc.SetReadBuffer(serverConnUDPListenerKernelReadBufferSize)
if err != nil {
return nil, err
}