diff --git a/cmd/main.go b/cmd/main.go index dd3e0dd6..24c10aab 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -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. diff --git a/cmd/main_test.go b/cmd/main_test.go index 9cdc1f2e..e8deb004 100644 --- a/cmd/main_test.go +++ b/cmd/main_test.go @@ -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") + +} diff --git a/cws/cws.go b/cws/cws.go index 2a02da42..42a484e3 100644 --- a/cws/cws.go +++ b/cws/cws.go @@ -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) diff --git a/emulator/director.go b/emulator/director.go index 770e5227..6eba51a5 100644 --- a/emulator/director.go +++ b/emulator/director.go @@ -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) } diff --git a/emulator/gameview.go b/emulator/gameview.go index 28051fa8..5a4c270f 100644 --- a/emulator/gameview.go +++ b/emulator/gameview.go @@ -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 } } diff --git a/emulator/util.go b/emulator/util.go index 79cbec16..7f5a2d78 100644 --- a/emulator/util.go +++ b/emulator/util.go @@ -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) diff --git a/games/Mega Man.nes b/games/Mega Man.nes deleted file mode 100644 index 0669d5f4..00000000 Binary files a/games/Mega Man.nes and /dev/null differ diff --git a/handler/browser.go b/handler/browser.go index d591aefd..6c174f5f 100644 --- a/handler/browser.go +++ b/handler/browser.go @@ -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{ diff --git a/handler/handlers.go b/handler/handlers.go index 0e70cbae..958b4438 100644 --- a/handler/handlers.go +++ b/handler/handlers.go @@ -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 { diff --git a/handler/overlord.go b/handler/overlord.go index 2e1ef59e..3243ec5f 100644 --- a/handler/overlord.go +++ b/handler/overlord.go @@ -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 diff --git a/handler/room/media.go b/handler/room/media.go index 3a3581ab..4daed757 100644 --- a/handler/room/media.go +++ b/handler/room/media.go @@ -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() } } diff --git a/handler/room/room.go b/handler/room/room.go index 2797a755..a5b6f5f0 100644 --- a/handler/room/room.go +++ b/handler/room/room.go @@ -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 } } diff --git a/handler/session.go b/handler/session.go index c344693d..3e03ed3c 100644 --- a/handler/session.go +++ b/handler/session.go @@ -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() +} diff --git a/overlord/overlord.go b/overlord/overlord.go index c73527a7..4b7cfb5e 100644 --- a/overlord/overlord.go +++ b/overlord/overlord.go @@ -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() +} diff --git a/static/js/ws.js b/static/js/ws.js index 34de43ca..b18539a1 100644 --- a/static/js/ws.js +++ b/static/js/ws.js @@ -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(); } diff --git a/vpx-encoder/encoder.go b/vpx-encoder/encoder.go index 5392559c..1c66f22a 100644 --- a/vpx-encoder/encoder.go +++ b/vpx-encoder/encoder.go @@ -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 } diff --git a/webrtc/util.go b/webrtc/util.go new file mode 100644 index 00000000..5d428176 --- /dev/null +++ b/webrtc/util.go @@ -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 +} diff --git a/webrtc/webrtc.go b/webrtc/webrtc.go index a614ec55..f1b0b1cd 100644 --- a/webrtc/webrtc.go +++ b/webrtc/webrtc.go @@ -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) } }()