ynerant
/
ghostream
mirror of https://gitlab.crans.org/nounous/ghostream.git
1
0
Fork 0
Code Issues Releases Wiki Activity

Handle client connection closure

This commit is contained in:
Yohann D'ANELLO 2020-10-04 13:08:46 +02:00
parent 2a24598aea
commit 610ddfab28
1 changed files with 16 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
stream/srt/srt.go
View File

@ -52,9 +52,7 @@ func Serve(cfg *Options, authBackend auth.Backend, forwardingChannel chan Packet
log.Fatal("Unable to listen for SRT clients:", err) log.Fatal("Unable to listen for SRT clients:", err)
} }
// FIXME Better structure clientDataChannels := make([]chan Packet, 0, cfg.MaxClients)
clientDataChannels := make([]chan Packet, cfg.MaxClients)
listeners := 0
for { for {
// Wait for new connection // Wait for new connection
@ -84,21 +82,20 @@ func Serve(cfg *Options, authBackend auth.Backend, forwardingChannel chan Packet
} }
} }
go handleStreamer(s, name, clientDataChannels, &listeners, forwardingChannel) go handleStreamer(s, name, &clientDataChannels, forwardingChannel)
} else { } else {
// password was not provided so it is a viewer // password was not provided so it is a viewer
name := split[0] name := split[0]
dataChannel := make(chan Packet, 2048) dataChannel := make(chan Packet, 4096)
clientDataChannels[listeners] = dataChannel clientDataChannels = append(clientDataChannels, dataChannel)
listeners++
go handleViewer(s, name, dataChannel) go handleViewer(s, name, dataChannel, &clientDataChannels)
} }
} }
} }
func handleStreamer(s *srtgo.SrtSocket, name string, clientDataChannels []chan Packet, listeners *int, forwardingChannel chan Packet) { func handleStreamer(s *srtgo.SrtSocket, name string, clientDataChannels *[]chan Packet, forwardingChannel chan Packet) {
log.Printf("New SRT streamer for stream %s", name) log.Printf("New SRT streamer for stream %s", name)
// Create a new buffer // Create a new buffer
@ -127,8 +124,8 @@ func handleStreamer(s *srtgo.SrtSocket, name string, clientDataChannels []chan P
data := make([]byte, n) data := make([]byte, n)
copy(data, buff[:n]) copy(data, buff[:n])
forwardingChannel <- Packet{StreamName: name, PacketType: "sendData", Data: data} forwardingChannel <- Packet{StreamName: name, PacketType: "sendData", Data: data}
for i := 0; i < *listeners; i++ { for _, dataChannel := range *clientDataChannels {
clientDataChannels[i] <- Packet{StreamName: name, PacketType: "sendData", Data: data} dataChannel <- Packet{StreamName: name, PacketType: "sendData", Data: data}
} }
// TODO: Send to WebRTC // TODO: Send to WebRTC
@ -139,7 +136,9 @@ func handleStreamer(s *srtgo.SrtSocket, name string, clientDataChannels []chan P
forwardingChannel <- Packet{StreamName: name, PacketType: "close", Data: nil} forwardingChannel <- Packet{StreamName: name, PacketType: "close", Data: nil}
} }
func handleViewer(s *srtgo.SrtSocket, name string, dataChannel chan Packet) { func handleViewer(s *srtgo.SrtSocket, name string, dataChannel chan Packet, dataChannels *[]chan Packet) {
// FIXME Should not pass all dataChannels to one viewer
log.Printf("New SRT viewer for stream %s", name) log.Printf("New SRT viewer for stream %s", name)
// Receive packets from channel and send them // Receive packets from channel and send them
@ -149,6 +148,11 @@ func handleViewer(s *srtgo.SrtSocket, name string, dataChannel chan Packet) {
_, err := s.Write(packet.Data, 10000) _, err := s.Write(packet.Data, 10000)
if err != nil { if err != nil {
s.Close() s.Close()
for i, channel := range *dataChannels {
if channel == dataChannel {
*dataChannels = append((*dataChannels)[:i], (*dataChannels)[i+1:]...)
}
}
return return
} }
} }