Refactor signal receiving in server.

There's one FIFO reader goroutine instead of one per bindaddr.
makePeerConnectionFromOffer gives you a PeerConnection with an answer
and also sets up callbacks to pass a webRTCConn to datachannelHandler
when ready.
This commit is contained in:
David Fifield 2016-01-18 21:13:08 -08:00
parent a77ae6b771
commit ac8669b38f

View file

@ -81,6 +81,8 @@ func (c *webRTCConn) SetWriteDeadline(t time.Time) error {
} }
func datachannelHandler(conn *webRTCConn) { func datachannelHandler(conn *webRTCConn) {
defer conn.Close()
handlerChan <- 1 handlerChan <- 1
defer func() { defer func() {
handlerChan <- -1 handlerChan <- -1
@ -91,58 +93,104 @@ func datachannelHandler(conn *webRTCConn) {
log.Printf("Failed to connect to ORPort: " + err.Error()) log.Printf("Failed to connect to ORPort: " + err.Error())
return return
} }
//defer or.Close() defer or.Close()
pr, pw := io.Pipe() copyLoop(conn, or)
conn.pr = pr
dc := conn.dc
dc.OnOpen = func() {
log.Println("OnOpen channel")
}
dc.OnClose = func() {
log.Println("OnClose channel")
pw.Close()
}
dc.OnMessage = func(msg []byte) {
// log.Printf("OnMessage channel %d %+q", len(msg), msg)
log.Printf("OnMessage <--- %d bytes", len(msg))
n, err := pw.Write(msg)
if err != nil {
pw.CloseWithError(err)
}
if n != len(msg) {
panic("short write")
}
}
go copyLoop(conn, or)
} }
func makePeerConnection(config *webrtc.Configuration) (*webrtc.PeerConnection, error) { // Create a PeerConnection from an SDP offer. Blocks until the gathering of ICE
pc, err := webrtc.NewPeerConnection(config) // candidates is complete and and answer is available in LocalDescription.
// Installs an OnDataChannel callback that creates a webRTCConn and passes it to
// datachannelHandler.
func makePeerConnectionFromOffer(sdp *webrtc.SessionDescription, config *webrtc.Configuration) (*webrtc.PeerConnection, error) {
errChan := make(chan error)
answerChan := make(chan *webrtc.SessionDescription)
pc, err := webrtc.NewPeerConnection(config)
if err != nil { if err != nil {
log.Printf("NewPeerConnection: %s", err) return nil, fmt.Errorf("accept: NewPeerConnection: %s", err)
return nil, err
} }
pc.OnNegotiationNeeded = func() { pc.OnNegotiationNeeded = func() {
panic("OnNegotiationNeeded") panic("OnNegotiationNeeded")
} }
pc.OnIceComplete = func() {
answerChan <- pc.LocalDescription()
}
pc.OnDataChannel = func(dc *data.Channel) { pc.OnDataChannel = func(dc *data.Channel) {
log.Println("OnDataChannel") log.Println("OnDataChannel")
datachannelHandler(&webRTCConn{pc: pc, dc: dc})
pr, pw := io.Pipe()
dc.OnOpen = func() {
log.Println("OnOpen channel")
}
dc.OnClose = func() {
log.Println("OnClose channel")
pw.Close()
}
dc.OnMessage = func(msg []byte) {
log.Printf("OnMessage <--- %d bytes", len(msg))
n, err := pw.Write(msg)
if err != nil {
pw.CloseWithError(err)
}
if n != len(msg) {
panic("short write")
}
}
conn := &webRTCConn{pc: pc, dc: dc, pr: pr}
go datachannelHandler(conn)
} }
pc.OnIceComplete = func() {
log.Printf("----------------") err = pc.SetRemoteDescription(sdp)
fmt.Fprintln(logFile, pc.LocalDescription().Serialize()) if err != nil {
log.Printf("----------------") pc.Close()
return nil, fmt.Errorf("accept: SetRemoteDescription: %s", err)
} }
log.Println("sdp offer successfully received.")
go func() {
log.Println("Generating answer...")
answer, err := pc.CreateAnswer() // blocking
if err != nil {
errChan <- err
return
}
err = pc.SetLocalDescription(answer)
if err != nil {
errChan <- err
return
}
}()
// Wait until answer is ready.
select {
case err = <-errChan:
pc.Close()
return nil, err
case <-answerChan:
}
return pc, nil return pc, nil
} }
func readSignalingMessages(signalChan chan *webrtc.SessionDescription, f *os.File) { // Create a signaling named pipe and feed offers from it into
s := bufio.NewScanner(f) // makePeerConnectionFromOffer.
func receiveSignalsFIFO(filename string, config *webrtc.Configuration) error {
err := syscall.Mkfifo(filename, 0600)
if err != nil {
if err.(syscall.Errno) != syscall.EEXIST {
return err
}
}
signalFile, err := os.OpenFile(filename, os.O_RDONLY, 0600)
if err != nil {
return err
}
defer signalFile.Close()
s := bufio.NewScanner(signalFile)
for s.Scan() { for s.Scan() {
msg := s.Text() msg := s.Text()
sdp := webrtc.DeserializeSessionDescription(msg) sdp := webrtc.DeserializeSessionDescription(msg)
@ -150,62 +198,18 @@ func readSignalingMessages(signalChan chan *webrtc.SessionDescription, f *os.Fil
log.Printf("ignoring invalid signal message %+q", msg) log.Printf("ignoring invalid signal message %+q", msg)
continue continue
} }
signalChan <- sdp
continue
}
if err := s.Err(); err != nil {
log.Printf("signal FIFO: %s", err)
}
}
func generateAnswer(pc *webrtc.PeerConnection) { pc, err := makePeerConnectionFromOffer(sdp, config)
fmt.Println("Generating answer...") if err != nil {
answer, err := pc.CreateAnswer() // blocking log.Printf("makePeerConnectionFromOffer: %s", err)
if err != nil { continue
fmt.Println(err)
return
}
pc.SetLocalDescription(answer)
}
func listenWebRTC(config *webrtc.Configuration, signal string) (*os.File, error) {
err := syscall.Mkfifo(signal, 0600)
if err != nil {
if err.(syscall.Errno) != syscall.EEXIST {
return nil, err
} }
// Write offer to log for manual signaling.
log.Printf("----------------")
fmt.Fprintln(logFile, pc.LocalDescription().Serialize())
log.Printf("----------------")
} }
signalFile, err := os.OpenFile(signal, os.O_RDONLY, 0600) return s.Err()
if err != nil {
return nil, err
}
//defer signalFile.Close()
var signalChan = make(chan *webrtc.SessionDescription)
go func() {
for {
select {
case sdp := <-signalChan:
pc, err := makePeerConnection(config)
if err != nil {
log.Printf("makePeerConnection: %s", err)
break
}
err = pc.SetRemoteDescription(sdp)
if err != nil {
fmt.Println("ERROR", err)
break
}
fmt.Println("sdp offer successfully received.")
go generateAnswer(pc)
}
}
}()
go readSignalingMessages(signalChan, signalFile)
log.Printf("waiting for offer")
return signalFile, nil
} }
func main() { func main() {
@ -228,18 +232,19 @@ func main() {
webRTCConfig := webrtc.NewConfiguration(webrtc.OptionIceServer("stun:stun.l.google.com:19302")) webRTCConfig := webrtc.NewConfiguration(webrtc.OptionIceServer("stun:stun.l.google.com:19302"))
listeners := make([]*os.File, 0) // Start FIFO-based signaling receiver.
go func() {
err := receiveSignalsFIFO("signal", webRTCConfig)
if err != nil {
log.Printf("receiveSignalsFIFO: %s", err)
}
}()
for _, bindaddr := range ptInfo.Bindaddrs { for _, bindaddr := range ptInfo.Bindaddrs {
switch bindaddr.MethodName { switch bindaddr.MethodName {
case ptMethodName: case ptMethodName:
ln, err := listenWebRTC(webRTCConfig, "signal") // meh
if err != nil {
pt.SmethodError(bindaddr.MethodName, err.Error())
break
}
bindaddr.Addr.Port = 12345 // lies!!! bindaddr.Addr.Port = 12345 // lies!!!
pt.Smethod(bindaddr.MethodName, bindaddr.Addr) pt.Smethod(bindaddr.MethodName, bindaddr.Addr)
listeners = append(listeners, ln)
default: default:
pt.SmethodError(bindaddr.MethodName, "no such method") pt.SmethodError(bindaddr.MethodName, "no such method")
} }
@ -260,9 +265,6 @@ func main() {
case sig = <-sigChan: case sig = <-sigChan:
} }
} }
for _, ln := range listeners {
ln.Close()
}
if sig == syscall.SIGTERM { if sig == syscall.SIGTERM {
return return