ynerant/ghostream
1
0
Fork 0
mirror of https://gitlab.crans.org/nounous/ghostream.git synced 2024-12-23 00:22:19 +00:00
Code Issues Releases Wiki Activity

Move srt handlers in seperate file

Browse Source
This commit is contained in:
Alexandre Iooss 2020-10-04 14:24:00 +02:00
parent 610ddfab28
commit dd6318256f
No known key found for this signature in database
GPG Key ID: 6C79278F3FCDCC02
3 changed files with 68 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
main.go
View File

@ -103,7 +103,7 @@ func main() {
remoteSdpChan := make(chan webrtc.SessionDescription) remoteSdpChan := make(chan webrtc.SessionDescription)
localSdpChan := make(chan webrtc.SessionDescription) localSdpChan := make(chan webrtc.SessionDescription)
// SRT channel, to propagate forwarding // SRT channel for forwarding
forwardingChannel := make(chan srt.Packet, 65536) forwardingChannel := make(chan srt.Packet, 65536)
// Start stream, web and monitoring server, and stream forwarding // Start stream, web and monitoring server, and stream forwarding

67
stream/srt/handler.go Normal file
View File

@ -0,0 +1,67 @@
package srt
import (
"log"
"github.com/haivision/srtgo"
)
func handleStreamer(s *srtgo.SrtSocket, name string, clientDataChannels *[]chan Packet, forwardingChannel chan Packet) {
log.Printf("New SRT streamer for stream %s", name)
// Create a new buffer
buff := make([]byte, 2048)
// Setup stream forwarding
forwardingChannel <- Packet{StreamName: name, PacketType: "register", Data: nil}
// Read RTP packets forever and send them to the WebRTC Client
for {
// UDP packet cannot be larger than MTU (1500)
n, err := s.Read(buff, 1500)
if err != nil {
log.Println("Error occured while reading SRT socket:", err)
break
}
if n == 0 {
// End of stream
log.Printf("Received no bytes, stopping stream.")
break
}
// Send raw packet to other streams
// Copy data in another buffer to ensure that the data would not be overwritten
data := make([]byte, n)
copy(data, buff[:n])
forwardingChannel <- Packet{StreamName: name, PacketType: "sendData", Data: data}
for _, dataChannel := range *clientDataChannels {
dataChannel <- Packet{StreamName: name, PacketType: "sendData", Data: data}
}
}
forwardingChannel <- Packet{StreamName: name, PacketType: "close", Data: nil}
}
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)
// Receive packets from channel and send them
for {
packet := <-dataChannel
if packet.PacketType == "sendData" {
_, err := s.Write(packet.Data, 10000)
if err != nil {
s.Close()
for i, channel := range *dataChannels {
if channel == dataChannel {
*dataChannels = append((*dataChannels)[:i], (*dataChannels)[i+1:]...)
}
}
return
}
}
}
}

64
stream/srt/srt.go
View File

@ -94,67 +94,3 @@ func Serve(cfg *Options, authBackend auth.Backend, 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)
// Create a new buffer
buff := make([]byte, 2048)
// Setup stream forwarding
forwardingChannel <- Packet{StreamName: name, PacketType: "register", Data: nil}
// Read RTP packets forever and send them to the WebRTC Client
for {
n, err := s.Read(buff, 10000)
if err != nil {
log.Println("Error occured while reading SRT socket:", err)
break
}
if n == 0 {
// End of stream
log.Printf("Received no bytes, stopping stream.")
break
}
// log.Printf("Received %d bytes", n)
// Send raw packet to other streams
// Copy data in another buffer to ensure that the data would not be overwritten
data := make([]byte, n)
copy(data, buff[:n])
forwardingChannel <- Packet{StreamName: name, PacketType: "sendData", Data: data}
for _, dataChannel := range *clientDataChannels {
dataChannel <- Packet{StreamName: name, PacketType: "sendData", Data: data}
}
// TODO: Send to WebRTC
// See https://github.com/ebml-go/webm/blob/master/reader.go
//err := videoTrack.WriteSample(media.Sample{Data: data, Samples: uint32(sampleCount)})
}
forwardingChannel <- Packet{StreamName: name, PacketType: "close", Data: nil}
}
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)
// Receive packets from channel and send them
for {
packet := <-dataChannel
if packet.PacketType == "sendData" {
_, err := s.Write(packet.Data, 10000)
if err != nil {
s.Close()
for i, channel := range *dataChannels {
if channel == dataChannel {
*dataChannels = append((*dataChannels)[:i], (*dataChannels)[i+1:]...)
}
}
return
}
}
}
}