2019-12-31 12:48:17 +00:00
package main
import (
2020-05-10 20:56:46 +00:00
"errors"
2019-12-31 12:48:17 +00:00
"fmt"
"io"
"net"
2020-07-13 09:51:17 +00:00
"os"
2020-02-15 17:25:29 +00:00
"os/exec"
2020-08-03 15:35:34 +00:00
"strconv"
2019-12-31 12:48:17 +00:00
"strings"
2020-06-21 09:30:30 +00:00
"time"
2019-12-31 12:48:17 +00:00
2020-01-20 09:21:05 +00:00
"github.com/aler9/gortsplib"
2020-07-19 09:51:28 +00:00
"github.com/aler9/sdp/v3"
2019-12-31 12:48:17 +00:00
)
2020-06-21 09:30:30 +00:00
const (
2020-07-13 07:45:45 +00:00
clientCheckStreamInterval = 5 * time . Second
clientReceiverReportInterval = 10 * time . Second
2020-07-20 06:52:14 +00:00
clientTcpReadBufferSize = 128 * 1024
clientTcpWriteBufferSize = 128 * 1024
clientUdpReadBufferSize = 2048
clientUdpWriteBufferSize = 128 * 1024
2020-06-21 09:30:30 +00:00
)
2020-08-05 12:59:38 +00:00
type udpClient struct {
client * client
trackId int
streamType gortsplib . StreamType
}
type udpClientAddr struct {
2020-08-30 13:51:28 +00:00
ip [ net . IPv6len ] byte // use a fixed-size array to enable the equality operator
2020-08-05 12:59:38 +00:00
port int
}
func makeUdpClientAddr ( ip net . IP , port int ) udpClientAddr {
ret := udpClientAddr {
port : port ,
}
if len ( ip ) == net . IPv4len {
copy ( ret . ip [ 0 : ] , [ ] byte { 0 , 0 , 0 , 0 , 0 , 0 , 0 , 0 , 0 , 0 , 0xff , 0xff } ) // v4InV6Prefix
copy ( ret . ip [ 12 : ] , ip )
} else {
copy ( ret . ip [ : ] , ip )
}
return ret
}
2020-07-29 21:30:42 +00:00
type clientTrack struct {
2020-07-19 15:54:31 +00:00
rtpPort int
rtcpPort int
}
2020-08-31 13:31:37 +00:00
type describeRes struct {
sdp [ ] byte
err error
2020-07-12 10:34:35 +00:00
}
2020-07-29 21:30:42 +00:00
type clientState int
2020-01-20 13:41:04 +00:00
const (
2020-07-29 21:30:42 +00:00
clientStateInitial clientState = iota
2020-08-31 13:31:37 +00:00
clientStateWaitDescription
2020-07-13 07:45:45 +00:00
clientStatePrePlay
clientStatePlay
clientStatePreRecord
clientStateRecord
2020-01-20 13:41:04 +00:00
)
2020-07-29 21:30:42 +00:00
func ( cs clientState ) String ( ) string {
2020-05-21 19:46:22 +00:00
switch cs {
2020-07-29 21:30:42 +00:00
case clientStateInitial :
2020-08-31 13:31:37 +00:00
return "Initial"
2020-05-21 19:46:22 +00:00
2020-08-31 13:31:37 +00:00
case clientStateWaitDescription :
return "WaitDescription"
2020-05-21 19:46:22 +00:00
2020-07-13 07:45:45 +00:00
case clientStatePrePlay :
2020-08-31 13:31:37 +00:00
return "PrePlay"
2020-05-21 19:46:22 +00:00
2020-07-13 07:45:45 +00:00
case clientStatePlay :
2020-08-31 13:31:37 +00:00
return "Play"
2020-05-21 19:46:22 +00:00
2020-07-13 07:45:45 +00:00
case clientStatePreRecord :
2020-08-31 13:31:37 +00:00
return "PreRecord"
2020-05-21 19:46:22 +00:00
2020-07-13 07:45:45 +00:00
case clientStateRecord :
2020-08-31 13:31:37 +00:00
return "Record"
2020-05-21 19:46:22 +00:00
}
2020-08-31 13:31:37 +00:00
return "Invalid"
2020-05-21 19:46:22 +00:00
}
2020-07-29 21:30:42 +00:00
type client struct {
p * program
conn * gortsplib . ConnServer
state clientState
2020-08-30 13:51:28 +00:00
path * path
2020-07-29 21:30:42 +00:00
authUser string
authPass string
authHelper * gortsplib . AuthServer
authFailures int
streamProtocol gortsplib . StreamProtocol
2020-08-03 15:35:34 +00:00
streamTracks map [ int ] * clientTrack
2020-07-29 21:30:42 +00:00
rtcpReceivers [ ] * gortsplib . RtcpReceiver
2020-08-31 13:31:37 +00:00
describeCSeq gortsplib . HeaderValue
describeUrl string
2020-07-29 21:30:42 +00:00
2020-08-31 13:31:37 +00:00
describe chan describeRes
tcpFrame chan * gortsplib . InterleavedFrame
terminate chan struct { }
done chan struct { }
2019-12-31 12:48:17 +00:00
}
2020-07-30 11:31:18 +00:00
func newClient ( p * program , nconn net . Conn ) * client {
2020-07-29 21:30:42 +00:00
c := & client {
2020-05-07 20:30:03 +00:00
p : p ,
conn : gortsplib . NewConnServer ( gortsplib . ConnServerConf {
2020-07-12 20:53:22 +00:00
Conn : nconn ,
2020-06-27 15:23:43 +00:00
ReadTimeout : p . conf . ReadTimeout ,
WriteTimeout : p . conf . WriteTimeout ,
2020-05-07 20:30:03 +00:00
} ) ,
2020-08-03 15:35:34 +00:00
state : clientStateInitial ,
streamTracks : make ( map [ int ] * clientTrack ) ,
2020-08-31 13:31:37 +00:00
describe : make ( chan describeRes ) ,
tcpFrame : make ( chan * gortsplib . InterleavedFrame ) ,
terminate : make ( chan struct { } ) ,
2020-08-03 15:35:34 +00:00
done : make ( chan struct { } ) ,
2019-12-31 12:48:17 +00:00
}
2020-05-10 14:23:57 +00:00
go c . run ( )
2019-12-31 12:48:17 +00:00
return c
}
2020-07-29 21:30:42 +00:00
func ( c * client ) log ( format string , args ... interface { } ) {
2020-06-27 12:18:16 +00:00
c . p . log ( "[client %s] " + format , append ( [ ] interface { } { c . conn . NetConn ( ) . RemoteAddr ( ) . String ( ) } , args ... ) ... )
2019-12-31 12:48:17 +00:00
}
2020-07-29 21:30:42 +00:00
func ( c * client ) isPublisher ( ) { }
func ( c * client ) ip ( ) net . IP {
2020-05-03 21:26:41 +00:00
return c . conn . NetConn ( ) . RemoteAddr ( ) . ( * net . TCPAddr ) . IP
}
2020-07-29 21:30:42 +00:00
func ( c * client ) zone ( ) string {
2020-05-03 21:26:41 +00:00
return c . conn . NetConn ( ) . RemoteAddr ( ) . ( * net . TCPAddr ) . Zone
}
2020-08-31 13:31:37 +00:00
var errRunTerminate = errors . New ( "terminate" )
var errRunWaitDescription = errors . New ( "wait description" )
var errRunPlay = errors . New ( "play" )
var errRunRecord = errors . New ( "record" )
2020-07-29 21:30:42 +00:00
func ( c * client ) run ( ) {
2020-07-30 11:31:18 +00:00
var onConnectCmd * exec . Cmd
2020-07-13 09:51:17 +00:00
if c . p . conf . RunOnConnect != "" {
2020-07-30 11:31:18 +00:00
onConnectCmd = exec . Command ( "/bin/sh" , "-c" , c . p . conf . RunOnConnect )
onConnectCmd . Stdout = os . Stdout
onConnectCmd . Stderr = os . Stderr
err := onConnectCmd . Start ( )
2020-02-16 12:04:43 +00:00
if err != nil {
c . log ( "ERR: %s" , err )
}
}
2020-02-15 17:25:29 +00:00
2019-12-31 12:48:17 +00:00
for {
2020-08-31 13:31:37 +00:00
if ! c . runInitial ( ) {
break
2020-01-03 22:05:06 +00:00
}
}
2020-05-10 14:23:57 +00:00
2020-07-30 11:31:18 +00:00
if onConnectCmd != nil {
onConnectCmd . Process . Signal ( os . Interrupt )
onConnectCmd . Wait ( )
2020-07-13 09:51:17 +00:00
}
2020-07-11 14:40:19 +00:00
2020-08-31 13:31:37 +00:00
close ( c . describe )
close ( c . tcpFrame )
close ( c . done )
2020-01-03 22:05:06 +00:00
}
2019-12-31 12:48:17 +00:00
2020-08-30 11:06:26 +00:00
func ( c * client ) writeResError ( cseq gortsplib . HeaderValue , code gortsplib . StatusCode , err error ) {
2020-01-03 22:05:06 +00:00
c . log ( "ERR: %s" , err )
2019-12-31 12:48:17 +00:00
2020-05-03 13:22:41 +00:00
c . conn . WriteResponse ( & gortsplib . Response {
2020-05-03 12:40:06 +00:00
StatusCode : code ,
2020-08-30 11:06:26 +00:00
Header : gortsplib . Header {
"CSeq" : cseq ,
} ,
2020-05-03 12:40:06 +00:00
} )
2019-12-31 12:48:17 +00:00
}
2020-05-10 20:56:46 +00:00
var errAuthCritical = errors . New ( "auth critical" )
var errAuthNotCritical = errors . New ( "auth not critical" )
2020-07-29 21:30:42 +00:00
func ( c * client ) authenticate ( ips [ ] interface { } , user string , pass string , req * gortsplib . Request ) error {
2020-07-10 09:24:57 +00:00
// validate ip
2020-06-15 20:41:14 +00:00
err := func ( ) error {
if ips == nil {
return nil
}
2020-05-10 20:56:46 +00:00
2020-07-10 09:24:57 +00:00
ip := c . ip ( )
2020-07-11 14:40:19 +00:00
if ! ipEqualOrInRange ( ip , ips ) {
c . log ( "ERR: ip '%s' not allowed" , ip )
return errAuthCritical
2020-05-10 20:56:46 +00:00
}
2020-07-11 14:40:19 +00:00
return nil
2020-06-15 20:41:14 +00:00
} ( )
if err != nil {
return err
}
2020-07-10 09:24:57 +00:00
// validate credentials
2020-06-15 20:41:14 +00:00
err = func ( ) error {
if user == "" {
return nil
}
2020-05-10 20:56:46 +00:00
2020-07-10 09:24:57 +00:00
// reset authHelper every time the credentials change
if c . authHelper == nil || c . authUser != user || c . authPass != pass {
c . authUser = user
c . authPass = pass
2020-07-12 11:16:33 +00:00
c . authHelper = gortsplib . NewAuthServer ( user , pass , c . p . conf . authMethodsParsed )
2020-05-10 20:56:46 +00:00
}
2020-07-10 09:24:57 +00:00
err := c . authHelper . ValidateHeader ( req . Header [ "Authorization" ] , req . Method , req . Url )
2020-06-15 20:41:14 +00:00
if err != nil {
2020-07-10 09:24:57 +00:00
c . authFailures += 1
// vlc with login prompt sends 4 requests:
// 1) without credentials
// 2) with password but without the username
// 3) without credentials
// 4) with password and username
// hence we must allow up to 3 failures
var retErr error
if c . authFailures > 3 {
2020-06-15 20:41:14 +00:00
c . log ( "ERR: unauthorized: %s" , err )
2020-07-10 09:24:57 +00:00
retErr = errAuthCritical
} else if c . authFailures > 1 {
c . log ( "WARN: unauthorized: %s" , err )
retErr = errAuthNotCritical
} else {
retErr = errAuthNotCritical
2020-06-15 20:41:14 +00:00
}
c . conn . WriteResponse ( & gortsplib . Response {
StatusCode : gortsplib . StatusUnauthorized ,
Header : gortsplib . Header {
2020-06-27 19:22:50 +00:00
"CSeq" : req . Header [ "CSeq" ] ,
2020-07-10 09:24:57 +00:00
"WWW-Authenticate" : c . authHelper . GenerateHeader ( ) ,
2020-06-15 20:41:14 +00:00
} ,
} )
2020-07-10 09:24:57 +00:00
return retErr
2020-06-15 20:41:14 +00:00
}
2020-06-21 09:30:30 +00:00
2020-07-10 09:24:57 +00:00
// reset authFailures after a successful login
c . authFailures = 0
2020-06-15 20:41:14 +00:00
return nil
} ( )
if err != nil {
return err
2020-05-10 20:56:46 +00:00
}
return nil
}
2020-08-31 13:31:37 +00:00
func ( c * client ) handleRequest ( req * gortsplib . Request ) error {
2020-05-03 14:07:56 +00:00
c . log ( string ( req . Method ) )
2020-01-03 22:05:06 +00:00
2020-01-26 11:41:26 +00:00
cseq , ok := req . Header [ "CSeq" ]
if ! ok || len ( cseq ) != 1 {
2020-08-30 11:06:26 +00:00
c . writeResError ( nil , gortsplib . StatusBadRequest , fmt . Errorf ( "cseq missing" ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2019-12-31 12:48:17 +00:00
}
2020-08-31 13:31:37 +00:00
pathName := req . Url . Path
if len ( pathName ) < 1 || pathName [ 0 ] != '/' {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "path must begin with a slash" ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2020-08-03 14:56:45 +00:00
}
2020-08-31 13:31:37 +00:00
pathName = pathName [ 1 : ] // strip leading slash
2019-12-31 13:55:46 +00:00
2019-12-31 12:48:17 +00:00
switch req . Method {
2020-05-03 14:07:56 +00:00
case gortsplib . OPTIONS :
2020-05-03 13:22:41 +00:00
c . conn . WriteResponse ( & gortsplib . Response {
2020-05-03 12:40:06 +00:00
StatusCode : gortsplib . StatusOK ,
2020-01-26 11:41:26 +00:00
Header : gortsplib . Header {
2020-06-27 11:38:35 +00:00
"CSeq" : cseq ,
2020-07-18 11:48:09 +00:00
"Public" : gortsplib . HeaderValue { strings . Join ( [ ] string {
2020-05-03 14:07:56 +00:00
string ( gortsplib . DESCRIBE ) ,
string ( gortsplib . ANNOUNCE ) ,
string ( gortsplib . SETUP ) ,
string ( gortsplib . PLAY ) ,
string ( gortsplib . RECORD ) ,
string ( gortsplib . TEARDOWN ) ,
2020-01-26 11:41:26 +00:00
} , ", " ) } ,
2019-12-31 12:48:17 +00:00
} ,
2020-01-03 22:05:06 +00:00
} )
2020-08-31 13:31:37 +00:00
return nil
2019-12-31 12:48:17 +00:00
2020-05-03 14:07:56 +00:00
case gortsplib . DESCRIBE :
2020-07-29 21:30:42 +00:00
if c . state != clientStateInitial {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest ,
2020-07-29 21:30:42 +00:00
fmt . Errorf ( "client is in state '%s' instead of '%s'" , c . state , clientStateInitial ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2019-12-31 12:48:17 +00:00
}
2020-08-31 13:31:37 +00:00
confp := c . p . findConfForPathName ( pathName )
2020-07-30 11:31:18 +00:00
if confp == nil {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest ,
2020-08-31 13:31:37 +00:00
fmt . Errorf ( "unable to find a valid configuration for path '%s'" , pathName ) )
return errRunTerminate
2020-06-27 19:22:50 +00:00
}
2020-07-30 11:31:18 +00:00
err := c . authenticate ( confp . readIpsParsed , confp . ReadUser , confp . ReadPass , req )
2020-05-10 20:56:46 +00:00
if err != nil {
if err == errAuthCritical {
2020-08-31 13:31:37 +00:00
return errRunTerminate
2020-05-10 20:56:46 +00:00
}
2020-08-31 13:31:37 +00:00
return nil
2020-05-10 20:56:46 +00:00
}
2020-08-31 13:31:37 +00:00
c . p . events <- programEventClientDescribe { c , pathName }
2019-12-31 12:48:17 +00:00
2020-08-31 13:31:37 +00:00
c . describeCSeq = cseq
c . describeUrl = req . Url . String ( )
return errRunWaitDescription
2019-12-31 12:48:17 +00:00
2020-05-03 14:07:56 +00:00
case gortsplib . ANNOUNCE :
2020-07-29 21:30:42 +00:00
if c . state != clientStateInitial {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest ,
2020-07-29 21:30:42 +00:00
fmt . Errorf ( "client is in state '%s' instead of '%s'" , c . state , clientStateInitial ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2019-12-31 12:48:17 +00:00
}
2020-08-31 13:31:37 +00:00
if len ( pathName ) == 0 {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "empty base path" ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2020-08-03 14:56:45 +00:00
}
2020-08-31 13:31:37 +00:00
err := checkPathName ( pathName )
2020-08-05 10:17:04 +00:00
if err != nil {
2020-08-31 13:31:37 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "invalid path name: %s (%s)" , err , pathName ) )
return errRunTerminate
2020-07-12 15:24:12 +00:00
}
2020-08-31 13:31:37 +00:00
confp := c . p . findConfForPathName ( pathName )
2020-07-30 11:31:18 +00:00
if confp == nil {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest ,
2020-08-31 13:31:37 +00:00
fmt . Errorf ( "unable to find a valid configuration for path '%s'" , pathName ) )
return errRunTerminate
2020-06-27 19:22:50 +00:00
}
2020-08-05 10:17:04 +00:00
err = c . authenticate ( confp . publishIpsParsed , confp . PublishUser , confp . PublishPass , req )
2020-05-10 20:56:46 +00:00
if err != nil {
if err == errAuthCritical {
2020-08-31 13:31:37 +00:00
return errRunTerminate
2020-02-16 15:05:08 +00:00
}
2020-08-31 13:31:37 +00:00
return nil
2020-02-16 15:05:08 +00:00
}
2020-01-26 11:41:26 +00:00
ct , ok := req . Header [ "Content-Type" ]
if ! ok || len ( ct ) != 1 {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "Content-Type header missing" ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2019-12-31 12:48:17 +00:00
}
2020-01-26 11:41:26 +00:00
if ct [ 0 ] != "application/sdp" {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "unsupported Content-Type '%s'" , ct ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2019-12-31 12:48:17 +00:00
}
2020-07-12 15:24:12 +00:00
sdpParsed := & sdp . SessionDescription { }
2020-07-19 09:51:28 +00:00
err = sdpParsed . Unmarshal ( req . Content )
2019-12-31 12:48:17 +00:00
if err != nil {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "invalid SDP: %s" , err ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2019-12-31 12:48:17 +00:00
}
2020-01-21 09:08:53 +00:00
2020-07-12 15:24:12 +00:00
if len ( sdpParsed . MediaDescriptions ) == 0 {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "no tracks defined" ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2020-06-27 11:38:35 +00:00
}
2019-12-31 12:48:17 +00:00
2020-07-18 11:48:09 +00:00
var tracks [ ] * gortsplib . Track
for i , media := range sdpParsed . MediaDescriptions {
tracks = append ( tracks , & gortsplib . Track {
Id : i ,
Media : media ,
} )
}
sdpParsed , req . Content = sdpForServer ( tracks )
2020-06-27 11:38:35 +00:00
res := make ( chan error )
2020-08-31 13:31:37 +00:00
c . p . events <- programEventClientAnnounce { res , c , pathName , req . Content , sdpParsed }
2020-06-27 11:38:35 +00:00
err = <- res
2019-12-31 12:48:17 +00:00
if err != nil {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , err )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2019-12-31 12:48:17 +00:00
}
2020-05-03 13:22:41 +00:00
c . conn . WriteResponse ( & gortsplib . Response {
2020-05-03 12:40:06 +00:00
StatusCode : gortsplib . StatusOK ,
2020-01-26 11:41:26 +00:00
Header : gortsplib . Header {
2020-06-27 11:38:35 +00:00
"CSeq" : cseq ,
2019-12-31 12:48:17 +00:00
} ,
2020-01-03 22:05:06 +00:00
} )
2020-08-31 13:31:37 +00:00
return nil
2019-12-31 12:48:17 +00:00
2020-05-03 14:07:56 +00:00
case gortsplib . SETUP :
2020-07-18 11:48:09 +00:00
th , err := gortsplib . ReadHeaderTransport ( req . Header [ "Transport" ] )
if err != nil {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "transport header: %s" , err ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2019-12-31 12:48:17 +00:00
}
2020-06-14 15:46:27 +00:00
if _ , ok := th [ "multicast" ] ; ok {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "multicast is not supported" ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2019-12-31 12:48:17 +00:00
}
2020-08-31 13:31:37 +00:00
basePath , controlPath , err := splitPath ( pathName )
2020-08-03 14:56:45 +00:00
if err != nil {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , err )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2020-08-03 14:56:45 +00:00
}
2019-12-31 12:48:17 +00:00
switch c . state {
// play
2020-07-29 21:30:42 +00:00
case clientStateInitial , clientStatePrePlay :
2020-08-30 11:20:18 +00:00
confp := c . p . findConfForPathName ( basePath )
2020-07-30 11:31:18 +00:00
if confp == nil {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest ,
2020-08-03 14:56:45 +00:00
fmt . Errorf ( "unable to find a valid configuration for path '%s'" , basePath ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2020-06-27 19:22:50 +00:00
}
2020-07-30 11:31:18 +00:00
err := c . authenticate ( confp . readIpsParsed , confp . ReadUser , confp . ReadPass , req )
2020-05-10 20:56:46 +00:00
if err != nil {
if err == errAuthCritical {
2020-08-31 13:31:37 +00:00
return errRunTerminate
2020-05-10 20:56:46 +00:00
}
2020-08-31 13:31:37 +00:00
return nil
2020-05-10 20:56:46 +00:00
}
2020-08-30 13:51:28 +00:00
if c . path != nil && basePath != c . path . name {
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "path has changed, was '%s', now is '%s'" , c . path . name , basePath ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2020-08-03 15:35:34 +00:00
}
if ! strings . HasPrefix ( controlPath , "trackID=" ) {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "invalid control path (%s)" , controlPath ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2020-08-03 15:35:34 +00:00
}
tmp , err := strconv . ParseInt ( controlPath [ len ( "trackID=" ) : ] , 10 , 64 )
if err != nil || tmp < 0 {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "invalid track id (%s)" , controlPath ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2020-08-03 15:35:34 +00:00
}
trackId := int ( tmp )
if _ , ok := c . streamTracks [ trackId ] ; ok {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "track %d has already been setup" , trackId ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2020-08-03 15:35:34 +00:00
}
2019-12-31 12:48:17 +00:00
// play via UDP
2020-08-31 13:31:37 +00:00
if th . IsUdp ( ) {
2020-07-19 15:54:31 +00:00
if _ , ok := c . p . conf . protocolsParsed [ gortsplib . StreamProtocolUdp ] ; ! ok {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusUnsupportedTransport , fmt . Errorf ( "UDP streaming is disabled" ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2020-01-15 21:48:51 +00:00
}
2020-08-03 15:35:34 +00:00
if len ( c . streamTracks ) > 0 && c . streamProtocol != gortsplib . StreamProtocolUdp {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "can't receive tracks with different protocols" ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2019-12-31 13:55:46 +00:00
}
2020-08-03 15:35:34 +00:00
rtpPort , rtcpPort := th . Ports ( "client_port" )
if rtpPort == 0 || rtcpPort == 0 {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "transport header does not have valid client ports (%v)" , req . Header [ "Transport" ] ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2020-06-27 11:38:35 +00:00
}
2019-12-31 12:48:17 +00:00
2020-06-27 11:38:35 +00:00
res := make ( chan error )
2020-08-03 15:35:34 +00:00
c . p . events <- programEventClientSetupPlay { res , c , basePath , trackId }
2020-06-27 11:38:35 +00:00
err = <- res
2019-12-31 12:48:17 +00:00
if err != nil {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , err )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2019-12-31 12:48:17 +00:00
}
2020-08-03 15:35:34 +00:00
c . streamProtocol = gortsplib . StreamProtocolUdp
c . streamTracks [ trackId ] = & clientTrack {
rtpPort : rtpPort ,
rtcpPort : rtcpPort ,
}
2020-05-03 13:22:41 +00:00
c . conn . WriteResponse ( & gortsplib . Response {
2020-05-03 12:40:06 +00:00
StatusCode : gortsplib . StatusOK ,
2020-01-26 11:41:26 +00:00
Header : gortsplib . Header {
2020-06-27 11:38:35 +00:00
"CSeq" : cseq ,
2020-07-18 11:48:09 +00:00
"Transport" : gortsplib . HeaderValue { strings . Join ( [ ] string {
2020-01-14 20:33:53 +00:00
"RTP/AVP/UDP" ,
2019-12-31 12:48:17 +00:00
"unicast" ,
fmt . Sprintf ( "client_port=%d-%d" , rtpPort , rtcpPort ) ,
2020-06-27 15:23:43 +00:00
fmt . Sprintf ( "server_port=%d-%d" , c . p . conf . RtpPort , c . p . conf . RtcpPort ) ,
2020-01-26 11:41:26 +00:00
} , ";" ) } ,
2020-07-18 11:48:09 +00:00
"Session" : gortsplib . HeaderValue { "12345678" } ,
2019-12-31 12:48:17 +00:00
} ,
2020-01-03 22:05:06 +00:00
} )
2020-08-31 13:31:37 +00:00
return nil
2019-12-31 12:48:17 +00:00
// play via TCP
2020-08-31 13:31:37 +00:00
} else if th . IsTcp ( ) {
2020-07-19 15:54:31 +00:00
if _ , ok := c . p . conf . protocolsParsed [ gortsplib . StreamProtocolTcp ] ; ! ok {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusUnsupportedTransport , fmt . Errorf ( "TCP streaming is disabled" ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2020-01-15 21:48:51 +00:00
}
2020-07-19 15:54:31 +00:00
if len ( c . streamTracks ) > 0 && c . streamProtocol != gortsplib . StreamProtocolTcp {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "can't receive tracks with different protocols" ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2020-06-27 11:38:35 +00:00
}
2019-12-31 12:48:17 +00:00
2020-06-27 11:38:35 +00:00
res := make ( chan error )
2020-08-03 15:35:34 +00:00
c . p . events <- programEventClientSetupPlay { res , c , basePath , trackId }
2020-06-27 11:38:35 +00:00
err = <- res
2019-12-31 12:48:17 +00:00
if err != nil {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , err )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2019-12-31 12:48:17 +00:00
}
2020-08-03 15:35:34 +00:00
c . streamProtocol = gortsplib . StreamProtocolTcp
c . streamTracks [ trackId ] = & clientTrack {
rtpPort : 0 ,
rtcpPort : 0 ,
}
interleaved := fmt . Sprintf ( "%d-%d" , ( ( trackId ) * 2 ) , ( ( trackId ) * 2 ) + 1 )
2019-12-31 12:48:17 +00:00
2020-05-03 13:22:41 +00:00
c . conn . WriteResponse ( & gortsplib . Response {
2020-05-03 12:40:06 +00:00
StatusCode : gortsplib . StatusOK ,
2020-01-26 11:41:26 +00:00
Header : gortsplib . Header {
2020-06-27 11:38:35 +00:00
"CSeq" : cseq ,
2020-07-18 11:48:09 +00:00
"Transport" : gortsplib . HeaderValue { strings . Join ( [ ] string {
2019-12-31 12:48:17 +00:00
"RTP/AVP/TCP" ,
"unicast" ,
fmt . Sprintf ( "interleaved=%s" , interleaved ) ,
2020-01-26 11:41:26 +00:00
} , ";" ) } ,
2020-07-18 11:48:09 +00:00
"Session" : gortsplib . HeaderValue { "12345678" } ,
2019-12-31 12:48:17 +00:00
} ,
2020-01-03 22:05:06 +00:00
} )
2020-08-31 13:31:37 +00:00
return nil
2019-12-31 12:48:17 +00:00
} else {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "transport header does not contain a valid protocol (RTP/AVP, RTP/AVP/UDP or RTP/AVP/TCP) (%s)" , req . Header [ "Transport" ] ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2019-12-31 12:48:17 +00:00
}
// record
2020-08-31 13:31:37 +00:00
case clientStatePreRecord :
2020-07-18 11:53:24 +00:00
if strings . ToLower ( th . Value ( "mode" ) ) != "record" {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "transport header does not contain mode=record" ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2019-12-31 12:48:17 +00:00
}
2020-08-30 13:51:28 +00:00
// after ANNOUNCE, c.path is already set
if basePath != c . path . name {
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "path has changed, was '%s', now is '%s'" , c . path . name , basePath ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2019-12-31 13:55:46 +00:00
}
2019-12-31 12:48:17 +00:00
// record via UDP
2020-08-31 13:31:37 +00:00
if th . IsUdp ( ) {
2020-07-19 15:54:31 +00:00
if _ , ok := c . p . conf . protocolsParsed [ gortsplib . StreamProtocolUdp ] ; ! ok {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusUnsupportedTransport , fmt . Errorf ( "UDP streaming is disabled" ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2020-01-15 21:48:51 +00:00
}
2020-08-03 15:35:34 +00:00
if len ( c . streamTracks ) > 0 && c . streamProtocol != gortsplib . StreamProtocolUdp {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "can't publish tracks with different protocols" ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2019-12-31 12:48:17 +00:00
}
2020-08-03 15:35:34 +00:00
rtpPort , rtcpPort := th . Ports ( "client_port" )
if rtpPort == 0 || rtcpPort == 0 {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "transport header does not have valid client ports (%s)" , req . Header [ "Transport" ] ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2020-06-27 11:38:35 +00:00
}
2019-12-31 12:48:17 +00:00
2020-08-30 13:51:28 +00:00
if len ( c . streamTracks ) >= len ( c . path . publisherSdpParsed . MediaDescriptions ) {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "all the tracks have already been setup" ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2019-12-31 12:48:17 +00:00
}
2020-08-03 15:35:34 +00:00
c . streamProtocol = gortsplib . StreamProtocolUdp
c . streamTracks [ len ( c . streamTracks ) ] = & clientTrack {
rtpPort : rtpPort ,
rtcpPort : rtcpPort ,
}
2020-05-03 13:22:41 +00:00
c . conn . WriteResponse ( & gortsplib . Response {
2020-05-03 12:40:06 +00:00
StatusCode : gortsplib . StatusOK ,
2020-01-26 11:41:26 +00:00
Header : gortsplib . Header {
2020-06-27 11:38:35 +00:00
"CSeq" : cseq ,
2020-07-18 11:48:09 +00:00
"Transport" : gortsplib . HeaderValue { strings . Join ( [ ] string {
2020-01-14 20:33:53 +00:00
"RTP/AVP/UDP" ,
2019-12-31 12:48:17 +00:00
"unicast" ,
fmt . Sprintf ( "client_port=%d-%d" , rtpPort , rtcpPort ) ,
2020-06-27 15:23:43 +00:00
fmt . Sprintf ( "server_port=%d-%d" , c . p . conf . RtpPort , c . p . conf . RtcpPort ) ,
2020-01-26 11:41:26 +00:00
} , ";" ) } ,
2020-07-18 11:48:09 +00:00
"Session" : gortsplib . HeaderValue { "12345678" } ,
2019-12-31 12:48:17 +00:00
} ,
2020-01-03 22:05:06 +00:00
} )
2020-08-31 13:31:37 +00:00
return nil
2019-12-31 12:48:17 +00:00
// record via TCP
2020-08-31 13:31:37 +00:00
} else if th . IsTcp ( ) {
2020-07-19 15:54:31 +00:00
if _ , ok := c . p . conf . protocolsParsed [ gortsplib . StreamProtocolTcp ] ; ! ok {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusUnsupportedTransport , fmt . Errorf ( "TCP streaming is disabled" ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2020-01-15 21:48:51 +00:00
}
2020-07-19 15:54:31 +00:00
if len ( c . streamTracks ) > 0 && c . streamProtocol != gortsplib . StreamProtocolTcp {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "can't publish tracks with different protocols" ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2020-06-27 11:38:35 +00:00
}
2019-12-31 12:48:17 +00:00
2020-07-18 11:53:24 +00:00
interleaved := th . Value ( "interleaved" )
2020-06-27 11:38:35 +00:00
if interleaved == "" {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "transport header does not contain the interleaved field" ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2020-06-27 11:38:35 +00:00
}
2019-12-31 12:48:17 +00:00
2020-06-27 11:38:35 +00:00
expInterleaved := fmt . Sprintf ( "%d-%d" , 0 + len ( c . streamTracks ) * 2 , 1 + len ( c . streamTracks ) * 2 )
if interleaved != expInterleaved {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "wrong interleaved value, expected '%s', got '%s'" , expInterleaved , interleaved ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2020-06-27 11:38:35 +00:00
}
2019-12-31 12:48:17 +00:00
2020-08-30 13:51:28 +00:00
if len ( c . streamTracks ) >= len ( c . path . publisherSdpParsed . MediaDescriptions ) {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "all the tracks have already been setup" ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2019-12-31 12:48:17 +00:00
}
2020-08-03 15:35:34 +00:00
c . streamProtocol = gortsplib . StreamProtocolTcp
c . streamTracks [ len ( c . streamTracks ) ] = & clientTrack {
rtpPort : 0 ,
rtcpPort : 0 ,
}
2020-05-03 13:22:41 +00:00
c . conn . WriteResponse ( & gortsplib . Response {
2020-05-03 12:40:06 +00:00
StatusCode : gortsplib . StatusOK ,
2020-01-26 11:41:26 +00:00
Header : gortsplib . Header {
2020-06-27 11:38:35 +00:00
"CSeq" : cseq ,
2020-07-18 11:48:09 +00:00
"Transport" : gortsplib . HeaderValue { strings . Join ( [ ] string {
2019-12-31 12:48:17 +00:00
"RTP/AVP/TCP" ,
"unicast" ,
fmt . Sprintf ( "interleaved=%s" , interleaved ) ,
2020-01-26 11:41:26 +00:00
} , ";" ) } ,
2020-07-18 11:48:09 +00:00
"Session" : gortsplib . HeaderValue { "12345678" } ,
2019-12-31 12:48:17 +00:00
} ,
2020-01-03 22:05:06 +00:00
} )
2020-08-31 13:31:37 +00:00
return nil
2019-12-31 12:48:17 +00:00
} else {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "transport header does not contain a valid protocol (RTP/AVP, RTP/AVP/UDP or RTP/AVP/TCP) (%s)" , req . Header [ "Transport" ] ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2019-12-31 12:48:17 +00:00
}
default :
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "client is in state '%s'" , c . state ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2019-12-31 12:48:17 +00:00
}
2020-05-03 14:07:56 +00:00
case gortsplib . PLAY :
2020-07-13 07:45:45 +00:00
if c . state != clientStatePrePlay {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest ,
2020-07-13 07:45:45 +00:00
fmt . Errorf ( "client is in state '%s' instead of '%s'" , c . state , clientStatePrePlay ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2019-12-31 12:48:17 +00:00
}
2020-08-03 14:56:45 +00:00
// path can end with a slash, remove it
2020-08-31 13:31:37 +00:00
pathName = strings . TrimSuffix ( pathName , "/" )
2020-08-03 14:56:45 +00:00
2020-08-31 13:31:37 +00:00
if pathName != c . path . name {
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "path has changed, was '%s', now is '%s'" , c . path . name , pathName ) )
return errRunTerminate
2019-12-31 13:55:46 +00:00
}
2020-08-30 13:51:28 +00:00
if len ( c . streamTracks ) == 0 {
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "no tracks have been setup" ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2019-12-31 12:48:17 +00:00
}
2020-06-27 11:38:35 +00:00
// write response before setting state
// otherwise, in case of TCP connections, RTP packets could be sent
2020-01-03 22:05:06 +00:00
// before the response
2020-05-03 13:22:41 +00:00
c . conn . WriteResponse ( & gortsplib . Response {
2020-05-03 12:40:06 +00:00
StatusCode : gortsplib . StatusOK ,
2020-01-26 11:41:26 +00:00
Header : gortsplib . Header {
2020-06-27 11:38:35 +00:00
"CSeq" : cseq ,
2020-07-18 11:48:09 +00:00
"Session" : gortsplib . HeaderValue { "12345678" } ,
2019-12-31 12:48:17 +00:00
} ,
2020-01-03 22:05:06 +00:00
} )
2020-08-31 13:31:37 +00:00
return errRunPlay
2019-12-31 12:48:17 +00:00
2020-05-03 14:07:56 +00:00
case gortsplib . RECORD :
2020-07-13 07:45:45 +00:00
if c . state != clientStatePreRecord {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest ,
2020-07-13 07:45:45 +00:00
fmt . Errorf ( "client is in state '%s' instead of '%s'" , c . state , clientStatePreRecord ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2019-12-31 12:48:17 +00:00
}
2020-08-03 14:56:45 +00:00
// path can end with a slash, remove it
2020-08-31 13:31:37 +00:00
pathName = strings . TrimSuffix ( pathName , "/" )
2020-08-03 14:56:45 +00:00
2020-08-31 13:31:37 +00:00
if pathName != c . path . name {
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "path has changed, was '%s', now is '%s'" , c . path . name , pathName ) )
return errRunTerminate
2019-12-31 13:55:46 +00:00
}
2020-08-30 13:51:28 +00:00
if len ( c . streamTracks ) != len ( c . path . publisherSdpParsed . MediaDescriptions ) {
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "not all tracks have been setup" ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
2019-12-31 12:48:17 +00:00
}
2020-05-03 13:22:41 +00:00
c . conn . WriteResponse ( & gortsplib . Response {
2020-05-03 12:40:06 +00:00
StatusCode : gortsplib . StatusOK ,
2020-01-26 11:41:26 +00:00
Header : gortsplib . Header {
2020-06-27 11:38:35 +00:00
"CSeq" : cseq ,
2020-07-18 11:48:09 +00:00
"Session" : gortsplib . HeaderValue { "12345678" } ,
2019-12-31 12:48:17 +00:00
} ,
2020-01-03 22:05:06 +00:00
} )
2020-08-31 13:31:37 +00:00
return errRunRecord
2020-07-13 09:12:20 +00:00
case gortsplib . TEARDOWN :
// close connection silently
2020-08-31 13:31:37 +00:00
return errRunTerminate
2020-07-13 09:12:20 +00:00
default :
2020-08-30 11:06:26 +00:00
c . writeResError ( cseq , gortsplib . StatusBadRequest , fmt . Errorf ( "unhandled method '%s'" , req . Method ) )
2020-08-31 13:31:37 +00:00
return errRunTerminate
}
}
func ( c * client ) runInitial ( ) bool {
readDone := make ( chan error )
go func ( ) {
for {
req , err := c . conn . ReadRequest ( )
if err != nil {
readDone <- err
break
}
err = c . handleRequest ( req )
if err != nil {
readDone <- err
break
}
}
} ( )
select {
case err := <- readDone :
switch err {
case errRunWaitDescription :
return c . runWaitDescription ( )
case errRunPlay :
return c . runPlay ( )
case errRunRecord :
return c . runRecord ( )
default :
c . conn . Close ( )
if err != io . EOF && err != errRunTerminate {
c . log ( "ERR: %s" , err )
}
c . p . events <- programEventClientClose { c }
<- c . terminate
return false
}
case <- c . terminate :
c . conn . Close ( )
<- readDone
2020-07-13 09:12:20 +00:00
return false
}
}
2020-08-31 13:31:37 +00:00
func ( c * client ) runWaitDescription ( ) bool {
select {
case res := <- c . describe :
if res . err != nil {
c . writeResError ( c . describeCSeq , gortsplib . StatusNotFound , res . err )
return true
}
c . conn . WriteResponse ( & gortsplib . Response {
StatusCode : gortsplib . StatusOK ,
Header : gortsplib . Header {
"CSeq" : c . describeCSeq ,
"Content-Base" : gortsplib . HeaderValue { c . describeUrl + "/" } ,
"Content-Type" : gortsplib . HeaderValue { "application/sdp" } ,
} ,
Content : res . sdp ,
} )
return true
2020-07-13 09:51:17 +00:00
2020-08-31 13:31:37 +00:00
case <- c . terminate :
c . conn . Close ( )
return false
2020-07-13 09:12:20 +00:00
}
2020-08-31 13:31:37 +00:00
}
2020-07-13 09:12:20 +00:00
2020-08-31 13:31:37 +00:00
func ( c * client ) runPlay ( ) bool {
2020-08-03 15:35:34 +00:00
// start sending frames only after sending the response to the PLAY request
2020-08-31 13:31:37 +00:00
c . p . events <- programEventClientPlay { c }
2020-07-13 09:12:20 +00:00
2020-08-30 13:51:28 +00:00
c . log ( "is receiving on path '%s', %d %s via %s" , c . path . name , len ( c . streamTracks ) , func ( ) string {
2020-07-13 09:12:20 +00:00
if len ( c . streamTracks ) == 1 {
return "track"
2020-07-12 10:34:35 +00:00
}
2020-07-13 09:12:20 +00:00
return "tracks"
} ( ) , c . streamProtocol )
2020-07-12 10:34:35 +00:00
2020-07-30 11:31:18 +00:00
var onReadCmd * exec . Cmd
2020-08-31 13:31:37 +00:00
if c . path . confp . RunOnRead != "" {
onReadCmd = exec . Command ( "/bin/sh" , "-c" , c . path . confp . RunOnRead )
2020-08-04 20:11:16 +00:00
onReadCmd . Env = append ( os . Environ ( ) ,
2020-08-31 13:31:37 +00:00
"RTSP_SERVER_PATH=" + c . path . name ,
2020-08-04 20:11:16 +00:00
)
2020-07-30 11:31:18 +00:00
onReadCmd . Stdout = os . Stdout
onReadCmd . Stderr = os . Stderr
err := onReadCmd . Start ( )
2020-07-13 09:51:17 +00:00
if err != nil {
c . log ( "ERR: %s" , err )
}
}
2020-07-28 13:11:37 +00:00
if c . streamProtocol == gortsplib . StreamProtocolUdp {
2020-08-31 13:31:37 +00:00
c . runPlayUdp ( )
} else {
c . runPlayTcp ( )
}
if onReadCmd != nil {
onReadCmd . Process . Signal ( os . Interrupt )
onReadCmd . Wait ( )
}
return false
}
func ( c * client ) runPlayUdp ( ) {
readDone := make ( chan error )
go func ( ) {
2020-07-28 13:11:37 +00:00
for {
req , err := c . conn . ReadRequest ( )
if err != nil {
2020-08-31 13:31:37 +00:00
readDone <- err
2020-07-28 13:11:37 +00:00
break
}
2020-08-31 13:31:37 +00:00
err = c . handleRequest ( req )
if err != nil {
readDone <- err
2020-07-28 13:11:37 +00:00
break
}
}
2020-08-31 13:31:37 +00:00
} ( )
2020-07-28 13:11:37 +00:00
2020-08-31 13:31:37 +00:00
select {
case err := <- readDone :
c . conn . Close ( )
if err != io . EOF && err != errRunTerminate {
c . log ( "ERR: %s" , err )
}
c . p . events <- programEventClientClose { c }
<- c . terminate
return
case <- c . terminate :
c . conn . Close ( )
<- readDone
return
}
}
2020-08-29 13:21:56 +00:00
2020-08-31 13:31:37 +00:00
func ( c * client ) runPlayTcp ( ) {
readDone := make ( chan error )
go func ( ) {
frame := & gortsplib . InterleavedFrame { }
readBuf := make ( [ ] byte , clientTcpReadBufferSize )
2020-08-29 13:21:56 +00:00
2020-08-31 13:31:37 +00:00
for {
frame . Content = readBuf
frame . Content = frame . Content [ : cap ( frame . Content ) ]
2020-08-29 13:21:56 +00:00
2020-08-31 13:31:37 +00:00
recv , err := c . conn . ReadFrameOrRequest ( frame , false )
if err != nil {
readDone <- err
break
2020-07-13 09:12:20 +00:00
}
2020-08-31 13:31:37 +00:00
switch recvt := recv . ( type ) {
case * gortsplib . InterleavedFrame :
// rtcp feedback is handled by gortsplib
2020-06-27 12:18:16 +00:00
2020-08-31 13:31:37 +00:00
case * gortsplib . Request :
err := c . handleRequest ( recvt )
if err != nil {
readDone <- err
break
2020-07-13 09:12:20 +00:00
}
2020-01-03 22:05:06 +00:00
}
2020-07-13 09:12:20 +00:00
}
2020-08-31 13:31:37 +00:00
} ( )
2020-01-03 22:05:06 +00:00
2020-08-31 13:31:37 +00:00
for {
select {
case err := <- readDone :
c . conn . Close ( )
if err != io . EOF && err != errRunTerminate {
c . log ( "ERR: %s" , err )
2020-07-13 09:12:20 +00:00
}
2020-08-31 13:31:37 +00:00
go func ( ) {
for range c . tcpFrame {
}
} ( )
c . p . events <- programEventClientClose { c }
<- c . terminate
return
case frame := <- c . tcpFrame :
c . conn . WriteFrame ( frame )
case <- c . terminate :
c . conn . Close ( )
<- readDone
return
}
2020-07-13 09:51:17 +00:00
}
2020-07-13 09:12:20 +00:00
}
2020-08-31 13:31:37 +00:00
func ( c * client ) runRecord ( ) bool {
2020-07-19 15:54:31 +00:00
c . rtcpReceivers = make ( [ ] * gortsplib . RtcpReceiver , len ( c . streamTracks ) )
2020-07-13 09:12:20 +00:00
for trackId := range c . streamTracks {
2020-07-19 15:54:31 +00:00
c . rtcpReceivers [ trackId ] = gortsplib . NewRtcpReceiver ( )
2020-07-13 09:12:20 +00:00
}
2020-08-31 13:31:37 +00:00
c . p . events <- programEventClientRecord { c }
2020-07-13 09:12:20 +00:00
2020-08-30 13:51:28 +00:00
c . log ( "is publishing on path '%s', %d %s via %s" , c . path . name , len ( c . streamTracks ) , func ( ) string {
2020-07-13 09:12:20 +00:00
if len ( c . streamTracks ) == 1 {
return "track"
}
return "tracks"
} ( ) , c . streamProtocol )
2020-07-30 11:31:18 +00:00
var onPublishCmd * exec . Cmd
2020-08-31 13:31:37 +00:00
if c . path . confp . RunOnPublish != "" {
onPublishCmd = exec . Command ( "/bin/sh" , "-c" , c . path . confp . RunOnPublish )
2020-08-04 20:11:16 +00:00
onPublishCmd . Env = append ( os . Environ ( ) ,
2020-08-31 13:31:37 +00:00
"RTSP_SERVER_PATH=" + c . path . name ,
2020-08-04 20:11:16 +00:00
)
2020-07-30 11:31:18 +00:00
onPublishCmd . Stdout = os . Stdout
onPublishCmd . Stderr = os . Stderr
err := onPublishCmd . Start ( )
2020-07-13 09:51:17 +00:00
if err != nil {
c . log ( "ERR: %s" , err )
}
}
2020-07-28 13:10:33 +00:00
if c . streamProtocol == gortsplib . StreamProtocolUdp {
2020-08-31 13:31:37 +00:00
c . runRecordUdp ( )
} else {
c . runRecordTcp ( )
}
2020-07-13 09:12:20 +00:00
2020-08-31 13:31:37 +00:00
if onPublishCmd != nil {
onPublishCmd . Process . Signal ( os . Interrupt )
onPublishCmd . Wait ( )
}
2020-07-13 09:12:20 +00:00
2020-08-31 13:31:37 +00:00
for trackId := range c . streamTracks {
c . rtcpReceivers [ trackId ] . Close ( )
}
return false
}
2020-07-13 09:12:20 +00:00
2020-08-31 13:31:37 +00:00
func ( c * client ) runRecordUdp ( ) {
readDone := make ( chan error )
go func ( ) {
2020-07-13 09:12:20 +00:00
for {
2020-08-31 13:31:37 +00:00
req , err := c . conn . ReadRequest ( )
if err != nil {
readDone <- err
break
}
2020-07-13 09:12:20 +00:00
2020-08-31 13:31:37 +00:00
err = c . handleRequest ( req )
if err != nil {
readDone <- err
break
2020-07-13 09:12:20 +00:00
}
}
2020-08-31 13:31:37 +00:00
} ( )
2020-07-13 09:12:20 +00:00
2020-08-31 13:31:37 +00:00
checkStreamTicker := time . NewTicker ( clientCheckStreamInterval )
defer checkStreamTicker . Stop ( )
2020-07-28 13:10:33 +00:00
2020-08-31 13:31:37 +00:00
receiverReportTicker := time . NewTicker ( clientReceiverReportInterval )
defer receiverReportTicker . Stop ( )
2020-07-28 13:10:33 +00:00
2020-08-31 13:31:37 +00:00
for {
select {
case err := <- readDone :
c . conn . Close ( )
if err != io . EOF && err != errRunTerminate {
c . log ( "ERR: %s" , err )
}
c . p . events <- programEventClientClose { c }
<- c . terminate
return
case <- checkStreamTicker . C :
for trackId := range c . streamTracks {
if time . Since ( c . rtcpReceivers [ trackId ] . LastFrameTime ( ) ) >= c . p . conf . ReadTimeout {
c . log ( "ERR: stream is dead" )
c . conn . Close ( )
<- readDone
c . p . events <- programEventClientClose { c }
<- c . terminate
return
2020-07-13 09:12:20 +00:00
}
2020-08-31 13:31:37 +00:00
}
2020-07-13 09:12:20 +00:00
2020-08-31 13:31:37 +00:00
case <- receiverReportTicker . C :
for trackId := range c . streamTracks {
frame := c . rtcpReceivers [ trackId ] . Report ( )
c . p . serverRtcp . writeChan <- & udpAddrBufPair {
addr : & net . UDPAddr {
IP : c . ip ( ) ,
Zone : c . zone ( ) ,
Port : c . streamTracks [ trackId ] . rtcpPort ,
} ,
buf : frame ,
2020-07-13 09:12:20 +00:00
}
}
2020-08-31 13:31:37 +00:00
case <- c . terminate :
c . conn . Close ( )
<- readDone
return
}
}
}
2020-07-13 09:12:20 +00:00
2020-08-31 13:31:37 +00:00
func ( c * client ) runRecordTcp ( ) {
frame := & gortsplib . InterleavedFrame { }
readBuf := newMultiBuffer ( 3 , clientTcpReadBufferSize )
readDone := make ( chan error )
go func ( ) {
2020-07-13 09:12:20 +00:00
for {
2020-08-31 13:31:37 +00:00
frame . Content = readBuf . next ( )
frame . Content = frame . Content [ : cap ( frame . Content ) ]
recv , err := c . conn . ReadFrameOrRequest ( frame , true )
if err != nil {
readDone <- err
break
}
switch recvt := recv . ( type ) {
case * gortsplib . InterleavedFrame :
if frame . TrackId >= len ( c . streamTracks ) {
readDone <- fmt . Errorf ( "invalid track id '%d'" , frame . TrackId )
break
2020-07-13 09:12:20 +00:00
}
2020-08-31 13:31:37 +00:00
c . rtcpReceivers [ frame . TrackId ] . OnFrame ( frame . StreamType , frame . Content )
c . p . events <- programEventClientFrameTcp {
c . path ,
frame . TrackId ,
frame . StreamType ,
frame . Content ,
}
case * gortsplib . Request :
err := c . handleRequest ( recvt )
if err != nil {
readDone <- err
break
2020-07-13 09:12:20 +00:00
}
}
}
2020-08-31 13:31:37 +00:00
} ( )
2020-07-13 09:12:20 +00:00
2020-08-31 13:31:37 +00:00
receiverReportTicker := time . NewTicker ( clientReceiverReportInterval )
defer receiverReportTicker . Stop ( )
2020-07-13 09:12:20 +00:00
2020-08-31 13:31:37 +00:00
for {
select {
case err := <- readDone :
c . conn . Close ( )
if err != io . EOF && err != errRunTerminate {
c . log ( "ERR: %s" , err )
}
c . p . events <- programEventClientClose { c }
<- c . terminate
return
case <- receiverReportTicker . C :
for trackId := range c . streamTracks {
frame := c . rtcpReceivers [ trackId ] . Report ( )
c . conn . WriteFrame ( & gortsplib . InterleavedFrame {
TrackId : trackId ,
StreamType : gortsplib . StreamTypeRtcp ,
Content : frame ,
} )
}
2020-07-13 09:51:17 +00:00
2020-08-31 13:31:37 +00:00
case <- c . terminate :
c . conn . Close ( )
<- readDone
return
}
2020-07-13 09:51:17 +00:00
}
2019-12-31 12:48:17 +00:00
}