2019-12-28 21:07:03 +00:00
package main
import (
"fmt"
2020-06-27 15:23:43 +00:00
"io"
2019-12-28 21:07:03 +00:00
"log"
2020-06-15 20:41:14 +00:00
"net"
2020-06-27 13:42:54 +00:00
"net/http"
_ "net/http/pprof"
2019-12-28 21:07:03 +00:00
"os"
2020-07-12 10:34:35 +00:00
"github.com/aler9/gortsplib"
2020-07-19 09:51:28 +00:00
"github.com/aler9/sdp/v3"
2019-12-28 21:07:03 +00:00
"gopkg.in/alecthomas/kingpin.v2"
)
2020-06-26 09:00:10 +00:00
var Version = "v0.0.0"
2019-12-28 21:07:03 +00:00
2020-06-27 11:38:35 +00:00
type programEvent interface {
isProgramEvent ( )
}
type programEventClientNew struct {
nconn net . Conn
}
func ( programEventClientNew ) isProgramEvent ( ) { }
type programEventClientClose struct {
done chan struct { }
client * serverClient
}
func ( programEventClientClose ) isProgramEvent ( ) { }
2020-06-30 13:12:39 +00:00
type programEventClientDescribe struct {
2020-06-27 11:38:35 +00:00
path string
res chan [ ] byte
}
2020-06-30 13:12:39 +00:00
func ( programEventClientDescribe ) isProgramEvent ( ) { }
2020-06-27 11:38:35 +00:00
type programEventClientAnnounce struct {
2020-06-30 13:12:39 +00:00
res chan error
client * serverClient
path string
2020-06-27 11:38:35 +00:00
}
func ( programEventClientAnnounce ) isProgramEvent ( ) { }
type programEventClientSetupPlay struct {
res chan error
client * serverClient
path string
2020-07-19 15:54:31 +00:00
protocol gortsplib . StreamProtocol
2020-06-27 11:38:35 +00:00
rtpPort int
rtcpPort int
}
func ( programEventClientSetupPlay ) isProgramEvent ( ) { }
type programEventClientSetupRecord struct {
res chan error
client * serverClient
2020-07-19 15:54:31 +00:00
protocol gortsplib . StreamProtocol
2020-06-27 11:38:35 +00:00
rtpPort int
rtcpPort int
}
func ( programEventClientSetupRecord ) isProgramEvent ( ) { }
type programEventClientPlay1 struct {
res chan error
client * serverClient
}
func ( programEventClientPlay1 ) isProgramEvent ( ) { }
type programEventClientPlay2 struct {
2020-07-13 09:12:20 +00:00
done chan struct { }
2020-06-27 11:38:35 +00:00
client * serverClient
}
func ( programEventClientPlay2 ) isProgramEvent ( ) { }
2020-07-13 09:12:20 +00:00
type programEventClientPlayStop struct {
done chan struct { }
client * serverClient
}
func ( programEventClientPlayStop ) isProgramEvent ( ) { }
2020-06-27 11:38:35 +00:00
type programEventClientRecord struct {
2020-07-13 09:12:20 +00:00
done chan struct { }
2020-06-27 11:38:35 +00:00
client * serverClient
}
func ( programEventClientRecord ) isProgramEvent ( ) { }
2020-07-13 09:12:20 +00:00
type programEventClientRecordStop struct {
done chan struct { }
client * serverClient
}
func ( programEventClientRecordStop ) isProgramEvent ( ) { }
2020-06-30 13:12:39 +00:00
type programEventClientFrameUdp struct {
2020-07-12 20:53:22 +00:00
addr * net . UDPAddr
streamType gortsplib . StreamType
buf [ ] byte
2020-06-27 11:38:35 +00:00
}
2020-06-30 13:12:39 +00:00
func ( programEventClientFrameUdp ) isProgramEvent ( ) { }
2020-06-27 11:38:35 +00:00
2020-06-30 13:12:39 +00:00
type programEventClientFrameTcp struct {
2020-07-12 20:53:22 +00:00
path string
trackId int
streamType gortsplib . StreamType
buf [ ] byte
2020-06-27 11:38:35 +00:00
}
2020-06-30 13:12:39 +00:00
func ( programEventClientFrameTcp ) isProgramEvent ( ) { }
type programEventStreamerReady struct {
2020-07-14 07:19:07 +00:00
source * source
2020-06-30 13:12:39 +00:00
}
func ( programEventStreamerReady ) isProgramEvent ( ) { }
type programEventStreamerNotReady struct {
2020-07-14 07:19:07 +00:00
source * source
2020-06-30 13:12:39 +00:00
}
func ( programEventStreamerNotReady ) isProgramEvent ( ) { }
type programEventStreamerFrame struct {
2020-07-14 07:19:07 +00:00
source * source
2020-07-12 20:53:22 +00:00
trackId int
streamType gortsplib . StreamType
buf [ ] byte
2020-06-30 13:12:39 +00:00
}
func ( programEventStreamerFrame ) isProgramEvent ( ) { }
2020-06-27 11:38:35 +00:00
type programEventTerminate struct { }
func ( programEventTerminate ) isProgramEvent ( ) { }
2020-07-14 07:19:07 +00:00
// a publisher can be either a serverClient or a source
2020-06-30 13:12:39 +00:00
type publisher interface {
publisherIsReady ( ) bool
publisherSdpText ( ) [ ] byte
2020-07-12 15:24:12 +00:00
publisherSdpParsed ( ) * sdp . SessionDescription
2020-06-30 13:12:39 +00:00
}
2020-05-10 13:33:42 +00:00
type program struct {
2020-06-27 15:23:43 +00:00
conf * conf
2020-07-12 10:34:35 +00:00
rtspl * serverTcpListener
rtpl * serverUdpListener
rtcpl * serverUdpListener
2020-06-27 12:18:16 +00:00
clients map [ * serverClient ] struct { }
2020-07-14 07:19:07 +00:00
sources [ ] * source
2020-06-30 13:12:39 +00:00
publishers map [ string ] publisher
2020-06-27 12:18:16 +00:00
publisherCount int
2020-07-20 09:42:56 +00:00
readerCount int
2020-06-27 11:38:35 +00:00
events chan programEvent
done chan struct { }
2020-05-10 13:33:42 +00:00
}
2020-02-16 15:05:08 +00:00
2020-07-20 09:42:56 +00:00
func newProgram ( args [ ] string , stdin io . Reader ) ( * program , error ) {
2020-06-27 19:22:50 +00:00
k := kingpin . New ( "rtsp-simple-server" ,
"rtsp-simple-server " + Version + "\n\nRTSP server." )
2020-05-11 07:31:56 +00:00
2020-06-27 19:22:50 +00:00
argVersion := k . Flag ( "version" , "print version" ) . Bool ( )
2020-07-13 15:27:04 +00:00
argConfPath := k . Arg ( "confpath" , "path to a config file. The default is rtsp-simple-server.yml. Use 'stdin' to read config from stdin" ) . Default ( "rtsp-simple-server.yml" ) . String ( )
2020-05-11 07:31:56 +00:00
2020-07-20 09:42:56 +00:00
kingpin . MustParse ( k . Parse ( args ) )
2020-05-11 07:31:56 +00:00
2020-06-27 15:23:43 +00:00
if * argVersion == true {
2020-05-11 07:39:49 +00:00
fmt . Println ( Version )
2020-05-10 19:32:40 +00:00
os . Exit ( 0 )
2020-01-26 11:58:56 +00:00
}
2020-06-27 15:23:43 +00:00
conf , err := loadConf ( * argConfPath , stdin )
if err != nil {
return nil , err
}
2020-06-30 13:12:39 +00:00
p := & program {
conf : conf ,
clients : make ( map [ * serverClient ] struct { } ) ,
publishers : make ( map [ string ] publisher ) ,
events : make ( chan programEvent ) ,
done : make ( chan struct { } ) ,
}
for path , pconf := range conf . Paths {
if pconf . Source != "record" {
2020-07-20 09:42:56 +00:00
s := newSource ( p , path , pconf . sourceUrl , pconf . sourceProtocolParsed )
2020-07-14 07:19:07 +00:00
p . sources = append ( p . sources , s )
2020-06-30 13:12:39 +00:00
p . publishers [ path ] = s
}
2019-12-28 21:07:03 +00:00
}
2020-06-27 12:18:16 +00:00
p . log ( "rtsp-simple-server %s" , Version )
2020-06-27 15:23:43 +00:00
if conf . Pprof {
2020-06-27 13:42:54 +00:00
go func ( mux * http . ServeMux ) {
server := & http . Server {
Addr : ":9999" ,
Handler : mux ,
}
p . log ( "pprof is available on :9999" )
panic ( server . ListenAndServe ( ) )
} ( http . DefaultServeMux )
http . DefaultServeMux = http . NewServeMux ( )
}
2020-07-12 20:53:22 +00:00
p . rtpl , err = newServerUdpListener ( p , conf . RtpPort , gortsplib . StreamTypeRtp )
2019-12-28 21:07:03 +00:00
if err != nil {
return nil , err
}
2020-07-12 20:53:22 +00:00
p . rtcpl , err = newServerUdpListener ( p , conf . RtcpPort , gortsplib . StreamTypeRtcp )
2019-12-28 21:07:03 +00:00
if err != nil {
return nil , err
}
2020-07-12 10:34:35 +00:00
p . rtspl , err = newServerTcpListener ( p )
2019-12-28 21:07:03 +00:00
if err != nil {
return nil , err
}
2020-07-12 10:34:35 +00:00
go p . rtpl . run ( )
go p . rtcpl . run ( )
go p . rtspl . run ( )
2020-07-14 07:19:07 +00:00
for _ , s := range p . sources {
2020-06-30 13:12:39 +00:00
go s . run ( )
}
2020-06-27 11:38:35 +00:00
go p . run ( )
2019-12-28 21:07:03 +00:00
2020-05-10 13:33:42 +00:00
return p , nil
2019-12-28 21:07:03 +00:00
}
2020-06-27 12:18:16 +00:00
func ( p * program ) log ( format string , args ... interface { } ) {
log . Printf ( "[%d/%d/%d] " + format , append ( [ ] interface { } { len ( p . clients ) ,
2020-07-20 09:42:56 +00:00
p . publisherCount , p . readerCount } , args ... ) ... )
2020-06-27 12:18:16 +00:00
}
2020-06-27 11:38:35 +00:00
func ( p * program ) run ( ) {
outer :
for rawEvt := range p . events {
switch evt := rawEvt . ( type ) {
case programEventClientNew :
c := newServerClient ( p , evt . nconn )
p . clients [ c ] = struct { } { }
2020-06-27 12:18:16 +00:00
c . log ( "connected" )
2020-06-27 11:38:35 +00:00
case programEventClientClose :
// already deleted
if _ , ok := p . clients [ evt . client ] ; ! ok {
close ( evt . done )
continue
}
delete ( p . clients , evt . client )
if evt . client . path != "" {
if pub , ok := p . publishers [ evt . client . path ] ; ok && pub == evt . client {
delete ( p . publishers , evt . client . path )
}
}
2020-06-27 12:18:16 +00:00
evt . client . log ( "disconnected" )
2020-06-27 11:38:35 +00:00
close ( evt . done )
2020-06-30 13:12:39 +00:00
case programEventClientDescribe :
2020-06-27 11:38:35 +00:00
pub , ok := p . publishers [ evt . path ]
2020-06-30 13:12:39 +00:00
if ! ok || ! pub . publisherIsReady ( ) {
2020-06-27 11:38:35 +00:00
evt . res <- nil
continue
}
2020-06-30 13:12:39 +00:00
evt . res <- pub . publisherSdpText ( )
2020-06-27 11:38:35 +00:00
case programEventClientAnnounce :
_ , ok := p . publishers [ evt . path ]
if ok {
2020-06-30 13:12:39 +00:00
evt . res <- fmt . Errorf ( "someone is already publishing on path '%s'" , evt . path )
2020-06-27 11:38:35 +00:00
continue
}
evt . client . path = evt . path
2020-07-13 07:45:45 +00:00
evt . client . state = clientStateAnnounce
2020-06-27 11:38:35 +00:00
p . publishers [ evt . path ] = evt . client
evt . res <- nil
case programEventClientSetupPlay :
pub , ok := p . publishers [ evt . path ]
2020-06-30 13:12:39 +00:00
if ! ok || ! pub . publisherIsReady ( ) {
2020-06-27 11:38:35 +00:00
evt . res <- fmt . Errorf ( "no one is streaming on path '%s'" , evt . path )
continue
}
2020-06-30 13:12:39 +00:00
sdpParsed := pub . publisherSdpParsed ( )
2020-07-12 15:24:12 +00:00
if len ( evt . client . streamTracks ) >= len ( sdpParsed . MediaDescriptions ) {
2020-06-27 11:38:35 +00:00
evt . res <- fmt . Errorf ( "all the tracks have already been setup" )
continue
}
evt . client . path = evt . path
evt . client . streamProtocol = evt . protocol
2020-07-19 15:54:31 +00:00
evt . client . streamTracks = append ( evt . client . streamTracks , & serverClientTrack {
2020-06-27 11:38:35 +00:00
rtpPort : evt . rtpPort ,
rtcpPort : evt . rtcpPort ,
} )
2020-07-13 07:45:45 +00:00
evt . client . state = clientStatePrePlay
2020-06-27 11:38:35 +00:00
evt . res <- nil
case programEventClientSetupRecord :
evt . client . streamProtocol = evt . protocol
2020-07-19 15:54:31 +00:00
evt . client . streamTracks = append ( evt . client . streamTracks , & serverClientTrack {
2020-06-27 11:38:35 +00:00
rtpPort : evt . rtpPort ,
rtcpPort : evt . rtcpPort ,
} )
2020-07-13 07:45:45 +00:00
evt . client . state = clientStatePreRecord
2020-06-27 11:38:35 +00:00
evt . res <- nil
case programEventClientPlay1 :
pub , ok := p . publishers [ evt . client . path ]
2020-06-30 13:12:39 +00:00
if ! ok || ! pub . publisherIsReady ( ) {
2020-06-27 11:38:35 +00:00
evt . res <- fmt . Errorf ( "no one is streaming on path '%s'" , evt . client . path )
continue
}
2020-06-30 13:12:39 +00:00
sdpParsed := pub . publisherSdpParsed ( )
2020-07-12 15:24:12 +00:00
if len ( evt . client . streamTracks ) != len ( sdpParsed . MediaDescriptions ) {
2020-06-27 11:38:35 +00:00
evt . res <- fmt . Errorf ( "not all tracks have been setup" )
continue
}
evt . res <- nil
case programEventClientPlay2 :
2020-07-20 09:42:56 +00:00
p . readerCount += 1
2020-07-13 07:45:45 +00:00
evt . client . state = clientStatePlay
2020-07-13 09:12:20 +00:00
close ( evt . done )
case programEventClientPlayStop :
2020-07-20 09:42:56 +00:00
p . readerCount -= 1
2020-07-13 09:12:20 +00:00
evt . client . state = clientStatePrePlay
close ( evt . done )
2020-06-27 11:38:35 +00:00
case programEventClientRecord :
2020-06-27 12:18:16 +00:00
p . publisherCount += 1
2020-07-13 07:45:45 +00:00
evt . client . state = clientStateRecord
2020-07-13 09:12:20 +00:00
close ( evt . done )
case programEventClientRecordStop :
p . publisherCount -= 1
evt . client . state = clientStatePreRecord
// close all other clients that share the same path
for oc := range p . clients {
if oc != evt . client && oc . path == evt . client . path {
go oc . close ( )
}
}
close ( evt . done )
2020-06-27 11:38:35 +00:00
2020-06-30 13:12:39 +00:00
case programEventClientFrameUdp :
2020-07-12 20:53:22 +00:00
client , trackId := p . findPublisher ( evt . addr , evt . streamType )
2020-07-11 14:40:19 +00:00
if client == nil {
2020-06-27 11:38:35 +00:00
continue
}
2020-07-19 15:54:31 +00:00
client . rtcpReceivers [ trackId ] . OnFrame ( evt . streamType , evt . buf )
2020-07-12 20:53:22 +00:00
p . forwardFrame ( client . path , trackId , evt . streamType , evt . buf )
2020-06-27 11:38:35 +00:00
2020-06-30 13:12:39 +00:00
case programEventClientFrameTcp :
2020-07-12 20:53:22 +00:00
p . forwardFrame ( evt . path , evt . trackId , evt . streamType , evt . buf )
2020-06-27 11:38:35 +00:00
2020-06-30 13:12:39 +00:00
case programEventStreamerReady :
2020-07-14 07:19:07 +00:00
evt . source . ready = true
2020-06-30 13:12:39 +00:00
p . publisherCount += 1
2020-07-14 07:19:07 +00:00
evt . source . log ( "ready" )
2020-06-30 13:12:39 +00:00
case programEventStreamerNotReady :
2020-07-14 07:19:07 +00:00
evt . source . ready = false
2020-06-30 13:12:39 +00:00
p . publisherCount -= 1
2020-07-14 07:19:07 +00:00
evt . source . log ( "not ready" )
2020-06-30 13:12:39 +00:00
// close all clients that share the same path
for oc := range p . clients {
2020-07-14 07:19:07 +00:00
if oc . path == evt . source . path {
2020-06-30 13:12:39 +00:00
go oc . close ( )
}
}
case programEventStreamerFrame :
2020-07-14 07:19:07 +00:00
p . forwardFrame ( evt . source . path , evt . trackId , evt . streamType , evt . buf )
2020-06-30 13:12:39 +00:00
2020-06-27 11:38:35 +00:00
case programEventTerminate :
break outer
}
}
go func ( ) {
for rawEvt := range p . events {
switch evt := rawEvt . ( type ) {
case programEventClientClose :
close ( evt . done )
2020-06-30 13:12:39 +00:00
case programEventClientDescribe :
2020-06-27 11:38:35 +00:00
evt . res <- nil
case programEventClientAnnounce :
evt . res <- fmt . Errorf ( "terminated" )
case programEventClientSetupPlay :
evt . res <- fmt . Errorf ( "terminated" )
case programEventClientSetupRecord :
evt . res <- fmt . Errorf ( "terminated" )
case programEventClientPlay1 :
evt . res <- fmt . Errorf ( "terminated" )
case programEventClientPlay2 :
2020-07-13 09:12:20 +00:00
close ( evt . done )
case programEventClientPlayStop :
close ( evt . done )
2020-06-27 11:38:35 +00:00
case programEventClientRecord :
2020-07-13 09:12:20 +00:00
close ( evt . done )
case programEventClientRecordStop :
close ( evt . done )
2020-06-27 11:38:35 +00:00
}
}
} ( )
2020-07-14 07:19:07 +00:00
for _ , s := range p . sources {
2020-06-30 13:12:39 +00:00
s . close ( )
}
2020-07-12 10:34:35 +00:00
p . rtspl . close ( )
p . rtcpl . close ( )
p . rtpl . close ( )
2020-06-27 11:38:35 +00:00
for c := range p . clients {
c . close ( )
}
close ( p . events )
close ( p . done )
}
func ( p * program ) close ( ) {
p . events <- programEventTerminate { }
<- p . done
}
2020-07-12 20:53:22 +00:00
func ( p * program ) findPublisher ( addr * net . UDPAddr , streamType gortsplib . StreamType ) ( * serverClient , int ) {
2020-07-11 14:40:19 +00:00
for _ , pub := range p . publishers {
cl , ok := pub . ( * serverClient )
if ! ok {
continue
}
2020-07-19 15:54:31 +00:00
if cl . streamProtocol != gortsplib . StreamProtocolUdp ||
2020-07-13 07:45:45 +00:00
cl . state != clientStateRecord ||
2020-07-11 14:40:19 +00:00
! cl . ip ( ) . Equal ( addr . IP ) {
continue
}
for i , t := range cl . streamTracks {
2020-07-12 20:53:22 +00:00
if streamType == gortsplib . StreamTypeRtp {
2020-07-11 14:40:19 +00:00
if t . rtpPort == addr . Port {
return cl , i
}
} else {
if t . rtcpPort == addr . Port {
return cl , i
}
}
}
}
return nil , - 1
}
2020-07-12 20:53:22 +00:00
func ( p * program ) forwardFrame ( path string , trackId int , streamType gortsplib . StreamType , frame [ ] byte ) {
2020-07-12 10:34:35 +00:00
for client := range p . clients {
2020-07-13 07:45:45 +00:00
if client . path == path && client . state == clientStatePlay {
2020-07-19 15:54:31 +00:00
if client . streamProtocol == gortsplib . StreamProtocolUdp {
2020-07-12 20:53:22 +00:00
if streamType == gortsplib . StreamTypeRtp {
2020-07-12 10:34:35 +00:00
p . rtpl . write ( & udpAddrBufPair {
2020-07-11 14:40:19 +00:00
addr : & net . UDPAddr {
2020-07-12 10:34:35 +00:00
IP : client . ip ( ) ,
Zone : client . zone ( ) ,
Port : client . streamTracks [ trackId ] . rtpPort ,
2020-07-11 14:40:19 +00:00
} ,
buf : frame ,
} )
2020-06-27 11:38:35 +00:00
} else {
2020-07-12 10:34:35 +00:00
p . rtcpl . write ( & udpAddrBufPair {
2020-07-11 14:40:19 +00:00
addr : & net . UDPAddr {
2020-07-12 10:34:35 +00:00
IP : client . ip ( ) ,
Zone : client . zone ( ) ,
Port : client . streamTracks [ trackId ] . rtcpPort ,
2020-07-11 14:40:19 +00:00
} ,
buf : frame ,
} )
2020-06-27 11:38:35 +00:00
}
} else {
2020-07-12 10:34:35 +00:00
buf := client . writeBuf . swap ( )
buf = buf [ : len ( frame ) ]
copy ( buf , frame )
client . events <- serverClientEventFrameTcp {
frame : & gortsplib . InterleavedFrame {
2020-07-13 07:45:45 +00:00
TrackId : trackId ,
StreamType : streamType ,
Content : buf ,
2020-07-12 10:34:35 +00:00
} ,
}
2020-06-27 11:38:35 +00:00
}
}
}
2020-05-10 14:33:20 +00:00
}
2019-12-28 21:07:03 +00:00
func main ( ) {
2020-06-27 15:23:43 +00:00
_ , err := newProgram ( os . Args [ 1 : ] , os . Stdin )
2019-12-28 21:07:03 +00:00
if err != nil {
2020-07-28 13:10:33 +00:00
log . Fatal ( "ERR:" , err )
2019-12-28 21:07:03 +00:00
}
2020-06-24 20:42:39 +00:00
select { }
2019-12-28 21:07:03 +00:00
}