ComunicRTCProxy/relay.go

254 lines
6.5 KiB
Go
Raw Normal View History

2020-04-11 08:30:28 +00:00
/// RTC Relay
///
/// @author Pierre Hubert
package main
import (
"encoding/json"
2020-04-11 10:14:55 +00:00
"fmt"
"io"
2020-04-11 08:30:28 +00:00
"log"
2020-04-11 09:26:51 +00:00
"time"
2020-04-11 08:30:28 +00:00
2020-04-11 10:14:55 +00:00
"github.com/pion/rtcp"
2020-04-11 08:30:28 +00:00
"github.com/pion/webrtc/v2"
)
const (
// SDP This is a SDP signal
SDP = iota
// CANDIDATE This is a candidate
CANDIDATE = iota
)
type receivedSignal struct {
2020-04-11 09:26:51 +00:00
peerID string
callHash string
2020-04-11 08:30:28 +00:00
sigType uint
offer webrtc.SessionDescription
candidate webrtc.ICECandidateInit
}
/// We keep for each connection its channel
2020-04-11 09:26:51 +00:00
var closeChan = make(chan string)
2020-04-11 08:30:28 +00:00
var connections = make(map[string]chan receivedSignal)
/// Process incoming messages
func onSignal(callHash, peerID string, data map[string]interface{}) {
2020-04-11 09:26:51 +00:00
// Close all the channels that requested so
processCloseRequests()
2020-04-11 08:30:28 +00:00
// Decode received signal
2020-04-11 09:26:51 +00:00
newSignal := receivedSignal{
peerID: peerID,
callHash: callHash,
}
2020-04-11 08:30:28 +00:00
if data["type"] == "SDP" {
newSignal.sigType = SDP
newSignal.offer.Type = webrtc.SDPTypeOffer
newSignal.offer.SDP = data["data"].(map[string]interface{})["sdp"].(string)
} else if data["type"] == "CANDIDATE" {
newSignal.sigType = CANDIDATE
// I have to re-encode data to initialize ICECandidate
var enc []byte
enc, err := json.Marshal(data["data"])
if err != nil {
log.Printf("Could not re-encode candidate ! %s", err)
return
}
err = json.Unmarshal(enc, &newSignal.candidate)
if err != nil {
log.Printf("Discarding invalid candidate: %s", err)
return
}
} else {
log.Fatalf("Invalid signal type: %s !", data["type"])
}
2020-04-11 09:26:51 +00:00
// Check if we are attempting to connect as viewer to a non existing channel
if _, ok := connections[callHash]; !ok {
if peerID != "0" || newSignal.sigType != SDP {
println("Attempting to connect as viewer | send candidate to a non-ready broadcast!")
return
}
}
// Handle new offers
if newSignal.sigType == SDP && peerID == "0" {
// Check if we are overwriting another connection
if _, ok := connections[callHash]; ok {
closeConnection(callHash)
}
connections[callHash] = make(chan receivedSignal, 10)
go newCall(newSignal, connections[callHash])
} else {
// Forward the message to the channel
connections[callHash] <- newSignal
}
}
/// Close a connection
func closeConnection(callHash string) {
log.Printf("Closing call %s", callHash)
if val, ok := connections[callHash]; ok {
close(val)
delete(connections, callHash)
}
}
// Ask for a channel to be closed
func askForClose(callHash string) {
closeChan <- callHash
}
// Process channel close requests (in thread safe way)
func processCloseRequests() {
for {
select {
case id := <-closeChan:
closeConnection(id)
case <-time.After(time.Millisecond * 10):
return
}
}
}
/// Start new call
func newCall(mainOffer receivedSignal, ch chan receivedSignal) {
2020-04-11 10:14:55 +00:00
log.Printf("Starting new call: %s", mainOffer.callHash)
2020-04-11 09:26:51 +00:00
// Since we are answering use PayloadTypes declared by offerer
mediaEngine := webrtc.MediaEngine{}
err := mediaEngine.PopulateFromSDP(mainOffer.offer)
if err != nil {
log.Println("Error: invalid data in offer!", err)
askForClose(mainOffer.callHash)
return
}
2020-04-11 10:14:55 +00:00
// Create the API object with the MediaEngine
api := webrtc.NewAPI(webrtc.WithMediaEngine(mediaEngine))
// Setup config
peerConnectionConfig := webrtc.Configuration{
ICEServers: callConf.iceServers,
}
// Create a new RTCPeerConnection
peerConnection, err := api.NewPeerConnection(peerConnectionConfig)
if err != nil {
log.Println("Error: could not create peer connection!", err)
askForClose(mainOffer.callHash)
return
}
// Allow us to receive 1 audio & 1 video track
if _, err = peerConnection.AddTransceiverFromKind(webrtc.RTPCodecTypeVideo); err != nil {
log.Println("Error: could not prepare to receive video track!", err)
askForClose(mainOffer.callHash)
return
}
if _, err = peerConnection.AddTransceiverFromKind(webrtc.RTPCodecTypeAudio); err != nil {
log.Println("Error: could not prepare to receive audio track!", err)
askForClose(mainOffer.callHash)
return
}
localTrackChan := make(chan *webrtc.Track)
// Set a handler for when a new remote track starts, this just distributes all our packets
// to connected peers
peerConnection.OnTrack(func(remoteTrack *webrtc.Track, receiver *webrtc.RTPReceiver) {
// Send a PLI on an interval so that the publisher is pushing a keyframe every rtcpPLIInterval
// This can be less wasteful by processing incoming RTCP events, then we would emit a NACK/PLI when a viewer requests it
go func() {
rtcpPLIInterval := time.Second * 3
ticker := time.NewTicker(rtcpPLIInterval)
for range ticker.C {
if rtcpSendErr := peerConnection.WriteRTCP([]rtcp.Packet{&rtcp.PictureLossIndication{MediaSSRC: remoteTrack.SSRC()}}); rtcpSendErr != nil {
fmt.Println(rtcpSendErr)
}
}
}()
// Create a local track, all our SFU clients will be fed via this track
localTrack, newTrackErr := peerConnection.NewTrack(remoteTrack.PayloadType(), remoteTrack.SSRC(), remoteTrack.ID(), remoteTrack.Label())
if newTrackErr != nil {
log.Println("New track error!", err)
askForClose(mainOffer.callHash)
return
}
// Send the track to the main goroutine (in order to respond correctly to SDP requests & responses)
localTrackChan <- localTrack
rtpBuf := make([]byte, 1400)
for {
i, readErr := remoteTrack.Read(rtpBuf)
if readErr != nil {
// Could not read from remote track
log.Println("Read error!", err)
askForClose(mainOffer.callHash)
break
}
// ErrClosedPipe means we don't have any subscribers, this is ok if no peers have connected yet
if _, err = localTrack.Write(rtpBuf[:i]); err != nil && err != io.ErrClosedPipe {
log.Println("Write error!", err)
askForClose(mainOffer.callHash)
break
}
}
})
// Set the remote SessionDescription
err = peerConnection.SetRemoteDescription(mainOffer.offer)
if err != nil {
log.Println("Set remote description error!", err)
askForClose(mainOffer.callHash)
return
}
// Create answer
answer, err := peerConnection.CreateAnswer(nil)
if err != nil {
log.Println("Create answer error!", err)
askForClose(mainOffer.callHash)
return
}
// Sets the LocalDescription, and starts our UDP listeners
err = peerConnection.SetLocalDescription(answer)
if err != nil {
log.Println("Set local description error!", err)
askForClose(mainOffer.callHash)
return
}
// Forward ice candidates
peerConnection.OnICECandidate(func(c *webrtc.ICECandidate) {
if c != nil {
sendSignal(mainOffer.callHash, mainOffer.peerID, c.ToJSON())
}
})
// Send anwser
sendSignal(mainOffer.callHash, mainOffer.peerID, answer)
2020-04-11 08:30:28 +00:00
}