Mirrors/cloud-game
1
0
Fork 0
mirror of https://github.com/giongto35/cloud-game.git synced 2026-07-22 09:37:09 +00:00
Code Issues Projects Releases Packages Wiki Activity

Merge branch 'master' into frontend2

Browse source
This commit is contained in:
trichimtrich 2019-05-15 02:10:18 +08:00
commit acff5d8bc9
18 changed files with 667 additions and 349 deletions
Download patch file Download diff file Expand all files Collapse all files

11
cmd/main.go
View file

@ -6,6 +6,7 @@ import (
"math/rand"
"net/http"
_ "net/http/pprof"
"runtime"
"strings"
"time"
@ -65,12 +66,22 @@ func initializeServer() {
http.ListenAndServe(":"+*config.Port, nil)
}
func monitor() {
c := time.Tick(time.Second)
for range c {
log.Printf("#goroutines: %d\n", runtime.NumGoroutine())
}
}
func main() {
flag.Parse()
log.Println("Usage: ./game [-debug]")
rand.Seed(time.Now().UTC().UnixNano())
//if *config.IsMonitor {
go monitor()
//}
// There are two server mode
// Overlord is coordinator. If the OvelordHost Param is `overlord`, we spawn a new host as Overlord.
// else we spawn new server as normal server connecting to OverlordHost.

270
cmd/main_test.go
View file

@ -5,8 +5,10 @@ import (
"log"
"net/http"
"net/http/httptest"
"runtime"
"strings"
"testing"
"time"
"github.com/giongto35/cloud-game/cws"
"github.com/giongto35/cloud-game/handler"
@ -62,6 +64,7 @@ func initClient(t *testing.T, host string) (client *cws.Client) {
t.Fatalf("%v", err)
}
handshakedone := make(chan struct{})
// Simulate peerconnection initialization from client
fmt.Println("Simulating PeerConnection")
peerConnection, err := webrtc.NewPeerConnection(webrtcconfig)
@ -79,7 +82,6 @@ func initClient(t *testing.T, host string) (client *cws.Client) {
if err != nil {
panic(err)
}
// Send offer to server
log.Println("Browser Client")
client = cws.NewClient(ws)
@ -93,7 +95,7 @@ func initClient(t *testing.T, host string) (client *cws.Client) {
fmt.Println("Waiting sdp...")
client.Receive("sdp", func(resp cws.WSPacket) cws.WSPacket {
fmt.Println("received", resp.Data)
log.Println("Received SDP", resp.Data, "client: ", client)
answer := webrtc.SessionDescription{}
gamertc.Decode(resp.Data, &answer)
// Apply the answer as the remote description
@ -102,6 +104,9 @@ func initClient(t *testing.T, host string) (client *cws.Client) {
panic(err)
}
// TODO: may block in the second call
handshakedone <- struct{}{}
return cws.EmptyPacket
})
@ -124,18 +129,27 @@ func initClient(t *testing.T, host string) (client *cws.Client) {
if err != nil {
panic(err)
}
log.Println("return offer", offer)
log.Println("return offer")
return cws.WSPacket{
ID: "initwebrtc",
Data: gamertc.Encode(offer),
}
})
<-handshakedone
return client
// If receive roomID, the server is running correctly
}
func TestSingleServerNoOverlord(t *testing.T) {
/*
Case scenario:
- A server X are initilized
- Client join room with no coordinator
Expected behavior:
- Room received not empty
*/
// Init slave server
s := initServer(t, nil)
defer s.Close()
@ -159,10 +173,19 @@ func TestSingleServerNoOverlord(t *testing.T) {
fmt.Println("RoomID should not be empty")
t.Fail()
}
time.Sleep(3 * time.Second)
fmt.Println("Done")
}
func TestSingleServerOneOverlord(t *testing.T) {
/*
Case scenario:
- A server X are initilized
- Client join room with coordinator
Expected behavior:
- Room received not empty.
*/
o := initOverlord()
defer o.Close()
@ -192,10 +215,22 @@ func TestSingleServerOneOverlord(t *testing.T) {
fmt.Println("RoomID should not be empty")
t.Fail()
}
time.Sleep(time.Second)
fmt.Println("Done")
}
func TestTwoServerOneOverlord(t *testing.T) {
/*
Case scenario:
- Two server X, Y are initilized
- Client A creates a room on server X
- Client B creates a room on server Y
- Client B join a room created by A
Expected behavior:
- Bridge connection will be conducted between server Y and X
- Client B can join a room hosted on A
*/
o := initOverlord()
defer o.Close()
@ -205,7 +240,6 @@ func TestTwoServerOneOverlord(t *testing.T) {
defer s1.Close()
oconn2 := connectTestOverlordServer(t, o.URL)
// TODO: two different oconn
s2 := initServer(t, oconn2)
defer s2.Close()
@ -234,6 +268,7 @@ func TestTwoServerOneOverlord(t *testing.T) {
// Client2 trying to create a random room and later join the the room on server1
client2 := initClient(t, s2.URL)
defer client2.Close()
// Wait
// Doing the same create local room.
localRoomID := make(chan string)
client2.Send(cws.WSPacket{
@ -251,6 +286,7 @@ func TestTwoServerOneOverlord(t *testing.T) {
fmt.Println("Request the room from server 1", remoteRoomID)
log.Println("Server2 trying to join server1 room")
// After trying loging in to one session, login to other with the roomID
bridgeRoom := make(chan string)
client2.Send(cws.WSPacket{
ID: "start",
Data: "Contra.nes",
@ -258,14 +294,32 @@ func TestTwoServerOneOverlord(t *testing.T) {
PlayerIndex: 1,
}, func(resp cws.WSPacket) {
fmt.Println("RoomID:", resp.RoomID)
roomID <- resp.RoomID
bridgeRoom <- resp.RoomID
})
<-bridgeRoom
//respRoomID := <-bridgeRoom
//if respRoomID == "" {
//fmt.Println("The room ID should be equal to the saved room")
//t.Fail()
//}
// If receive roomID, the server is running correctly
time.Sleep(time.Second)
fmt.Println("Done")
}
func TestReconnectRoomNoOverlord(t *testing.T) {
/*
Case scenario:
- A server X is initialized
- Client A creates a room K on server X
- Server X is turned down, Client is closed
- Spawn a new server and a new client connecting to the same room K
Expected behavior:
- The game should be continue
TODO: Current test just make sure the game is running, not check if the game is the same
*/
// Init slave server
s := initServer(t, nil)
@ -319,71 +373,171 @@ func TestReconnectRoomNoOverlord(t *testing.T) {
t.Fail()
}
time.Sleep(time.Second)
fmt.Println("Done")
}
// This test currently doesn't work
//func TestReconnectRoomWithOverlord(t *testing.T) {
//o := initOverlord()
//defer o.Close()
func TestReconnectRoomWithOverlord(t *testing.T) {
/*
Case scenario:
- A server X is initialized connecting to overlord
- Client A creates a room K on server X
- Server X is turned down, Client is closed
- Spawn a new server and a new client connecting to the same room K
Expected behavior:
- The game should be continue
TODO: Current test just make sure the game is running, not check if the game is the same
*/
//oconn := connectTestOverlordServer(t, o.URL)
//defer oconn.Close()
//// Init slave server
//s := initServer(t, oconn)
o := initOverlord()
defer o.Close()
//client := initClient(t, s.URL)
oconn := connectTestOverlordServer(t, o.URL)
// Init slave server
s := initServer(t, oconn)
//fmt.Println("Sending start...")
//roomID := make(chan string)
//client.Send(cws.WSPacket{
//ID: "start",
//Data: "Contra.nes",
//RoomID: "",
//PlayerIndex: 1,
//}, func(resp cws.WSPacket) {
//fmt.Println("RoomID:", resp.RoomID)
//roomID <- resp.RoomID
//})
client := initClient(t, s.URL)
//saveRoomID := <-roomID
//if saveRoomID == "" {
//fmt.Println("RoomID should not be empty")
//t.Fail()
//}
fmt.Println("Sending start...")
roomID := make(chan string)
client.Send(cws.WSPacket{
ID: "start",
Data: "Contra.nes",
RoomID: "",
PlayerIndex: 1,
}, func(resp cws.WSPacket) {
fmt.Println("RoomID:", resp.RoomID)
roomID <- resp.RoomID
})
//log.Println("Closing room and server")
//client.Close()
//s.Close()
//// Close server and reconnect
saveRoomID := <-roomID
if saveRoomID == "" {
fmt.Println("RoomID should not be empty")
t.Fail()
}
//log.Println("Server respawn")
//// Init slave server
//s = initServer(t, oconn)
//defer s.Close()
log.Println("Closing room and server")
client.Close()
s.Close()
oconn.Close()
//client = initClient(t, s.URL)
//defer client.Close()
// Close server and reconnect
//fmt.Println("Re-access room ", saveRoomID)
//roomID = make(chan string)
//client.Send(cws.WSPacket{
//ID: "start",
//Data: "Contra.nes",
//RoomID: saveRoomID,
//PlayerIndex: 1,
//}, func(resp cws.WSPacket) {
//fmt.Println("RoomID:", resp.RoomID)
//roomID <- resp.RoomID
//})
log.Println("Server respawn")
// Init slave server again
oconn = connectTestOverlordServer(t, o.URL)
defer oconn.Close()
s = initServer(t, oconn)
defer s.Close()
//respRoomID := <-roomID
//if respRoomID == "" || respRoomID != saveRoomID {
//fmt.Println("The room ID should be equal to the saved room")
//t.Fail()
//}
client = initClient(t, s.URL)
defer client.Close()
//fmt.Println("Done")
fmt.Println("Re-access room ", saveRoomID)
roomID = make(chan string)
client.Send(cws.WSPacket{
ID: "start",
Data: "Contra.nes",
RoomID: saveRoomID,
PlayerIndex: 1,
}, func(resp cws.WSPacket) {
fmt.Println("RoomID:", resp.RoomID)
roomID <- resp.RoomID
})
//}
respRoomID := <-roomID
if respRoomID == "" || respRoomID != saveRoomID {
fmt.Println("The room ID should be equal to the saved room")
t.Fail()
}
time.Sleep(time.Second)
fmt.Println("Done")
}
func TestRejoinNoOverlordMultiple(t *testing.T) {
/*
Case scenario:
- A server X without connecting to overlord
- Client A keeps creating a new room
Expected behavior:
- The game should running normally
*/
// Init slave server
s := initServer(t, nil)
defer s.Close()
fmt.Println("Num goRoutine before start: ", runtime.NumGoroutine())
client := initClient(t, s.URL)
for i := 0; i < 100; i++ {
fmt.Println("Sending start...")
// Keep starting game
roomID := make(chan string)
client.Send(cws.WSPacket{
ID: "start",
Data: "Contra.nes",
RoomID: "",
PlayerIndex: 1,
}, func(resp cws.WSPacket) {
fmt.Println("RoomID:", resp.RoomID)
roomID <- resp.RoomID
})
respRoomID := <-roomID
if respRoomID == "" {
fmt.Println("The room ID should be equal to the saved room")
t.Fail()
}
}
time.Sleep(time.Second)
fmt.Println("Num goRoutine should be small: ", runtime.NumGoroutine())
fmt.Println("Done")
}
func TestRejoinWithOverlordMultiple(t *testing.T) {
/*
Case scenario:
- A server X is initialized connecting to overlord
- Client A keeps creating a new room
Expected behavior:
- The game should running normally
*/
// Init slave server
o := initOverlord()
defer o.Close()
oconn := connectTestOverlordServer(t, o.URL)
// Init slave server
s := initServer(t, oconn)
defer s.Close()
fmt.Println("Num goRoutine before start: ", runtime.NumGoroutine())
client := initClient(t, s.URL)
for i := 0; i < 100; i++ {
fmt.Println("Sending start...")
// Keep starting game
roomID := make(chan string)
client.Send(cws.WSPacket{
ID: "start",
Data: "Contra.nes",
RoomID: "",
PlayerIndex: 1,
}, func(resp cws.WSPacket) {
fmt.Println("RoomID:", resp.RoomID)
roomID <- resp.RoomID
})
respRoomID := <-roomID
if respRoomID == "" {
fmt.Println("The room ID should be equal to the saved room")
t.Fail()
}
}
fmt.Println("Num goRoutine should be small: ", runtime.NumGoroutine())
fmt.Println("Done")
}

28
cws/cws.go
View file

@ -21,6 +21,8 @@ type Client struct {
sendCallbackLock sync.Mutex
// recvCallback is callback when receive based on ID of the packet
recvCallback map[string]func(req WSPacket)
Done chan struct{}
}
type WSPacket struct {
@ -49,6 +51,8 @@ func NewClient(conn *websocket.Conn) *Client {
sendCallback: sendCallback,
recvCallback: recvCallback,
Done: make(chan struct{}),
}
}
@ -120,43 +124,45 @@ func (c *Client) Heartbeat() {
timer := time.Tick(time.Second)
for range timer {
select {
case <-c.Done:
log.Println("Close heartbeat")
return
default:
}
c.Send(WSPacket{ID: "heartbeat"}, nil)
}
}
func (c *Client) Listen() {
for {
//log.Println("Waiting for message ...")
_, rawMsg, err := c.conn.ReadMessage()
if err != nil {
log.Println("[!] read:", err)
// TODO: Check explicit disconnect error to break
close(c.Done)
break
}
wspacket := WSPacket{}
err = json.Unmarshal(rawMsg, &wspacket)
//log.Println( "Received: ", wspacket)
if err != nil {
continue
}
// Check if some async send is waiting for the response based on packetID
// TODO: Change to read lock
c.sendCallbackLock.Lock()
//log.Println("Listening: Callback waiting list: ", c.id, c.sendCallback)
// TODO: Change to read lock.
//c.sendCallbackLock.Lock()
callback, ok := c.sendCallback[wspacket.PacketID]
//log.Println("Has callback: ", ok, "ClientID: ", c.id, "PacketID ", wspacket.PacketID)
c.sendCallbackLock.Unlock()
//c.sendCallbackLock.Unlock()
if ok {
go callback(wspacket)
c.sendCallbackLock.Lock()
//log.Println("Deleteing Packet ", wspacket.PacketID)
//c.sendCallbackLock.Lock()
delete(c.sendCallback, wspacket.PacketID)
c.sendCallbackLock.Unlock()
//c.sendCallbackLock.Unlock()
// Skip receiveCallback to avoid duplication
continue
}
//log.Println("Listening: Callback waiting list: ", c.id, c.recvCallback)
// Check if some receiver with the ID is registered
if callback, ok := c.recvCallback[wspacket.ID]; ok {
go callback(wspacket)

37
emulator/director.go
View file

@ -19,21 +19,18 @@ type Director struct {
Done chan struct{}
roomID string
// Hash represents a game state (roomID, gamePath).
// It is used for save file name
hash string
}
const FPS = 60
func NewDirector(roomID string, imageChannel chan *image.RGBA, audioChannel chan float32, inputChannel chan int) *Director {
// TODO: return image channel from where it write
director := Director{}
director.Done = make(chan struct{})
director.audioChannel = audioChannel
director.imageChannel = imageChannel
director.inputChannel = inputChannel
director.roomID = roomID
director.hash = ""
return &director
}
@ -68,6 +65,7 @@ func (d *Director) Start(paths []string) {
// audio := NewAudio()
// audio.Start()
// d.audio = audio
log.Println("Start game: ", paths)
if len(paths) == 1 {
d.PlayGame(paths[0])
@ -81,12 +79,13 @@ L:
for range c {
// for {
// quit game
// TODO: Anyway not using select because it will slow down
// TODO: How to not using select because it will slow down
select {
// if there is event from close channel => the game is ended
//case input := <-d.inputChannel:
//d.UpdateInput(input)
case <-d.Done:
log.Println("Closing Director")
break L
default:
}
@ -94,27 +93,22 @@ L:
d.Step()
}
d.SetView(nil)
log.Println("Closed Director")
}
func (d *Director) PlayGame(path string) {
// Generate hash that is indentifier of a room (game path + ropomID)
hash, err := hashFile(path, d.roomID)
if err != nil {
log.Fatalln(err)
}
d.hash = hash
console, err := nes.NewConsole(path)
if err != nil {
log.Fatalln(err)
log.Println("Err: Cannot load path, Got:", err)
}
// Set GameView as current view
d.SetView(NewGameView(console, path, hash, d.imageChannel, d.audioChannel, d.inputChannel))
d.SetView(NewGameView(console, path, d.roomID, d.imageChannel, d.audioChannel, d.inputChannel))
}
// SaveGame creates save events and doing extra step for load
func (d *Director) SaveGame(saveExtraFunc func() error) error {
if d.hash != "" {
d.view.Save(d.hash, saveExtraFunc)
if d.roomID != "" {
d.view.Save(d.roomID, saveExtraFunc)
return nil
} else {
return nil
@ -122,21 +116,16 @@ func (d *Director) SaveGame(saveExtraFunc func() error) error {
}
// LoadGame creates load events and doing extra step for load
func (d *Director) LoadGame(loadExtraFunc func() error) error {
if d.hash != "" {
d.view.Load(d.hash, loadExtraFunc)
func (d *Director) LoadGame() error {
if d.roomID != "" {
d.view.Load(d.roomID)
return nil
} else {
return nil
}
}
// GetHash return hash
func (d *Director) GetHash() string {
return d.hash
}
// GetHashPath return the full path to hash file
func (d *Director) GetHashPath() string {
return savePath(d.hash)
return savePath(d.roomID)
}

39
emulator/gameview.go
View file

@ -40,8 +40,9 @@ const (
type GameView struct {
console *nes.Console
title string
hash string
// saveFile is the filename gameview save to
saveFile string
// equivalent to the list key pressed const above
keyPressed [NumKeys * 2]bool
@ -58,11 +59,11 @@ type job struct {
extraFunc func() error
}
func NewGameView(console *nes.Console, title, hash string, imageChannel chan *image.RGBA, audioChannel chan float32, inputChannel chan int) *GameView {
func NewGameView(console *nes.Console, title, saveFile string, imageChannel chan *image.RGBA, audioChannel chan float32, inputChannel chan int) *GameView {
gameview := &GameView{
console: console,
title: title,
hash: hash,
saveFile: saveFile,
keyPressed: [NumKeys * 2]bool{false},
imageChannel: imageChannel,
audioChannel: audioChannel,
@ -85,7 +86,12 @@ func NewGameView(console *nes.Console, title, hash string, imageChannel chan *im
// ListenToInputChannel listen from input channel streamm, which is exposed to WebRTC session
func (view *GameView) ListenToInputChannel() {
for {
keysInBinary := <-view.inputChannel
// Adding ok here make thing slowdown
// TODO: Investigate slow
keysInBinary, ok := <-view.inputChannel
if !ok {
return
}
for i := 0; i < NumKeys*2; i++ {
b := ((keysInBinary & 1) == 1)
view.keyPressed[i] = (view.keyPressed[i] && b) || b
@ -99,18 +105,17 @@ func (view *GameView) Enter() {
view.console.SetAudioSampleRate(SampleRate)
view.console.SetAudioChannel(view.audioChannel)
// load state if the hash file existed (Join the old room)
if err := view.console.LoadState(savePath(view.hash)); err == nil {
// load state if the saveFile file existed in the server (Join the old room)
if err := view.console.LoadState(savePath(view.saveFile)); err == nil {
return
} else {
view.console.Reset()
}
//view.console.Reset()
// load sram
cartridge := view.console.Cartridge
if cartridge.Battery != 0 {
if sram, err := readSRAM(sramPath(view.hash)); err == nil {
if sram, err := readSRAM(sramPath(view.saveFile)); err == nil {
cartridge.SRAM = sram
}
}
@ -123,8 +128,12 @@ func (view *GameView) Exit() {
// save sram
cartridge := view.console.Cartridge
if cartridge.Battery != 0 {
writeSRAM(sramPath(view.hash), cartridge.SRAM)
writeSRAM(sramPath(view.saveFile), cartridge.SRAM)
}
// close producer
close(view.imageChannel)
close(view.audioChannel)
}
// Update is called for every period of time, dt is the elapsed time from the last frame
@ -144,16 +153,16 @@ func (view *GameView) Update(t, dt float64) {
func (view *GameView) Save(hash string, extraSaveFunc func() error) {
// put saving event to queue, process in updateEvent
view.savingJob = &job{
path: savePath(view.hash),
path: savePath(view.saveFile),
extraFunc: extraSaveFunc,
}
}
func (view *GameView) Load(path string, extraLoadFunc func() error) {
func (view *GameView) Load(path string) {
// put saving event to queue, process in updateEvent
view.loadingJob = &job{
path: savePath(view.hash),
extraFunc: extraLoadFunc,
path: savePath(view.saveFile),
extraFunc: nil,
}
}
@ -169,7 +178,9 @@ func (view *GameView) UpdateEvents() {
if view.loadingJob != nil {
view.console.LoadState(view.loadingJob.path)
// Run extra function (online saving for example)
go view.loadingJob.extraFunc()
if view.loadingJob.extraFunc != nil {
go view.loadingJob.extraFunc()
}
view.loadingJob = nil
}
}

16
emulator/util.go
View file

@ -2,7 +2,6 @@
package emulator
import (
"crypto/md5"
"encoding/binary"
"fmt"
"image"
@ -10,7 +9,6 @@ import (
"image/draw"
"image/gif"
"image/png"
"io/ioutil"
"log"
"os"
"os/user"
@ -29,6 +27,11 @@ func init() {
homeDir = u.HomeDir
}
// Public call to get savePath
func GetSavePath(roomID string) string {
return savePath(roomID)
}
func sramPath(hash string) string {
return homeDir + "/.nes/sram/" + hash + ".dat"
}
@ -45,15 +48,6 @@ func combineButtons(a, b [8]bool) [8]bool {
return result
}
// hashFile : signature of a room, maybe not need path
func hashFile(path string, roomID string) (string, error) {
data, err := ioutil.ReadFile(path)
if err != nil {
return "", err
}
return fmt.Sprintf("%x", md5.Sum(append(data, []byte(roomID)...))), nil
}
func copyImage(src image.Image) *image.RGBA {
dst := image.NewRGBA(src.Bounds())
draw.Draw(dst, dst.Rect, src, image.ZP, draw.Src)

BIN
games/Mega Man.nes vendored
View file

Binary file not shown.

8
handler/browser.go
View file

@ -14,7 +14,8 @@ type BrowserClient struct {
*cws.Client
}
func (s *Session) RegisterBrowserClient() {
// RouteBrowser are all routes server received from browser
func (s *Session) RouteBrowser() {
browserClient := s.BrowserClient
browserClient.Receive("heartbeat", func(resp cws.WSPacket) cws.WSPacket {
@ -25,7 +26,10 @@ func (s *Session) RegisterBrowserClient() {
log.Println("Received user SDP")
localSession, err := s.peerconnection.StartClient(resp.Data, config.Width, config.Height)
if err != nil {
log.Fatalln(err)
if err != nil {
log.Println("Error: Cannot create new webrtc session", err)
return cws.EmptyPacket
}
}
return cws.WSPacket{

32
handler/handlers.go
View file

@ -1,7 +1,6 @@
package handler
import (
"fmt"
"io/ioutil"
"log"
"net/http"
@ -75,17 +74,20 @@ func (h *Handler) GetWeb(w http.ResponseWriter, r *http.Request) {
// WS handles normal traffic (from browser to host)
func (h *Handler) WS(w http.ResponseWriter, r *http.Request) {
c, err := upgrader.Upgrade(w, r, nil)
defer c.Close()
defer func() {
if r := recover(); r != nil {
log.Println("Warn: Something wrong. Recovered in f", r)
}
}()
c, err := upgrader.Upgrade(w, r, nil)
if err != nil {
log.Print("[!] WS upgrade:", err)
return
}
defer c.Close()
client := NewBrowserClient(c)
//client := NewClient(c)
////sessionID := strconv.Itoa(rand.Int())
sessionID := uuid.Must(uuid.NewV4()).String()
wssession := &Session{
ID: sessionID,
@ -94,14 +96,16 @@ func (h *Handler) WS(w http.ResponseWriter, r *http.Request) {
peerconnection: webrtc.NewWebRTC(),
handler: h,
}
defer wssession.Close()
if wssession.OverlordClient != nil {
wssession.RegisterOverlordClient()
wssession.RouteOverlord()
go wssession.OverlordClient.Heartbeat()
go wssession.OverlordClient.Listen()
}
wssession.RegisterBrowserClient()
fmt.Println("oclient : ", h.oClient)
wssession.RouteBrowser()
log.Println("oclient : ", h.oClient)
wssession.BrowserClient.Send(cws.WSPacket{
ID: "gamelist",
@ -109,10 +113,22 @@ func (h *Handler) WS(w http.ResponseWriter, r *http.Request) {
}, nil)
wssession.BrowserClient.Listen()
// TODO: Use callback
// If peerconnection is done (client.Done is signalled), we close peerconnection
go func() {
for {
<-client.Done
h.detachPeerConn(wssession.peerconnection)
return
}
}()
}
// Detach peerconnection detach/remove a peerconnection from current room
func (h *Handler) detachPeerConn(pc *webrtc.WebRTC) {
log.Println("Detach peerconnection")
roomID := pc.RoomID
room := h.getRoom(roomID)
if room == nil {

11
handler/overlord.go
View file

@ -29,8 +29,8 @@ func NewOverlordClient(oc *websocket.Conn) *OverlordClient {
return oclient
}
// RegisterOverlordClient routes overlord Client
func (s *Session) RegisterOverlordClient() {
// RouteOverlord are all routes server received from overlord
func (s *Session) RouteOverlord() {
oclient := s.OverlordClient
// Received from overlord the serverID
@ -58,7 +58,8 @@ func (s *Session) RegisterOverlordClient() {
oclient.peerconnections[resp.SessionID] = peerconnection
if err != nil {
log.Fatalln(err)
log.Println("Error: Cannot create new webrtc session", err)
return cws.EmptyPacket
}
return cws.WSPacket{
@ -74,7 +75,7 @@ func (s *Session) RegisterOverlordClient() {
"start",
func(resp cws.WSPacket) (req cws.WSPacket) {
log.Println("Received a start request from overlord")
log.Println("Add the connection to current room on the host")
log.Println("Add the connection to current room on the host ", resp.SessionID)
peerconnection := oclient.peerconnections[resp.SessionID]
log.Println("start session")
@ -125,7 +126,7 @@ func (s *Session) bridgeConnection(serverID string, gameName string, roomID stri
// Ask overlord to relay SDP packet to serverID
resp.TargetHostID = serverID
log.Println("Sending offer to overlord to relay message to target host", resp.TargetHostID, "with payload", resp)
log.Println("Sending offer to overlord to relay message to target host", resp.TargetHostID, "with payload")
remoteTargetSDP := s.OverlordClient.SyncSend(resp)
log.Println("Got back remote host SDP, sending to browser")
// Send back remote SDP of remote server to browser

124
handler/room/media.go
View file

@ -26,46 +26,52 @@ func (r *Room) startAudio() {
// fanout Audio
for {
select {
case <-r.Done:
r.Close()
sample, ok := <-r.audioChannel
if !ok {
// Just for guarding
log.Println("Warn: Room ", r.ID, " audio channel closed unexpectedly")
return
case sample := <-r.audioChannel:
pcm[idx] = sample
idx++
if idx == len(pcm) {
data := make([]byte, 640)
}
if r.Done {
log.Println("Room ", r.ID, " audio channel closed")
return
}
n, err := enc.EncodeFloat32(pcm, data)
// TODO: Use worker pool for encoding
pcm[idx] = sample
idx++
if idx == len(pcm) {
data := make([]byte, 640)
if err != nil {
log.Println("[!] Failed to decode")
continue
}
data = data[:n]
data = append(data, count)
r.sessionsLock.Lock()
for _, webRTC := range r.rtcSessions {
// Client stopped
if webRTC.IsClosed() {
continue
}
// encode frame
// fanout audioChannel
if webRTC.IsConnected() {
// NOTE: can block here
webRTC.AudioChannel <- data
}
//isRoomRunning = true
}
r.sessionsLock.Unlock()
idx = 0
count = (count + 1) & 0xff
n, err := enc.EncodeFloat32(pcm, data)
if err != nil {
log.Println("[!] Failed to decode")
continue
}
data = data[:n]
data = append(data, count)
// TODO: r.rtcSessions is rarely updated. Lock will hold down perf
//r.sessionsLock.Lock()
for _, webRTC := range r.rtcSessions {
// Client stopped
//if !webRTC.IsClosed() {
//continue
//}
// encode frame
// fanout audioChannel
if webRTC.IsConnected() {
// NOTE: can block here
webRTC.AudioChannel <- data
}
//isRoomRunning = true
}
//r.sessionsLock.Unlock()
idx = 0
count = (count + 1) & 0xff
}
}
}
@ -73,28 +79,34 @@ func (r *Room) startAudio() {
func (r *Room) startVideo() {
// fanout Screen
for {
select {
case <-r.Done:
r.Close()
image, ok := <-r.imageChannel
if !ok {
// Just for guarding, should not reached
log.Println("Warn: Room ", r.ID, " video channel closed unexpectedly")
return
case image := <-r.imageChannel:
yuv := util.RgbaToYuv(image)
r.sessionsLock.Lock()
for _, webRTC := range r.rtcSessions {
// Client stopped
if webRTC.IsClosed() {
continue
}
// encode frame
// fanout imageChannel
if webRTC.IsConnected() {
// NOTE: can block here
webRTC.ImageChannel <- yuv
}
}
r.sessionsLock.Unlock()
}
if r.Done {
log.Println("Room ", r.ID, " video channel closed")
return
}
// TODO: Use worker pool for encoding
yuv := util.RgbaToYuv(image)
// TODO: r.rtcSessions is rarely updated. Lock will hold down perf
//r.sessionsLock.Lock()
for _, webRTC := range r.rtcSessions {
// Client stopped
//if webRTC.IsClosed() {
//continue
//}
// encode frame
// fanout imageChannel
if webRTC.IsConnected() {
// NOTE: can block here
webRTC.ImageChannel <- yuv
}
}
//r.sessionsLock.Unlock()
}
}

140
handler/room/room.go
View file

@ -6,6 +6,7 @@ import (
"io/ioutil"
"log"
"math/rand"
"os"
"strconv"
"sync"
@ -23,7 +24,7 @@ type Room struct {
audioChannel chan float32
inputChannel chan int
// Done channel is to fire exit event when there is no webRTC session running
Done chan struct{}
Done bool
rtcSessions []*webrtc.WebRTC
sessionsLock *sync.Mutex
@ -57,13 +58,31 @@ func NewRoom(roomID, gamepath, gameName string, onlineStorage *storage.Client) *
rtcSessions: []*webrtc.WebRTC{},
sessionsLock: &sync.Mutex{},
director: director,
Done: make(chan struct{}),
Done: false,
onlineStorage: onlineStorage,
}
go room.startVideo()
go room.startAudio()
go director.Start([]string{gamepath + "/" + gameName})
// Check if room is on local storage, if not, pull from GCS to local storage
path := gamepath + "/" + gameName
go func(path, roomID string) {
// Check room is on local or fetch from server
savepath := emulator.GetSavePath(roomID)
log.Println("Check ", savepath, " on local : ", room.isGameOnLocal(savepath))
if !room.isGameOnLocal(savepath) {
// Fetch room from GCP to server
log.Println("Load room from online storage", savepath)
if err := room.saveOnlineRoomToLocal(roomID, savepath); err != nil {
log.Printf("Warn: Room %s is not in online storage, error %s", roomID, err)
}
}
log.Printf("Room %s started", roomID)
director.Start([]string{path})
log.Printf("Room %s ended", roomID)
}(path, roomID)
return room
}
@ -76,6 +95,11 @@ func generateRoomID() string {
return roomID
}
func (r *Room) isGameOnLocal(savepath string) bool {
_, err := os.Open(savepath)
return err == nil
}
func (r *Room) AddConnectionToRoom(peerconnection *webrtc.WebRTC, playerIndex int) {
peerconnection.AttachRoomID(r.ID)
r.rtcSessions = append(r.rtcSessions, peerconnection)
@ -83,42 +107,49 @@ func (r *Room) AddConnectionToRoom(peerconnection *webrtc.WebRTC, playerIndex in
go r.startWebRTCSession(peerconnection, playerIndex)
}
// startWebRTCSession fan-in of the same room to inputChannel
func (r *Room) startWebRTCSession(peerconnection *webrtc.WebRTC, playerIndex int) {
inputChannel := r.inputChannel
for {
select {
case <-peerconnection.Done:
r.removeSession(peerconnection)
default:
}
// Client stopped
if peerconnection.IsClosed() {
return
defer func() {
if r := recover(); r != nil {
log.Println("Warn: Recovered when sent to close inputChannel")
}
}()
// encode frame
if peerconnection.IsConnected() {
input := <-peerconnection.InputChannel
// the first 8 bits belong to player 1
// the next 8 belongs to player 2 ...
// We standardize and put it to inputChannel (16 bits)
input = input << ((uint(playerIndex) - 1) * emulator.NumKeys)
inputChannel <- input
go func() {
for {
input, ok := <-peerconnection.InputChannel
if !ok {
return
// might consider continue here
}
if peerconnection.Done || !peerconnection.IsConnected() || r.Done {
return
}
if peerconnection.IsConnected() {
// the first 8 bits belong to player 1
// the next 8 belongs to player 2 ...
// We standardize and put it to inputChannel (16 bits)
input = input << ((uint(playerIndex) - 1) * emulator.NumKeys)
select {
case r.inputChannel <- input:
default:
}
}
}
}
}()
log.Println("Peerconn done")
}
func (r *Room) CleanSession(peerconnection *webrtc.WebRTC) {
r.removeSession(peerconnection)
// TODO: Clean all channels
}
func (r *Room) removeSession(w *webrtc.WebRTC) {
fmt.Println("Cleaning session: ", w)
r.sessionsLock.Lock()
defer r.sessionsLock.Unlock()
fmt.Println("Sessions list", r.rtcSessions)
// TODO: get list of r.rtcSessions in lock
for i, s := range r.rtcSessions {
fmt.Println("found session: ", s, w)
if s.ID == w.ID {
@ -126,9 +157,12 @@ func (r *Room) removeSession(w *webrtc.WebRTC) {
fmt.Println("found session: ", len(r.rtcSessions))
// If room has no sessions, close room
// Note: this logic cannot be brought outside of forloop because we only close room if room had at least one session
if len(r.rtcSessions) == 0 {
log.Println("No session in room")
r.Done <- struct{}{}
r.Close()
// can consider sanding close to room and room do clean
//close(r.Done)
}
break
}
@ -136,14 +170,25 @@ func (r *Room) removeSession(w *webrtc.WebRTC) {
}
func (r *Room) Close() {
log.Println("Closing room", r)
r.director.Done <- struct{}{}
if r.Done {
return
}
log.Println("Closing room", r.ID)
log.Println("Closing director of room ", r.ID)
close(r.director.Done)
log.Println("Closing input of room ", r.ID)
close(r.inputChannel)
r.Done = true
// Close here is a bit wrong because this read channel
//close(r.imageChannel)
//close(r.audioChannel)
}
func (r *Room) SaveGame() error {
onlineSaveFunc := func() error {
// Try to save the game to gCloud
if err := r.onlineStorage.SaveFile(r.director.GetHash(), r.director.GetHashPath()); err != nil {
if err := r.onlineStorage.SaveFile(r.ID, r.director.GetHashPath()); err != nil {
return err
}
@ -158,27 +203,22 @@ func (r *Room) SaveGame() error {
return nil
}
func (r *Room) LoadGame() error {
// TODO: Fix, because load game always come to local, this logic is unnecessary. Move to load game
onlineLoadFunc := func() error {
log.Println("Loading game from cloud storage")
// If the game is not on local server
// Try to load from gcloud
data, err := r.onlineStorage.LoadFile(r.director.GetHash())
if err != nil {
return err
}
// Save the data fetched from gcloud to local server
ioutil.WriteFile(r.director.GetHashPath(), data, 0644)
// Reload game again
//err = r.director.LoadGame(nil)
//if err != nil {
//return err
//}
return nil
func (r *Room) saveOnlineRoomToLocal(roomID string, savepath string) error {
log.Println("Try loading game from cloud storage")
// If the game is not on local server
// Try to load from gcloud
data, err := r.onlineStorage.LoadFile(roomID)
if err != nil {
return err
}
// Save the data fetched from gcloud to local server
ioutil.WriteFile(savepath, data, 0644)
err := r.director.LoadGame(onlineLoadFunc)
return nil
}
func (r *Room) LoadGame() error {
err := r.director.LoadGame()
return err
}
@ -186,7 +226,7 @@ func (r *Room) LoadGame() error {
func (r *Room) IsRunning() bool {
// If there is running session
for _, s := range r.rtcSessions {
if !s.IsClosed() {
if s.IsConnected() {
return true
}
}

7
handler/session.go
View file

@ -5,7 +5,8 @@ import (
)
// Session represents a session connected from the browser to the current server
// It involves one connection to browser and one connection to the overlord
// It requires one connection to browser and one connection to the overlord
// connection to browser is 1-1. connection to overlord is n - 1
// Peerconnection can be from other server to ensure better latency
type Session struct {
ID string
@ -21,3 +22,7 @@ type Session struct {
RoomID string
PlayerIndex int
}
func (s *Session) Close() {
s.peerconnection.StopClient()
}

33
overlord/overlord.go
View file

@ -21,7 +21,9 @@ var upgrader = websocket.Upgrader{}
func NewServer() *Server {
return &Server{
servers: map[string]*cws.Client{},
// Mapping serverID to client
servers: map[string]*cws.Client{},
// Mapping roomID to server
roomToServer: map[string]string{},
}
}
@ -34,8 +36,6 @@ func (o *Server) WSO(w http.ResponseWriter, r *http.Request) {
log.Print("Overlord: [!] WS upgrade:", err)
return
}
defer c.Close()
// Register new server
serverID := strconv.Itoa(rand.Int())
log.Println("Overlord: A new server connected to Overlord", serverID)
@ -43,12 +43,7 @@ func (o *Server) WSO(w http.ResponseWriter, r *http.Request) {
// Register to servers map the client connection
client := cws.NewClient(c)
o.servers[serverID] = client
//wssession := &Session{
//client: client,
//peerconnection: webrtc.NewWebRTC(),
//// The server session is maintaining
//}
defer o.cleanConnection(client, serverID)
// Sendback the ID to server
client.Send(
@ -72,6 +67,7 @@ func (o *Server) WSO(w http.ResponseWriter, r *http.Request) {
// getRoom returns the server ID based on requested roomID.
client.Receive("getRoom", func(resp cws.WSPacket) cws.WSPacket {
log.Println("Overlord: Received a getroom request")
log.Println("Result: ", o.roomToServer[resp.Data])
return cws.WSPacket{
ID: "getRoom",
Data: o.roomToServer[resp.Data],
@ -84,7 +80,7 @@ func (o *Server) WSO(w http.ResponseWriter, r *http.Request) {
log.Println("Overlord: Received a relay sdp request from a host")
// TODO: Abstract
if resp.TargetHostID != serverID {
log.Println("Overlord: Sending relay sdp to target host", resp)
log.Println("Overlord: Sending relay sdp to target host")
// relay SDP to target host and get back sdp
// TODO: Async
sdp := o.servers[resp.TargetHostID].SyncSend(
@ -114,7 +110,8 @@ func (o *Server) WSO(w http.ResponseWriter, r *http.Request) {
log.Println("Overlord: Received a relay start request from a host")
// TODO: Abstract
if resp.TargetHostID != serverID {
// relay SDP to target host and get back sdp
// relay start to target host
log.Println("Sending to target host", resp.TargetHostID, " ", resp)
// TODO: Async
resp := o.servers[resp.TargetHostID].SyncSend(
resp,
@ -141,3 +138,17 @@ func (o *Server) WSO(w http.ResponseWriter, r *http.Request) {
client.Listen()
}
func (o *Server) cleanConnection(client *cws.Client, serverID string) {
log.Println("Unregister server from overlord")
// Remove serverID from servers
delete(o.servers, serverID)
// Clean all rooms connecting to that server
for roomID, roomServer := range o.roomToServer {
if roomServer == serverID {
delete(o.roomToServer, roomID)
}
}
client.Close()
}

2
static/js/ws.js vendored
View file

@ -8,7 +8,7 @@ conn = new WebSocket(`ws://${location.host}/ws`);
conn.onopen = () => {
log("WebSocket is opened. Send ping");
log("Send ping pong frequently")
// pingpongTimer = setInterval(sendPing, 1000 / PINGPONGPS)
pingpongTimer = setInterval(sendPing, 1000 / PINGPONGPS)
startWebRTC();
}

107
vpx-encoder/encoder.go
View file

@ -48,6 +48,9 @@ func NewVpxEncoder(w, h, fps, bitrate, keyframe int) (*VpxEncoder, error) {
v := &VpxEncoder{
Output: make(chan []byte, 5*chanSize),
Input: make(chan []byte, chanSize),
IsRunning: true,
Done: false,
// C
width: C.uint(w),
height: C.uint(h),
@ -66,9 +69,11 @@ func NewVpxEncoder(w, h, fps, bitrate, keyframe int) (*VpxEncoder, error) {
// VpxEncoder yuvI420 image to vp8 video
type VpxEncoder struct {
started bool
Output chan []byte // frame
Input chan []byte // yuvI420
Output chan []byte // frame
Input chan []byte // yuvI420
IsRunning bool
Done bool
// C
width C.uint
height C.uint
@ -109,58 +114,74 @@ func (v *VpxEncoder) init() error {
if C.call_vpx_codec_enc_init(&v.vpxCodexCtx, encoder, &cfg) != 0 {
return fmt.Errorf("Failed to initialize encoder")
}
v.started = true
v.IsRunning = true
go v.startLooping()
return nil
}
func (v *VpxEncoder) startLooping() {
go func() {
for {
beginEncoding := time.Now()
yuv := <-v.Input
// Add Image
v.vpxCodexIter = nil
C.vpx_img_read(&v.vpxImage, unsafe.Pointer(&yuv[0]))
var flags C.int
if v.keyFrameInterval > 0 && v.frameCount%v.keyFrameInterval == 0 {
flags |= C.VPX_EFLAG_FORCE_KF
}
if C.vpx_codec_encode(&v.vpxCodexCtx, &v.vpxImage, C.vpx_codec_pts_t(v.frameCount), 1, C.vpx_enc_frame_flags_t(flags), C.VPX_DL_REALTIME) != 0 {
fmt.Println("Failed to encode frame")
}
v.frameCount++
// Get Frame
for {
goBytes := C.get_frame_buffer(&v.vpxCodexCtx, &v.vpxCodexIter)
if goBytes.bs == nil {
break
}
bs := C.GoBytes(goBytes.bs, goBytes.size)
// if buffer is full skip frame
if len(v.Output) >= cap(v.Output) {
continue
}
v.Output <- bs
}
if *config.IsMonitor {
log.Println("Encoding time: ", time.Now().Sub(beginEncoding))
}
defer func() {
if r := recover(); r != nil {
log.Println("Warn: Recovered panic in encoding ", r)
}
}()
for {
beginEncoding := time.Now()
yuv, ok := <-v.Input
if !ok || v.Done == true {
// The first time we see IsRunning set to false, we release and return
v.Release()
return
}
// Add Image
v.vpxCodexIter = nil
C.vpx_img_read(&v.vpxImage, unsafe.Pointer(&yuv[0]))
var flags C.int
if v.keyFrameInterval > 0 && v.frameCount%v.keyFrameInterval == 0 {
flags |= C.VPX_EFLAG_FORCE_KF
}
if C.vpx_codec_encode(&v.vpxCodexCtx, &v.vpxImage, C.vpx_codec_pts_t(v.frameCount), 1, C.vpx_enc_frame_flags_t(flags), C.VPX_DL_REALTIME) != 0 {
fmt.Println("Failed to encode frame")
}
v.frameCount++
// Get Frame
for {
goBytes := C.get_frame_buffer(&v.vpxCodexCtx, &v.vpxCodexIter)
if goBytes.bs == nil {
break
}
bs := C.GoBytes(goBytes.bs, goBytes.size)
// if buffer is full skip frame
if len(v.Output) >= cap(v.Output) {
continue
}
v.Output <- bs
}
if *config.IsMonitor {
log.Println("Encoding time: ", time.Now().Sub(beginEncoding))
}
}
}
// Release release memory and stop loop
func (v *VpxEncoder) Release() {
v.started = false
if v.started {
close(v.Input)
close(v.Output)
if v.IsRunning {
log.Println("Releasing encoder")
log.Println("Close output", v.Output)
C.vpx_img_free(&v.vpxImage)
C.vpx_codec_destroy(&v.vpxCodexCtx)
// TODO: Bug here, after close it will signal
close(v.Output)
if v.Input != nil {
close(v.Input)
}
}
v.IsRunning = false
// TODO: Can we merge IsRunning and Done together
}

43
webrtc/util.go Normal file
View file

@ -0,0 +1,43 @@
// credit to https://github.com/poi5305/go-yuv2webRTC/blob/master/webrtc/webrtc.go
package webrtc
import (
"bytes"
"compress/gzip"
"io/ioutil"
)
func zip(in []byte) []byte {
var b bytes.Buffer
gz := gzip.NewWriter(&b)
_, err := gz.Write(in)
if err != nil {
panic(err)
}
err = gz.Flush()
if err != nil {
panic(err)
}
err = gz.Close()
if err != nil {
panic(err)
}
return b.Bytes()
}
func unzip(in []byte) []byte {
var b bytes.Buffer
_, err := b.Write(in)
if err != nil {
panic(err)
}
r, err := gzip.NewReader(&b)
if err != nil {
panic(err)
}
res, err := ioutil.ReadAll(r)
if err != nil {
panic(err)
}
return res
}

108
webrtc/webrtc.go
View file

@ -2,12 +2,9 @@
package webrtc
import (
"bytes"
"compress/gzip"
"encoding/base64"
"encoding/json"
"fmt"
"io/ioutil"
"log"
"math/rand"
"time"
@ -24,41 +21,6 @@ var webrtcconfig = webrtc.Configuration{ICEServers: []webrtc.ICEServer{{URLs: []
// Allows compressing offer/answer to bypass terminal input limits.
const compress = false
func zip(in []byte) []byte {
var b bytes.Buffer
gz := gzip.NewWriter(&b)
_, err := gz.Write(in)
if err != nil {
panic(err)
}
err = gz.Flush()
if err != nil {
panic(err)
}
err = gz.Close()
if err != nil {
panic(err)
}
return b.Bytes()
}
func unzip(in []byte) []byte {
var b bytes.Buffer
_, err := b.Write(in)
if err != nil {
panic(err)
}
r, err := gzip.NewReader(&b)
if err != nil {
panic(err)
}
res, err := ioutil.ReadAll(r)
if err != nil {
panic(err)
}
return res
}
// Encode encodes the input in base64
// It can optionally zip the input before encoding
func Encode(obj interface{}) string {
@ -122,7 +84,7 @@ type WebRTC struct {
AudioChannel chan []byte
InputChannel chan int
Done chan struct{}
Done bool
lastTime time.Time
curFPS int
@ -203,9 +165,10 @@ func (w *WebRTC) StartClient(remoteSession string, width, height int) (string, e
})
inputTrack.OnClose(func() {
fmt.Println("closed webrtc")
w.Done <- struct{}{}
close(w.Done)
log.Println("Data channel closed")
log.Println("Closed webrtc")
//close(w.Done)
//w.StopClient()
})
// WebRTC state callback
@ -261,16 +224,27 @@ func (w *WebRTC) AddCandidate(candidate webrtc.ICECandidateInit) {
// StopClient disconnect
func (w *WebRTC) StopClient() {
// if stopped, bypass
if w.isConnected == false {
return
}
log.Println("===StopClient===")
w.isConnected = false
if w.encoder != nil {
w.encoder.Release()
// NOTE: We signal using bool value instead of channel for better performance
w.encoder.Done = true
}
if w.connection != nil {
w.connection.Close()
}
w.connection = nil
//w.isClosed = true
close(w.InputChannel)
// webrtc is producer, so we close
close(w.encoder.Input)
// NOTE: ImageChannel is waiting for input. Close in writer is not correct for this
close(w.ImageChannel)
close(w.AudioChannel)
}
// IsConnected comment
@ -278,17 +252,23 @@ func (w *WebRTC) IsConnected() bool {
return w.isConnected
}
func (w *WebRTC) IsClosed() bool {
return w.isClosed
}
// func (w *WebRTC) startStreaming(vp8Track *webrtc.Track, opusTrack *webrtc.Track) {
func (w *WebRTC) startStreaming(vp8Track *webrtc.Track, audioTrack *webrtc.DataChannel) {
log.Println("Start streaming")
// send screenshot
go func() {
defer func() {
if r := recover(); r != nil {
fmt.Println("Recovered when sent to close Image Channel")
}
}()
for w.isConnected {
yuv := <-w.ImageChannel
yuv, ok := <-w.ImageChannel
if !ok {
log.Println("Screenshot from emulator closed")
return
}
if len(w.encoder.Input) < cap(w.encoder.Input) {
w.encoder.Input <- yuv
}
@ -297,10 +277,21 @@ func (w *WebRTC) startStreaming(vp8Track *webrtc.Track, audioTrack *webrtc.DataC
// receive frame buffer
go func() {
defer func() {
if r := recover(); r != nil {
fmt.Println("Recovered when sent to closed encoder output channel")
}
}()
for w.isConnected {
bs := <-w.encoder.Output
bs, ok := <-w.encoder.Output
if !ok {
log.Println("WebRTC Video sending Closed")
return
}
if *config.IsMonitor {
log.Println("FPS : ", w.calculateFPS())
log.Println("Encoding FPS : ", w.calculateFPS())
}
vp8Track.WriteSample(media.Sample{Data: bs, Samples: 1})
}
@ -308,10 +299,19 @@ func (w *WebRTC) startStreaming(vp8Track *webrtc.Track, audioTrack *webrtc.DataC
// send audio
go func() {
defer func() {
if r := recover(); r != nil {
fmt.Println("Recovered when sent to closed Audio Channel")
}
}()
for w.isConnected {
data := <-w.AudioChannel
// time.Sleep()
// time.Sleep(time.Millisecond * time.Duration(rand.Intn(100)))
data, ok := <-w.AudioChannel
if !ok {
log.Println("WebRTC Audio sending Closed")
return
}
audioTrack.Send(data)
}
}()