ghostream/stream/srt/srt.go

178 lines
4.7 KiB
Go
Raw Normal View History

2020-09-27 09:14:22 +00:00
package srt
2020-10-02 07:44:31 +00:00
// #include <srt/srt.h>
import "C"
2020-10-02 07:46:15 +00:00
2020-09-27 09:14:22 +00:00
import (
"fmt"
2020-09-27 09:14:22 +00:00
"log"
"net"
"strconv"
2020-10-01 21:31:14 +00:00
"strings"
2020-10-02 07:46:15 +00:00
"github.com/haivision/srtgo"
2020-10-03 15:42:38 +00:00
"gitlab.crans.org/nounous/ghostream/auth"
2020-09-27 09:14:22 +00:00
)
// Options holds web package configuration
type Options struct {
ListenAddress string
MaxClients int
}
2020-10-01 17:05:39 +00:00
// Packet contains the necessary data to broadcast events like stream creating, packet receiving or stream closing.
type Packet struct {
Data []byte
PacketType string
StreamName string
}
// Split host and port from listen address
func splitHostPort(hostport string) (string, uint16) {
host, portS, err := net.SplitHostPort(hostport)
if err != nil {
log.Fatalf("Failed to split host and port from %s", hostport)
}
if host == "" {
host = "0.0.0.0"
}
port64, err := strconv.ParseUint(portS, 10, 16)
if err != nil {
log.Fatalf("Port is not a integer: %s", err)
}
return host, uint16(port64)
2020-09-27 09:14:22 +00:00
}
// Serve SRT server
2020-10-03 15:50:34 +00:00
func Serve(cfg *Options, authBackend auth.Backend, forwardingChannel chan Packet) {
2020-09-29 18:58:55 +00:00
// Start SRT in listen mode
log.Printf("SRT server listening on %s", cfg.ListenAddress)
host, port := splitHostPort(cfg.ListenAddress)
2020-10-03 15:42:38 +00:00
options := map[string]string{
"transtype": "file",
"mode": "listener",
}
2020-10-01 21:31:14 +00:00
sck := srtgo.NewSrtSocket(host, port, options)
2020-09-30 14:53:15 +00:00
if err := sck.Listen(cfg.MaxClients); err != nil {
log.Fatal("Unable to listen to SRT clients:", err)
}
2020-09-27 09:14:22 +00:00
2020-10-02 21:35:01 +00:00
// FIXME: Get the stream type
streamStarted := false
// FIXME Better structure
clientDataChannels := make([]chan Packet, cfg.MaxClients)
listeners := 0
2020-09-27 18:43:00 +00:00
for {
2020-09-29 18:58:55 +00:00
// Wait for new connection
2020-09-27 20:10:47 +00:00
s, err := sck.Accept()
if err != nil {
2020-10-02 21:35:01 +00:00
// log.Println("Error occurred while accepting request:", err)
continue // break // FIXME: should not break here
2020-09-27 18:43:00 +00:00
}
2020-09-27 20:10:47 +00:00
2020-10-02 21:35:01 +00:00
if !streamStarted {
2020-10-03 15:50:34 +00:00
go acceptCallerSocket(s, clientDataChannels, &listeners, authBackend, forwardingChannel)
2020-10-02 21:35:01 +00:00
streamStarted = true
} else {
2020-10-03 14:15:42 +00:00
dataChannel := make(chan Packet, 2048)
2020-10-02 21:35:01 +00:00
clientDataChannels[listeners] = dataChannel
2020-10-03 14:15:42 +00:00
listeners++
2020-10-02 21:35:01 +00:00
go acceptListeningSocket(s, dataChannel)
}
}
sck.Close()
}
2020-10-03 15:50:34 +00:00
func acceptCallerSocket(s *srtgo.SrtSocket, clientDataChannels []chan Packet, listeners *int, authBackend auth.Backend, forwardingChannel chan Packet) {
streamName, err := authenticateSocket(s, authBackend)
if err != nil {
log.Println("Authentication failure:", err)
s.Close()
return
}
log.Printf("Starting stream %s...", streamName)
// Create a new buffer
buff := make([]byte, 2048)
// Setup stream forwarding
forwardingChannel <- Packet{StreamName: streamName, PacketType: "register", Data: nil}
// Read RTP packets forever and send them to the WebRTC Client
for {
n, err := s.Read(buff, 10000)
2020-10-01 21:31:14 +00:00
if err != nil {
log.Println("Error occured while reading SRT socket:", err)
break
2020-10-01 21:31:14 +00:00
}
if n == 0 {
// End of stream
log.Printf("Received no bytes, stopping stream.")
break
2020-10-01 21:31:14 +00:00
}
// log.Printf("Received %d bytes", n)
2020-10-01 21:31:14 +00:00
// 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: streamName, PacketType: "sendData", Data: data}
2020-10-03 14:15:42 +00:00
for i := 0; i < *listeners; i++ {
2020-10-02 21:35:01 +00:00
clientDataChannels[i] <- Packet{StreamName: streamName, 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)})
2020-09-27 09:14:22 +00:00
}
2020-10-01 21:31:14 +00:00
forwardingChannel <- Packet{StreamName: streamName, PacketType: "close", Data: nil}
}
2020-10-02 21:35:01 +00:00
func acceptListeningSocket(s *srtgo.SrtSocket, dataChannel chan Packet) {
streamName, err := s.GetSockOptString(C.SRTO_STREAMID)
if err != nil {
panic(err)
}
log.Printf("New listener for stream %s", streamName)
for {
packet := <-dataChannel
if packet.PacketType == "sendData" {
_, err := s.Write(packet.Data, 10000)
if err != nil {
s.Close()
break
}
}
}
}
2020-10-03 15:50:34 +00:00
func authenticateSocket(s *srtgo.SrtSocket, authBackend auth.Backend) (string, error) {
streamID, err := s.GetSockOptString(C.SRTO_STREAMID)
if err != nil {
return "", fmt.Errorf("error while fetching stream key: %s", err)
}
log.Println(s.GetSockOptString(C.SRTO_PASSPHRASE))
if !strings.Contains(streamID, ":") {
return streamID, fmt.Errorf("warning: stream id must be at the format streamID:password. Input: %s", streamID)
}
splittedStreamID := strings.SplitN(streamID, ":", 2)
streamName, password := splittedStreamID[0], splittedStreamID[1]
2020-10-03 15:50:34 +00:00
if authBackend == nil {
// Bypass authentification if none provided
return streamName, nil
}
loggedIn, err := authBackend.Login(streamName, password)
if !loggedIn {
return streamID, fmt.Errorf("invalid credentials for stream %s", streamName)
}
return streamName, nil
2020-09-27 09:14:22 +00:00
}