diff --git a/cmd/main.go b/cmd/main.go new file mode 100644 index 00000000..c1974ff4 --- /dev/null +++ b/cmd/main.go @@ -0,0 +1,87 @@ +package main + +import ( + "flag" + "log" + "math/rand" + "net/http" + _ "net/http/pprof" + "strings" + "time" + + "github.com/giongto35/cloud-game/config" + "github.com/giongto35/cloud-game/handler" + "github.com/giongto35/cloud-game/overlord" + "github.com/gorilla/websocket" +) + +const ( + gameboyIndex = "./static/gameboy.html" + debugIndex = "./static/index_ws.html" + gamePath = "games" +) + +// Time allowed to write a message to the peer. +var upgrader = websocket.Upgrader{} + +func createOverlordConnection() (*websocket.Conn, error) { + c, _, err := websocket.DefaultDialer.Dial(*config.OverlordHost, nil) + if err != nil { + return nil, err + } + + return c, nil +} + +// initilizeOverlord setup an overlord server +func initilizeOverlord() { + overlord := overlord.NewServer() + + log.Println("http://localhost:9000") + + // Can consider Overlord works as server but it is complicated + http.HandleFunc("/wso", overlord.WSO) + http.ListenAndServe(":9000", nil) +} + +// initializeServer setup a server +func initializeServer() { + conn, err := createOverlordConnection() + if err != nil { + log.Println("Cannot connect to overlord") + log.Println("Run as a single server") + } + + handler := handler.NewHandler(conn, *config.IsDebug, gamePath) + + // ignore origin + upgrader.CheckOrigin = func(r *http.Request) bool { return true } + + http.HandleFunc("/", handler.GetWeb) + http.Handle("/static/", http.StripPrefix("/static/", http.FileServer(http.Dir("./static")))) + http.HandleFunc("/ws", handler.WS) + + log.Println("http://localhost:" + *config.Port) + http.ListenAndServe(":"+*config.Port, nil) +} + +func main() { + flag.Parse() + log.Println("Usage: ./game [-debug]") + + rand.Seed(time.Now().UTC().UnixNano()) + + // 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. + if *config.OverlordHost == "overlord" { + log.Println("Running as overlord ") + initilizeOverlord() + } else { + if strings.HasPrefix(*config.OverlordHost, "ws") && !strings.HasSuffix(*config.OverlordHost, "wso") { + log.Fatal("Overlord connection is invalid. Should have the form `ws://.../wso`") + } + log.Println("Running as slave ") + initializeServer() + } +} diff --git a/cmd/main_test.go b/cmd/main_test.go new file mode 100644 index 00000000..9cdc1f2e --- /dev/null +++ b/cmd/main_test.go @@ -0,0 +1,389 @@ +package main + +import ( + "fmt" + "log" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/giongto35/cloud-game/cws" + "github.com/giongto35/cloud-game/handler" + "github.com/giongto35/cloud-game/overlord" + gamertc "github.com/giongto35/cloud-game/webrtc" + "github.com/gorilla/websocket" + "github.com/pion/webrtc" +) + +var host = "http://localhost:8000" + +// Test is in cmd, so gamePath is in parent path +var testGamePath = "../games" +var webrtcconfig = webrtc.Configuration{ICEServers: []webrtc.ICEServer{{URLs: []string{"stun:stun.l.google.com:19302"}}}} + +func initOverlord() *httptest.Server { + server := overlord.NewServer() + overlord := httptest.NewServer(http.HandlerFunc(server.WSO)) + return overlord +} + +func initServer(t *testing.T, oconn *websocket.Conn) *httptest.Server { + fmt.Println("Spawn new server") + handler := handler.NewHandler(oconn, true, testGamePath) + server := httptest.NewServer(http.HandlerFunc(handler.WS)) + return server +} + +func connectTestOverlordServer(t *testing.T, overlordURL string) *websocket.Conn { + if overlordURL == "" { + return nil + } else { + overlordURL = "ws" + strings.TrimPrefix(overlordURL, "http") + fmt.Println("connecting to overlord: ", overlordURL) + } + + oconn, _, err := websocket.DefaultDialer.Dial(overlordURL, nil) + if err != nil { + t.Fatalf("%v", err) + } + + return oconn +} + +func initClient(t *testing.T, host string) (client *cws.Client) { + // Convert http://127.0.0.1 to ws://127.0.0. + u := "ws" + strings.TrimPrefix(host, "http") + + // Connect to the server + fmt.Println("Connecting to server") + ws, _, err := websocket.DefaultDialer.Dial(u, nil) + if err != nil { + t.Fatalf("%v", err) + } + + // Simulate peerconnection initialization from client + fmt.Println("Simulating PeerConnection") + peerConnection, err := webrtc.NewPeerConnection(webrtcconfig) + if err != nil { + t.Fatalf("%v", err) + } + + offer, err := peerConnection.CreateOffer(nil) + if err != nil { + t.Fatalf("%v", err) + } + + // Sets the LocalDescription, and starts our UDP listeners + err = peerConnection.SetLocalDescription(offer) + if err != nil { + panic(err) + } + + // Send offer to server + log.Println("Browser Client") + client = cws.NewClient(ws) + go client.Listen() + + fmt.Println("Sending offer...") + client.Send(cws.WSPacket{ + ID: "initwebrtc", + Data: gamertc.Encode(offer), + }, nil) + fmt.Println("Waiting sdp...") + + client.Receive("sdp", func(resp cws.WSPacket) cws.WSPacket { + fmt.Println("received", resp.Data) + answer := webrtc.SessionDescription{} + gamertc.Decode(resp.Data, &answer) + // Apply the answer as the remote description + err = peerConnection.SetRemoteDescription(answer) + if err != nil { + panic(err) + } + + return cws.EmptyPacket + }) + + // Request Offer routing + client.Receive("requestOffer", func(resp cws.WSPacket) cws.WSPacket { + log.Println("Frontend received requestOffer") + peerConnection, err = webrtc.NewPeerConnection(webrtcconfig) + if err != nil { + t.Fatalf("%v", err) + } + + log.Println("Recreating offer") + offer, err := peerConnection.CreateOffer(nil) + if err != nil { + t.Fatalf("%v", err) + } + + log.Println("Set localDesc") + err = peerConnection.SetLocalDescription(offer) + if err != nil { + panic(err) + } + log.Println("return offer", offer) + return cws.WSPacket{ + ID: "initwebrtc", + Data: gamertc.Encode(offer), + } + }) + + return client + // If receive roomID, the server is running correctly +} + +func TestSingleServerNoOverlord(t *testing.T) { + // Init slave server + s := initServer(t, nil) + defer s.Close() + + client := initClient(t, s.URL) + defer client.Close() + + 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("RoomID should not be empty") + t.Fail() + } + fmt.Println("Done") +} + +func TestSingleServerOneOverlord(t *testing.T) { + o := initOverlord() + defer o.Close() + + oconn := connectTestOverlordServer(t, o.URL) + defer oconn.Close() + // Init slave server + s := initServer(t, oconn) + defer s.Close() + + client := initClient(t, s.URL) + defer client.Close() + + 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 + }) + + respRoomID := <-roomID + if respRoomID == "" { + fmt.Println("RoomID should not be empty") + t.Fail() + } + fmt.Println("Done") +} + +func TestTwoServerOneOverlord(t *testing.T) { + o := initOverlord() + defer o.Close() + + oconn1 := connectTestOverlordServer(t, o.URL) + // Init slave server + s1 := initServer(t, oconn1) + defer s1.Close() + + oconn2 := connectTestOverlordServer(t, o.URL) + // TODO: two different oconn + s2 := initServer(t, oconn2) + defer s2.Close() + + client1 := initClient(t, s1.URL) + defer client1.Close() + + roomID := make(chan string) + client1.Send(cws.WSPacket{ + ID: "start", + Data: "Contra.nes", + RoomID: "", + PlayerIndex: 1, + }, func(resp cws.WSPacket) { + fmt.Println("RoomID:", resp.RoomID) + roomID <- resp.RoomID + }) + + remoteRoomID := <-roomID + if remoteRoomID == "" { + fmt.Println("RoomID should not be empty") + t.Fail() + } + fmt.Println("Done create a room in server 1") + + // ------------------------------------ + // Client2 trying to create a random room and later join the the room on server1 + client2 := initClient(t, s2.URL) + defer client2.Close() + // Doing the same create local room. + localRoomID := make(chan string) + client2.Send(cws.WSPacket{ + ID: "start", + Data: "Contra.nes", + RoomID: "", + PlayerIndex: 1, + }, func(resp cws.WSPacket) { + fmt.Println("RoomID:", resp.RoomID) + localRoomID <- resp.RoomID + }) + + <-localRoomID + + 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 + client2.Send(cws.WSPacket{ + ID: "start", + Data: "Contra.nes", + RoomID: remoteRoomID, + PlayerIndex: 1, + }, func(resp cws.WSPacket) { + fmt.Println("RoomID:", resp.RoomID) + roomID <- resp.RoomID + }) + + // If receive roomID, the server is running correctly + fmt.Println("Done") +} + +func TestReconnectRoomNoOverlord(t *testing.T) { + // Init slave server + s := initServer(t, nil) + + client := initClient(t, s.URL) + + 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 + }) + + saveRoomID := <-roomID + if saveRoomID == "" { + fmt.Println("RoomID should not be empty") + t.Fail() + } + + log.Println("Closing room and server") + client.Close() + s.Close() + // Close server and reconnect + + // Respawn slave server + s = initServer(t, nil) + defer s.Close() + + client = initClient(t, s.URL) + defer client.Close() + + 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() + } + + fmt.Println("Done") + +} + +// This test currently doesn't work +//func TestReconnectRoomWithOverlord(t *testing.T) { +//o := initOverlord() +//defer o.Close() + +//oconn := connectTestOverlordServer(t, o.URL) +//defer oconn.Close() +//// Init slave server +//s := initServer(t, oconn) + +//client := initClient(t, s.URL) + +//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 +//}) + +//saveRoomID := <-roomID +//if saveRoomID == "" { +//fmt.Println("RoomID should not be empty") +//t.Fail() +//} + +//log.Println("Closing room and server") +//client.Close() +//s.Close() +//// Close server and reconnect + +//log.Println("Server respawn") +//// Init slave server +//s = initServer(t, oconn) +//defer s.Close() + +//client = initClient(t, s.URL) +//defer client.Close() + +//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() +//} + +//fmt.Println("Done") + +//} diff --git a/config/config.go b/config/config.go index 3f03b0d4..99b9382d 100644 --- a/config/config.go +++ b/config/config.go @@ -8,3 +8,5 @@ var IsDebug = flag.Bool("debug", false, "Is game running in debug mode?") var OverlordHost = flag.String("overlordhost", defaultoverlord, "Specify the path for overlord. If the flag is `overlord`, the server will be run as overlord") var Port = flag.String("port", "8000", "Port of the game") var IsMonitor = flag.Bool("monitor", false, "Turn on monitor") +var Width = 256 +var Height = 240 diff --git a/ws.go b/cws/cws.go similarity index 64% rename from ws.go rename to cws/cws.go index 13e66a12..2a02da42 100644 --- a/ws.go +++ b/cws/cws.go @@ -1,4 +1,4 @@ -package main +package cws import ( "encoding/json" @@ -11,6 +11,8 @@ import ( ) type Client struct { + id string + conn *websocket.Conn sendLock sync.Mutex @@ -37,9 +39,12 @@ type WSPacket struct { var EmptyPacket = WSPacket{} func NewClient(conn *websocket.Conn) *Client { + id := uuid.Must(uuid.NewV4()).String() sendCallback := map[string]func(WSPacket){} recvCallback := map[string]func(WSPacket){} + return &Client{ + id: id, conn: conn, sendCallback: sendCallback, @@ -47,8 +52,8 @@ func NewClient(conn *websocket.Conn) *Client { } } -// send sends a packet and trigger callback when the packet comes back -func (c *Client) send(request WSPacket, callback func(response WSPacket)) { +// Send sends a packet and trigger callback when the packet comes back +func (c *Client) Send(request WSPacket, callback func(response WSPacket)) { request.PacketID = uuid.Must(uuid.NewV4()).String() data, err := json.Marshal(request) if err != nil { @@ -56,29 +61,33 @@ func (c *Client) send(request WSPacket, callback func(response WSPacket)) { } // TODO: Consider using lock free + // Wrap callback with sessionID and packetID + if callback != nil { + wrapperCallback := func(resp WSPacket) { + resp.PacketID = request.PacketID + resp.SessionID = request.SessionID + callback(resp) + } + c.sendCallbackLock.Lock() + c.sendCallback[request.PacketID] = wrapperCallback + c.sendCallbackLock.Unlock() + } + //log.Println("Registered requested callback", "ID :", request.ID, "PacketID: ", request.PacketID) + //log.Println("Callback waiting list:", c.id, c.sendCallback) + c.sendLock.Lock() c.conn.WriteMessage(websocket.TextMessage, data) c.sendLock.Unlock() - wrapperCallback := func(resp WSPacket) { - resp.PacketID = request.PacketID - resp.SessionID = request.SessionID - callback(resp) - } - if callback == nil { - return - } - c.sendCallbackLock.Lock() - c.sendCallback[request.PacketID] = wrapperCallback - c.sendCallbackLock.Unlock() } -// receive receive and response back -func (c *Client) receive(id string, f func(response WSPacket) (request WSPacket)) { +// Receive receive and response back +func (c *Client) Receive(id string, f func(response WSPacket) (request WSPacket)) { c.recvCallback[id] = func(response WSPacket) { req := f(response) // Add Meta data req.PacketID = response.PacketID req.SessionID = response.SessionID + //log.Println("Sending back request", req, "PacketID: ", req.PacketID, "SessionID: ", req.SessionID) // Skip rqeuest if it is EmptyPacket if req == EmptyPacket { @@ -95,27 +104,27 @@ func (c *Client) receive(id string, f func(response WSPacket) (request WSPacket) } } -// syncSend sends a packet and wait for callback till the packet comes back -func (c *Client) syncSend(request WSPacket) (response WSPacket) { +// SyncSend sends a packet and wait for callback till the packet comes back +func (c *Client) SyncSend(request WSPacket) (response WSPacket) { res := make(chan WSPacket) f := func(resp WSPacket) { res <- resp } - c.send(request, f) + c.Send(request, f) return <-res } -// heartbeat maintains connection to server -func (c *Client) heartbeat() { +// Heartbeat maintains connection to server +func (c *Client) Heartbeat() { // send heartbeat every 1s timer := time.Tick(time.Second) for range timer { - c.send(WSPacket{ID: "heartbeat"}, nil) + c.Send(WSPacket{ID: "heartbeat"}, nil) } } -func (c *Client) listen() { +func (c *Client) Listen() { for { //log.Println("Waiting for message ...") _, rawMsg, err := c.conn.ReadMessage() @@ -125,6 +134,8 @@ func (c *Client) listen() { } wspacket := WSPacket{} err = json.Unmarshal(rawMsg, &wspacket) + //log.Println( "Received: ", wspacket) + if err != nil { continue } @@ -132,19 +143,27 @@ func (c *Client) listen() { // 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) callback, ok := c.sendCallback[wspacket.PacketID] + //log.Println("Has callback: ", ok, "ClientID: ", c.id, "PacketID ", wspacket.PacketID) c.sendCallbackLock.Unlock() if ok { go callback(wspacket) c.sendCallbackLock.Lock() + //log.Println("Deleteing Packet ", wspacket.PacketID) delete(c.sendCallback, wspacket.PacketID) 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) } } } + +func (c *Client) Close() { + c.conn.Close() +} diff --git a/ui/director.go b/emulator/director.go similarity index 76% rename from ui/director.go rename to emulator/director.go index 349b3652..770e5227 100644 --- a/ui/director.go +++ b/emulator/director.go @@ -1,11 +1,11 @@ -package ui +package emulator import ( "image" "log" "time" - "github.com/giongto35/cloud-game/nes" + "github.com/giongto35/cloud-game/emulator/nes" // "github.com/gordonklaus/portaudio" ) @@ -19,7 +19,9 @@ type Director struct { Done chan struct{} roomID string - hash string + // Hash represents a game state (roomID, gamePath). + // It is used for save file name + hash string } const FPS = 60 @@ -77,7 +79,7 @@ func (d *Director) Run() { c := time.Tick(time.Second / FPS) L: for range c { - // for { + // for { // quit game // TODO: Anyway not using select because it will slow down select { @@ -109,20 +111,32 @@ func (d *Director) PlayGame(path string) { d.SetView(NewGameView(console, path, hash, d.imageChannel, d.audioChannel, d.inputChannel)) } -func (d *Director) SaveGame() error { +// 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) + d.view.Save(d.hash, saveExtraFunc) return nil } else { return nil } } -func (d *Director) LoadGame() 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) + d.view.Load(d.hash, loadExtraFunc) 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) +} diff --git a/ui/font.go b/emulator/font.go similarity index 99% rename from ui/font.go rename to emulator/font.go index 882b3a68..e41c7448 100644 --- a/ui/font.go +++ b/emulator/font.go @@ -1,5 +1,5 @@ // credit to https://github.com/fogleman/nes -package ui +package emulator import ( "bytes" diff --git a/ui/gameview.go b/emulator/gameview.go similarity index 76% rename from ui/gameview.go rename to emulator/gameview.go index dd6740f5..28051fa8 100644 --- a/ui/gameview.go +++ b/emulator/gameview.go @@ -1,10 +1,10 @@ // credit to https://github.com/fogleman/nes -package ui +package emulator import ( "image" - "github.com/giongto35/cloud-game/nes" + "github.com/giongto35/cloud-game/emulator/nes" ) // List key pressed @@ -32,11 +32,11 @@ const NumKeys = 8 // Audio consts const ( SampleRate = 16000 - Channels = 1 - TimeFrame = 60 + Channels = 1 + TimeFrame = 60 + AppAudio = 1 ) - type GameView struct { console *nes.Console title string @@ -45,14 +45,18 @@ type GameView struct { // equivalent to the list key pressed const above keyPressed [NumKeys * 2]bool - savingPath string - loadingPath string + savingJob *job + loadingJob *job imageChannel chan *image.RGBA audioChannel chan float32 inputChannel chan int } +type job struct { + path string + extraFunc func() error +} func NewGameView(console *nes.Console, title, hash string, imageChannel chan *image.RGBA, audioChannel chan float32, inputChannel chan int) *GameView { gameview := &GameView{ @@ -61,7 +65,7 @@ func NewGameView(console *nes.Console, title, hash string, imageChannel chan *im hash: hash, keyPressed: [NumKeys * 2]bool{false}, imageChannel: imageChannel, - audioChannel: audioChannel, + audioChannel: audioChannel, inputChannel: inputChannel, } @@ -95,7 +99,7 @@ func (view *GameView) Enter() { view.console.SetAudioSampleRate(SampleRate) view.console.SetAudioChannel(view.audioChannel) - // load state + // load state if the hash file existed (Join the old room) if err := view.console.LoadState(savePath(view.hash)); err == nil { return } else { @@ -137,26 +141,36 @@ func (view *GameView) Update(t, dt float64) { view.imageChannel <- console.Buffer() } -func (view *GameView) Save(hash string) { +func (view *GameView) Save(hash string, extraSaveFunc func() error) { // put saving event to queue, process in updateEvent - view.savingPath = savePath(view.hash) + view.savingJob = &job{ + path: savePath(view.hash), + extraFunc: extraSaveFunc, + } } -func (view *GameView) Load(path string) { +func (view *GameView) Load(path string, extraLoadFunc func() error) { // put saving event to queue, process in updateEvent - view.loadingPath = savePath(view.hash) + view.loadingJob = &job{ + path: savePath(view.hash), + extraFunc: extraLoadFunc, + } } func (view *GameView) UpdateEvents() { // If there is saving event, save and discard the save event - if view.savingPath != "" { - view.console.SaveState(view.savingPath) - view.savingPath = "" + if view.savingJob != nil { + view.console.SaveState(view.savingJob.path) + // Run extra function (online saving for example) + go view.savingJob.extraFunc() + view.savingJob = nil } // If there is loading event, save and discard the load event - if view.loadingPath != "" { - view.console.LoadState(view.loadingPath) - view.loadingPath = "" + if view.loadingJob != nil { + view.console.LoadState(view.loadingJob.path) + // Run extra function (online saving for example) + go view.loadingJob.extraFunc() + view.loadingJob = nil } } diff --git a/nes/apu.go b/emulator/nes/apu.go similarity index 100% rename from nes/apu.go rename to emulator/nes/apu.go diff --git a/nes/cartridge.go b/emulator/nes/cartridge.go similarity index 100% rename from nes/cartridge.go rename to emulator/nes/cartridge.go diff --git a/nes/console.go b/emulator/nes/console.go similarity index 100% rename from nes/console.go rename to emulator/nes/console.go diff --git a/nes/controller.go b/emulator/nes/controller.go similarity index 100% rename from nes/controller.go rename to emulator/nes/controller.go diff --git a/nes/cpu.go b/emulator/nes/cpu.go similarity index 100% rename from nes/cpu.go rename to emulator/nes/cpu.go diff --git a/nes/filter.go b/emulator/nes/filter.go similarity index 100% rename from nes/filter.go rename to emulator/nes/filter.go diff --git a/nes/ines.go b/emulator/nes/ines.go similarity index 100% rename from nes/ines.go rename to emulator/nes/ines.go diff --git a/nes/mapper.go b/emulator/nes/mapper.go similarity index 100% rename from nes/mapper.go rename to emulator/nes/mapper.go diff --git a/nes/mapper1.go b/emulator/nes/mapper1.go similarity index 100% rename from nes/mapper1.go rename to emulator/nes/mapper1.go diff --git a/nes/mapper2.go b/emulator/nes/mapper2.go similarity index 100% rename from nes/mapper2.go rename to emulator/nes/mapper2.go diff --git a/nes/mapper225.go b/emulator/nes/mapper225.go similarity index 100% rename from nes/mapper225.go rename to emulator/nes/mapper225.go diff --git a/nes/mapper3.go b/emulator/nes/mapper3.go similarity index 100% rename from nes/mapper3.go rename to emulator/nes/mapper3.go diff --git a/nes/mapper4.go b/emulator/nes/mapper4.go similarity index 100% rename from nes/mapper4.go rename to emulator/nes/mapper4.go diff --git a/nes/mapper7.go b/emulator/nes/mapper7.go similarity index 100% rename from nes/mapper7.go rename to emulator/nes/mapper7.go diff --git a/nes/memory.go b/emulator/nes/memory.go similarity index 100% rename from nes/memory.go rename to emulator/nes/memory.go diff --git a/nes/palette.go b/emulator/nes/palette.go similarity index 100% rename from nes/palette.go rename to emulator/nes/palette.go diff --git a/nes/ppu.go b/emulator/nes/ppu.go similarity index 100% rename from nes/ppu.go rename to emulator/nes/ppu.go diff --git a/ui/util.go b/emulator/util.go similarity index 97% rename from ui/util.go rename to emulator/util.go index 931ce8d8..79cbec16 100644 --- a/ui/util.go +++ b/emulator/util.go @@ -1,5 +1,5 @@ // credit to https://github.com/fogleman/nes -package ui +package emulator import ( "crypto/md5" @@ -16,7 +16,7 @@ import ( "os/user" "path" - "github.com/giongto35/cloud-game/nes" + "github.com/giongto35/cloud-game/emulator/nes" ) var homeDir string diff --git a/games.go b/games.go deleted file mode 100644 index 8c8923ad..00000000 --- a/games.go +++ /dev/null @@ -1,30 +0,0 @@ -package main - -import ( - "encoding/json" - "os" - "path/filepath" -) - -const gamePath = "games" - -// getGameList returns list of games stored in games -func getGameList() []string { - var games []string - filepath.Walk(gamePath, func(path string, info os.FileInfo, err error) error { - if !info.IsDir() { - // remove prefix - path = path[len(gamePath)+1:] - // Add to games list - games = append(games, path) - } - return nil - }) - - return games -} - -func getEncodedGameList() string { - encodedList, _ := json.Marshal(getGameList()) - return string(encodedList) -} diff --git a/handler/browser.go b/handler/browser.go new file mode 100644 index 00000000..d591aefd --- /dev/null +++ b/handler/browser.go @@ -0,0 +1,142 @@ +package handler + +import ( + "encoding/json" + "log" + + "github.com/giongto35/cloud-game/config" + "github.com/giongto35/cloud-game/cws" + "github.com/gorilla/websocket" + pionRTC "github.com/pion/webrtc" +) + +type BrowserClient struct { + *cws.Client +} + +func (s *Session) RegisterBrowserClient() { + browserClient := s.BrowserClient + + browserClient.Receive("heartbeat", func(resp cws.WSPacket) cws.WSPacket { + return resp + }) + + browserClient.Receive("initwebrtc", func(resp cws.WSPacket) cws.WSPacket { + log.Println("Received user SDP") + localSession, err := s.peerconnection.StartClient(resp.Data, config.Width, config.Height) + if err != nil { + log.Fatalln(err) + } + + return cws.WSPacket{ + ID: "sdp", + Data: localSession, + SessionID: s.ID, + } + }) + + // TODO: Add save and load + browserClient.Receive("save", func(resp cws.WSPacket) (req cws.WSPacket) { + log.Println("Saving game state") + req.ID = "save" + req.Data = "ok" + if s.RoomID != "" { + room := s.handler.getRoom(s.RoomID) + err := room.SaveGame() + if err != nil { + log.Println("[!] Cannot save game state: ", err) + req.Data = "error" + } + } else { + req.Data = "error" + } + + return req + }) + + browserClient.Receive("load", func(resp cws.WSPacket) (req cws.WSPacket) { + log.Println("Loading game state") + req.ID = "load" + req.Data = "ok" + if s.RoomID != "" { + room := s.handler.getRoom(s.RoomID) + err := room.LoadGame() + if err != nil { + log.Println("[!] Cannot load game state: ", err) + req.Data = "error" + } + } else { + req.Data = "error" + } + + return req + }) + + browserClient.Receive("start", func(resp cws.WSPacket) (req cws.WSPacket) { + s.GameName = resp.Data + s.RoomID = resp.RoomID + s.PlayerIndex = resp.PlayerIndex + + log.Println("Starting game") + // If we are connecting to overlord, request corresponding serverID based on roomID + if s.OverlordClient != nil { + roomServerID := getServerIDOfRoom(s.OverlordClient, s.RoomID) + log.Println("Server of RoomID ", s.RoomID, " is ", roomServerID, " while current server is ", s.ServerID) + // If the target serverID is different from current serverID + if roomServerID != "" && s.ServerID != roomServerID { + // TODO: Re -register + // Bridge Connection to the target serverID + go s.bridgeConnection(roomServerID, s.GameName, s.RoomID, s.PlayerIndex) + return + } + } + + // Create new room + // TODO: check if roomID is in the current server + room := s.handler.getRoom(s.RoomID) + log.Println("Got Room from local ", room, " ID: ", s.RoomID) + if room == nil { + room = s.handler.createNewRoom(s.GameName, s.RoomID, s.PlayerIndex) + } + + // Attach peerconnection to room + s.handler.detachPeerConn(s.peerconnection) + room.AddConnectionToRoom(s.peerconnection, s.PlayerIndex) + s.RoomID = room.ID + + // Register room to overlord if we are connecting to overlord + if room != nil && s.OverlordClient != nil { + s.OverlordClient.Send(cws.WSPacket{ + ID: "registerRoom", + Data: s.RoomID, + }, nil) + } + req.ID = "start" + req.RoomID = s.RoomID + req.SessionID = s.ID + + return req + }) + + browserClient.Receive("candidate", func(resp cws.WSPacket) (req cws.WSPacket) { + // Unuse code + hi := pionRTC.ICECandidateInit{} + err := json.Unmarshal([]byte(resp.Data), &hi) + if err != nil { + log.Println("[!] Cannot parse candidate: ", err) + } else { + // webRTC.AddCandidate(hi) + } + req.ID = "candidate" + + return req + }) + +} + +// NewOverlordClient returns a client connecting to browser. This connection exchanges information between clients and server +func NewBrowserClient(c *websocket.Conn) *BrowserClient { + return &BrowserClient{ + Client: cws.NewClient(c), + } +} diff --git a/handler/cloud-storage/storage.go b/handler/cloud-storage/storage.go new file mode 100644 index 00000000..3aa0d6c4 --- /dev/null +++ b/handler/cloud-storage/storage.go @@ -0,0 +1,78 @@ +package storage + +import ( + "context" + "io" + "io/ioutil" + "log" + "os" + + "cloud.google.com/go/storage" +) + +// TODO: Add interface, abstract out Gstorage +type Client struct { + bucket *storage.BucketHandle + gclient *storage.Client +} + +func NewInitClient() *Client { + projectID := os.Getenv("GCP_PROJECT") + bucketName := "game-save" + return NewClient(projectID, bucketName) +} + +// NewClient inits a new Client accessing to GCP +func NewClient(projectID string, bucketName string) *Client { + ctx := context.Background() + + // Sets your Google Cloud Platform project ID. + + // Creates a client. + gclient, err := storage.NewClient(ctx) + if err != nil { + log.Fatalf("Failed to create client: %v", err) + } + + // Creates a Bucket instance. + bucket := gclient.Bucket(bucketName) + + return &Client{ + bucket: bucket, + gclient: gclient, + } +} + +// Savefile save srcFile to GCP +func (c *Client) SaveFile(name string, srcFile string) (err error) { + reader, err := os.Open(srcFile) + if err != nil { + return err + } + + // Copy source file to GCP + wc := c.bucket.Object(name).NewWriter(context.Background()) + if _, err = io.Copy(wc, reader); err != nil { + return err + } + if err := wc.Close(); err != nil { + return err + } + + return nil +} + +// Loadfile load file from GCP +func (c *Client) LoadFile(name string) (data []byte, err error) { + rc, err := c.bucket.Object(name).NewReader(context.Background()) + if err != nil { + return nil, err + } + defer rc.Close() + + data, err = ioutil.ReadAll(rc) + if err != nil { + return nil, err + } + return data, nil +} diff --git a/handler/cloud-storage/storage_test.go b/handler/cloud-storage/storage_test.go new file mode 100644 index 00000000..16e739f1 --- /dev/null +++ b/handler/cloud-storage/storage_test.go @@ -0,0 +1,24 @@ +package storage + +import ( + "io/ioutil" + "log" + "testing" +) + +func TestSaveGame(t *testing.T) { + client := NewInitClient() + data := []byte("Test Hello") + ioutil.WriteFile("/tmp/TempFile", data, 0644) + err := client.SaveFile("Test", "/tmp/TempFile") + if err != nil { + log.Panic(err) + } + loadData, err := client.LoadFile("Test") + if err != nil { + log.Panic(err) + } + if string(data) != string(loadData) { + log.Panic("Failed") + } +} diff --git a/handler/gamelist/games.go b/handler/gamelist/games.go new file mode 100644 index 00000000..8cc714b8 --- /dev/null +++ b/handler/gamelist/games.go @@ -0,0 +1,32 @@ +package gamelist + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" +) + +// getGameList returns list of games stored in games +// TODO: change to class +func GetGameList(gamePath string) []string { + var games []string + filepath.Walk(gamePath, func(path string, info os.FileInfo, err error) error { + if info != nil && !info.IsDir() { + // Remove prefix to obtain file names + path = path[len(gamePath)+1:] + // Add to games list + games = append(games, path) + } + return nil + }) + + return games +} + +// GetEncodedGameList returns game list in encoded wspacket format +func GetEncodedGameList(gamePath string) string { + encodedList, _ := json.Marshal(GetGameList(gamePath)) + fmt.Println(encodedList) + return string(encodedList) +} diff --git a/handler/handlers.go b/handler/handlers.go new file mode 100644 index 00000000..2d444d2b --- /dev/null +++ b/handler/handlers.go @@ -0,0 +1,160 @@ +package handler + +import ( + "fmt" + "io/ioutil" + "log" + "net/http" + + "github.com/giongto35/cloud-game/cws" + storage "github.com/giongto35/cloud-game/handler/cloud-storage" + "github.com/giongto35/cloud-game/handler/gamelist" + "github.com/giongto35/cloud-game/handler/room" + "github.com/giongto35/cloud-game/webrtc" + "github.com/gorilla/websocket" + uuid "github.com/satori/go.uuid" +) + +const ( + gameboyIndex = "./static/gameboy.html" + debugIndex = "./static/index_ws.html" +) + +// Flag to determine if the server is overlord or not +var upgrader = websocket.Upgrader{} + +type Handler struct { + // Client that connects to overlord + oClient *OverlordClient + // Rooms map : RoomID -> Room + rooms map[string]*room.Room + // ID of the current server globalwise + serverID string + // isDebug determines the mode handler is running + isDebug bool + // Path to game list + gamePath string + // All webrtc peerconnections are handled by the server + // ID -> peerconnections + peerconnections map[string]*webrtc.WebRTC + // onlineStorage is client accessing to online storage (GCP) + onlineStorage *storage.Client +} + +// NewHandler returns a new server +func NewHandler(overlordConn *websocket.Conn, isDebug bool, gamePath string) *Handler { + log.Println("new OverlordClient") + + return &Handler{ + oClient: NewOverlordClient(overlordConn), + rooms: map[string]*room.Room{}, + peerconnections: map[string]*webrtc.WebRTC{}, + + isDebug: isDebug, + gamePath: gamePath, + + onlineStorage: storage.NewInitClient(), + } +} + +// GetWeb returns web frontend +func (h *Handler) GetWeb(w http.ResponseWriter, r *http.Request) { + indexFN := "" + if h.isDebug { + indexFN = debugIndex + } else { + indexFN = gameboyIndex + } + + bs, err := ioutil.ReadFile(indexFN) + if err != nil { + log.Fatal(err) + } + w.Write(bs) +} + +// 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() + + if err != nil { + log.Print("[!] WS upgrade:", err) + return + } + + client := NewBrowserClient(c) + //client := NewClient(c) + ////sessionID := strconv.Itoa(rand.Int()) + sessionID := uuid.Must(uuid.NewV4()).String() + wssession := &Session{ + ID: sessionID, + BrowserClient: client, + OverlordClient: h.oClient, + peerconnection: webrtc.NewWebRTC(), + handler: h, + } + if wssession.OverlordClient != nil { + wssession.RegisterOverlordClient() + go wssession.OverlordClient.Heartbeat() + go wssession.OverlordClient.Listen() + } + + wssession.RegisterBrowserClient() + fmt.Println("oclient : ", h.oClient) + + wssession.BrowserClient.Send(cws.WSPacket{ + ID: "gamelist", + Data: gamelist.GetEncodedGameList(h.gamePath), + }, nil) + + wssession.BrowserClient.Listen() +} + +// Detach peerconnection detach/remove a peerconnection from current room +func (h *Handler) detachPeerConn(pc *webrtc.WebRTC) { + roomID := pc.RoomID + room := h.getRoom(roomID) + if room == nil { + return + } + room.CleanSession(pc) +} + +// getRoom returns room from roomID +func (h *Handler) getRoom(roomID string) *room.Room { + room, ok := h.rooms[roomID] + if !ok { + return nil + } + + return room +} + +// createNewRoom creates a new room +// Return nil in case of room is existed +func (h *Handler) createNewRoom(gameName string, roomID string, playerIndex int) *room.Room { + // If the roomID is empty, + // or the roomID doesn't have any running sessions (room was closed) + // we spawn a new room + if roomID == "" || !h.isRoomRunning(roomID) { + room := room.NewRoom(roomID, h.gamePath, gameName, h.onlineStorage) + // TODO: Might have race condition + h.rooms[room.ID] = room + return room + } + + return nil +} + +// isRoomRunning check if there is any running sessions. +// TODO: If we remove sessions from room anytime a session is closed, we can check if the sessions list is empty or not. +func (h *Handler) isRoomRunning(roomID string) bool { + // If no roomID is registered + room, ok := h.rooms[roomID] + if !ok { + return false + } + + return room.IsRunning() +} diff --git a/handler/overlord.go b/handler/overlord.go new file mode 100644 index 00000000..2e1ef59e --- /dev/null +++ b/handler/overlord.go @@ -0,0 +1,146 @@ +package handler + +import ( + "log" + + "github.com/giongto35/cloud-game/config" + "github.com/giongto35/cloud-game/cws" + "github.com/giongto35/cloud-game/webrtc" + "github.com/gorilla/websocket" +) + +// OverlordClient maintans connection to overlord +// We expect only one OverlordClient for each server +type OverlordClient struct { + *cws.Client + peerconnections map[string]*webrtc.WebRTC +} + +// NewOverlordClient returns a client connecting to overlord for coordiation between different server +func NewOverlordClient(oc *websocket.Conn) *OverlordClient { + if oc == nil { + return nil + } + + oclient := &OverlordClient{ + Client: cws.NewClient(oc), + peerconnections: map[string]*webrtc.WebRTC{}, + } + return oclient +} + +// RegisterOverlordClient routes overlord Client +func (s *Session) RegisterOverlordClient() { + oclient := s.OverlordClient + + // Received from overlord the serverID + oclient.Receive( + "serverID", + func(response cws.WSPacket) (request cws.WSPacket) { + // Stick session with serverID got from overlord + log.Println("Received serverID ", response.Data) + s.ServerID = response.Data + + return cws.EmptyPacket + }, + ) + + // Received from overlord the sdp. This is happens when bridging + // TODO: refactor + oclient.Receive( + "initwebrtc", + func(resp cws.WSPacket) (req cws.WSPacket) { + log.Println("Received a sdp request from overlord") + log.Println("Start peerconnection from the sdp") + peerconnection := webrtc.NewWebRTC() + // init new peerconnection from sessionID + localSession, err := peerconnection.StartClient(resp.Data, config.Width, config.Height) + oclient.peerconnections[resp.SessionID] = peerconnection + + if err != nil { + log.Fatalln(err) + } + + return cws.WSPacket{ + ID: "sdp", + Data: localSession, + } + }, + ) + + // Received start from overlord. This happens when bridging + // TODO: refactor + oclient.Receive( + "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") + + peerconnection := oclient.peerconnections[resp.SessionID] + log.Println("start session") + + //room := s.handler.createNewRoom(s.GameName, s.RoomID, s.PlayerIndex) + // Request room from Server if roomID is existed on the server + room := s.handler.getRoom(s.RoomID) + if room == nil { + log.Println("Room not found ", s.RoomID) + return cws.EmptyPacket + } + s.handler.detachPeerConn(s.peerconnection) + room.AddConnectionToRoom(peerconnection, s.PlayerIndex) + //roomID, isNewRoom := startSession(peerconnection, resp.Data, resp.RoomID, resp.PlayerIndex) + log.Println("Done, sending back") + + req.ID = "start" + req.RoomID = room.ID + return req + }, + ) + // heartbeat to keep pinging overlord. We not ping from server to browser, so we don't call heartbeat in browserClient +} + +func getServerIDOfRoom(oc *OverlordClient, roomID string) string { + log.Println("Request overlord roomID ", roomID) + packet := oc.SyncSend( + cws.WSPacket{ + ID: "getRoom", + Data: roomID, + }, + ) + log.Println("Received roomID from overlord ", packet.Data) + + return packet.Data +} + +func (s *Session) bridgeConnection(serverID string, gameName string, roomID string, playerIndex int) { + log.Println("Bridging connection to other Host ", serverID) + client := s.BrowserClient + // Ask client to init + + log.Println("Requesting offer to browser", serverID) + resp := client.SyncSend(cws.WSPacket{ + ID: "requestOffer", + Data: "", + }) + + // 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) + remoteTargetSDP := s.OverlordClient.SyncSend(resp) + log.Println("Got back remote host SDP, sending to browser") + // Send back remote SDP of remote server to browser + s.BrowserClient.Send(cws.WSPacket{ + ID: "sdp", + Data: remoteTargetSDP.Data, + }, nil) + log.Println("Init session done, start game on target host") + + s.OverlordClient.SyncSend(cws.WSPacket{ + ID: "start", + Data: gameName, + TargetHostID: serverID, + RoomID: roomID, + PlayerIndex: playerIndex, + }) + log.Println("Game is started on remote host") +} diff --git a/handler/room/media.go b/handler/room/media.go new file mode 100644 index 00000000..3a3581ab --- /dev/null +++ b/handler/room/media.go @@ -0,0 +1,100 @@ +package room + +import ( + "log" + + "github.com/giongto35/cloud-game/emulator" + "github.com/giongto35/cloud-game/util" + "gopkg.in/hraban/opus.v2" +) + +func (r *Room) startAudio() { + log.Println("Enter fan audio") + + enc, err := opus.NewEncoder(emulator.SampleRate, emulator.Channels, opus.AppAudio) + + maxBufferSize := emulator.TimeFrame * emulator.SampleRate / 1000 + pcm := make([]float32, maxBufferSize) // 640 * 1000 / 16000 == 40 ms + idx := 0 + + if err != nil { + log.Println("[!] Cannot create audio encoder") + return + } + + var count byte = 0 + + // fanout Audio + for { + select { + case <-r.Done: + r.Close() + return + case sample := <-r.audioChannel: + pcm[idx] = sample + idx++ + if idx == len(pcm) { + data := make([]byte, 640) + + n, err := enc.EncodeFloat32(pcm, data) + + 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 + + } + } + } +} + +func (r *Room) startVideo() { + // fanout Screen + for { + select { + case <-r.Done: + r.Close() + 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() + } + } +} diff --git a/handler/room/room.go b/handler/room/room.go new file mode 100644 index 00000000..2797a755 --- /dev/null +++ b/handler/room/room.go @@ -0,0 +1,195 @@ +package room + +import ( + "fmt" + "image" + "io/ioutil" + "log" + "math/rand" + "strconv" + "sync" + + emulator "github.com/giongto35/cloud-game/emulator" + storage "github.com/giongto35/cloud-game/handler/cloud-storage" + "github.com/giongto35/cloud-game/webrtc" +) + +// Room is a game session. multi webRTC sessions can connect to a same game. +// A room stores all the channel for interaction between all webRTCs session and emulator +type Room struct { + ID string + + imageChannel chan *image.RGBA + audioChannel chan float32 + inputChannel chan int + // Done channel is to fire exit event when there is no webRTC session running + Done chan struct{} + + rtcSessions []*webrtc.WebRTC + sessionsLock *sync.Mutex + + director *emulator.Director + + // Cloud storage to store room state online + onlineStorage *storage.Client +} + +// NewRoom creates a new room +func NewRoom(roomID, gamepath, gameName string, onlineStorage *storage.Client) *Room { + // if no roomID is given, generate it + if roomID == "" { + roomID = generateRoomID() + } + log.Println("Init new room", roomID, gameName) + imageChannel := make(chan *image.RGBA, 100) + audioChannel := make(chan float32, emulator.SampleRate) + inputChannel := make(chan int, 100) + + // create director + director := emulator.NewDirector(roomID, imageChannel, audioChannel, inputChannel) + + room := &Room{ + ID: roomID, + + imageChannel: imageChannel, + audioChannel: audioChannel, + inputChannel: inputChannel, + rtcSessions: []*webrtc.WebRTC{}, + sessionsLock: &sync.Mutex{}, + director: director, + Done: make(chan struct{}), + onlineStorage: onlineStorage, + } + + go room.startVideo() + go room.startAudio() + go director.Start([]string{gamepath + "/" + gameName}) + + return room +} + +// generateRoomID generate a unique room ID containing 16 digits +func generateRoomID() string { + roomID := strconv.FormatInt(rand.Int63(), 16) + log.Println("Generate Room ID", roomID) + //roomID := uuid.Must(uuid.NewV4()).String() + return roomID +} + +func (r *Room) AddConnectionToRoom(peerconnection *webrtc.WebRTC, playerIndex int) { + peerconnection.AttachRoomID(r.ID) + r.rtcSessions = append(r.rtcSessions, peerconnection) + + 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 + } + + // 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 + } + } +} + +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) + for i, s := range r.rtcSessions { + fmt.Println("found session: ", s, w) + if s.ID == w.ID { + r.rtcSessions = append(r.rtcSessions[:i], r.rtcSessions[i+1:]...) + fmt.Println("found session: ", len(r.rtcSessions)) + + // If room has no sessions, close room + if len(r.rtcSessions) == 0 { + log.Println("No session in room") + r.Done <- struct{}{} + } + break + } + } +} + +func (r *Room) Close() { + log.Println("Closing room", r) + r.director.Done <- struct{}{} +} + +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 { + return err + } + + return nil + } + + // TODO: Move to game view + if err := r.director.SaveGame(onlineSaveFunc); err != nil { + return err + } + + 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 + } + + err := r.director.LoadGame(onlineLoadFunc) + + return err +} + +func (r *Room) IsRunning() bool { + // If there is running session + for _, s := range r.rtcSessions { + if !s.IsClosed() { + return true + } + } + + return false +} diff --git a/handler/session.go b/handler/session.go new file mode 100644 index 00000000..c344693d --- /dev/null +++ b/handler/session.go @@ -0,0 +1,23 @@ +package handler + +import ( + "github.com/giongto35/cloud-game/webrtc" +) + +// Session represents a session connected from the browser to the current server +// It involves one connection to browser and one connection to the overlord +// Peerconnection can be from other server to ensure better latency +type Session struct { + ID string + BrowserClient *BrowserClient + OverlordClient *OverlordClient + peerconnection *webrtc.WebRTC + + // TODO: Decouple this + handler *Handler + + ServerID string + GameName string + RoomID string + PlayerIndex int +} diff --git a/main.go b/main.go deleted file mode 100644 index d622e4fc..00000000 --- a/main.go +++ /dev/null @@ -1,617 +0,0 @@ -package main - -import ( - "encoding/json" - "flag" - "image" - "io/ioutil" - "log" - "math/rand" - "net/http" - _ "net/http/pprof" - "strconv" - "strings" - "sync" - "time" - - "github.com/giongto35/cloud-game/config" - "github.com/giongto35/cloud-game/ui" - "github.com/giongto35/cloud-game/util" - "github.com/giongto35/cloud-game/webrtc" - "github.com/gorilla/websocket" - pionRTC "github.com/pion/webrtc" - uuid "github.com/satori/go.uuid" - - "gopkg.in/hraban/opus.v2" -) - -const ( - width = 256 - height = 240 - scale = 3 - title = "NES" - gameboyIndex = "./static/gameboy.html" - debugIndex = "./static/index_ws.html" -) - -var indexFN = gameboyIndex - -// Time allowed to write a message to the peer. -var readWait = 30 * time.Second -var writeWait = 30 * time.Second - -var IsOverlord = false -var upgrader = websocket.Upgrader{} - -// Room is a game session. multi webRTC sessions can connect to a same game. -// A room stores all the channel for interaction between all webRTCs session and emulator -type Room struct { - imageChannel chan *image.RGBA - audioChannel chan float32 - inputChannel chan int - // Done channel is to fire exit event when there is no webRTC session running - Done chan struct{} - - rtcSessions []*webrtc.WebRTC - sessionsLock *sync.Mutex - - director *ui.Director -} - -var rooms = map[string]*Room{} - -// ID to peerconnection -var peerconnections = map[string]*webrtc.WebRTC{} -var serverID = "" -var oclient *Client - -func main() { - flag.Parse() - log.Println("Usage: ./game [debug]") - if *config.IsDebug { - // debug - indexFN = debugIndex - log.Println("Use debug version") - } - - if *config.OverlordHost == "overlord" { - log.Println("Running as overlord ") - IsOverlord = true - } else { - if strings.HasPrefix(*config.OverlordHost, "ws") && !strings.HasSuffix(*config.OverlordHost, "wso") { - log.Fatal("Overlord connection is invalid. Should have the form `ws://.../wso`") - } - log.Println("Running as slave ") - IsOverlord = false - } - - rand.Seed(time.Now().UTC().UnixNano()) - rooms = map[string]*Room{} - - // ignore origin - upgrader.CheckOrigin = func(r *http.Request) bool { return true } - - http.HandleFunc("/", getWeb) - http.Handle("/static/", http.StripPrefix("/static/", http.FileServer(http.Dir("./static")))) - http.HandleFunc("/ws", ws) - - if !IsOverlord { - conn, err := createOverlordConnection() - if err != nil { - log.Println("Cannot connect to overlord") - log.Println("Run as a single server") - oclient = nil - } else { - oclient = NewOverlordClient(conn) - } - } - - log.Println("oclient ", oclient) - if !IsOverlord { - log.Println("http://localhost:" + *config.Port) - http.ListenAndServe(":"+*config.Port, nil) - } else { - log.Println("http://localhost:9000") - // Overlord expose one more path for handle overlord connections - http.HandleFunc("/wso", wso) - http.ListenAndServe(":9000", nil) - } -} - -func getWeb(w http.ResponseWriter, r *http.Request) { - bs, err := ioutil.ReadFile(indexFN) - if err != nil { - log.Fatal(err) - } - w.Write(bs) -} - -// init initilizes a room returns roomID -func initRoom(roomID, gameName string) string { - // if no roomID is given, generate it - if roomID == "" { - roomID = generateRoomID() - } - log.Println("Init new room", roomID, gameName) - imageChannel := make(chan *image.RGBA, 100) - audioChannel := make(chan float32, ui.SampleRate) - inputChannel := make(chan int, 100) - - // create director - director := ui.NewDirector(roomID, imageChannel, audioChannel, inputChannel) - - room := &Room{ - imageChannel: imageChannel, - audioChannel: audioChannel, - inputChannel: inputChannel, - rtcSessions: []*webrtc.WebRTC{}, - sessionsLock: &sync.Mutex{}, - director: director, - Done: make(chan struct{}), - } - rooms[roomID] = room - - go room.startVideo() - go room.startAudio() - go director.Start([]string{"games/" + gameName}) - - return roomID -} - -// isRoomRunning check if there is any running sessions. -// TODO: If we remove sessions from room anytime a session is closed, we can check if the sessions list is empty or not. -func isRoomRunning(roomID string) bool { - // If no roomID is registered - if _, ok := rooms[roomID]; !ok { - return false - } - - // If there is running session - for _, s := range rooms[roomID].rtcSessions { - if !s.IsClosed() { - return true - } - } - return false -} - -// startSession handles one session call -func startSession(webRTC *webrtc.WebRTC, gameName string, roomID string, playerIndex int) (rRoomID string, isNewRoom bool) { - isNewRoom = false - cleanSession(webRTC) - // If the roomID is empty, - // or the roomID doesn't have any running sessions (room was closed) - // we spawn a new room - if roomID == "" || !isRoomRunning(roomID) { - roomID = initRoom(roomID, gameName) - isNewRoom = true - } - - // TODO: Might have race condition - rooms[roomID].rtcSessions = append(rooms[roomID].rtcSessions, webRTC) - room := rooms[roomID] - - webRTC.AttachRoomID(roomID) - go startWebRTCSession(room, webRTC, playerIndex) - - return roomID, isNewRoom -} - -// Session represents a session connected from the browser to the current server -type Session struct { - client *Client - peerconnection *webrtc.WebRTC - ServerID string -} - -// Handle normal traffic (from browser to host) -func ws(w http.ResponseWriter, r *http.Request) { - c, err := upgrader.Upgrade(w, r, nil) - if err != nil { - log.Print("[!] WS upgrade:", err) - return - } - defer c.Close() - var gameName string - var roomID string - var playerIndex int - - // Create connection to overlord - client := NewClient(c) - //sessionID := strconv.Itoa(rand.Int()) - sessionID := uuid.Must(uuid.NewV4()).String() - - wssession := &Session{ - client: client, - peerconnection: webrtc.NewWebRTC(), - // The server session is maintaining - } - - client.send(WSPacket{ - ID: "gamelist", - Data: getEncodedGameList(), - }, nil) - - client.receive("heartbeat", func(resp WSPacket) WSPacket { - return resp - }) - - client.receive("initwebrtc", func(resp WSPacket) WSPacket { - log.Println("Received user SDP") - localSession, err := wssession.peerconnection.StartClient(resp.Data, width, height) - if err != nil { - log.Fatalln(err) - } - - return WSPacket{ - ID: "sdp", - Data: localSession, - SessionID: sessionID, - } - }) - - client.receive("save", func(resp WSPacket) (req WSPacket) { - log.Println("Saving game state") - req.ID = "save" - req.Data = "ok" - if roomID != "" { - err = rooms[roomID].director.SaveGame() - if err != nil { - log.Println("[!] Cannot save game state: ", err) - req.Data = "error" - } - } else { - req.Data = "error" - } - - return req - }) - - client.receive("load", func(resp WSPacket) (req WSPacket) { - log.Println("Loading game state") - req.ID = "load" - req.Data = "ok" - if roomID != "" { - err = rooms[roomID].director.LoadGame() - if err != nil { - log.Println("[!] Cannot load game state: ", err) - req.Data = "error" - } - } else { - req.Data = "error" - } - - return req - }) - - client.receive("start", func(resp WSPacket) (req WSPacket) { - gameName = resp.Data - roomID = resp.RoomID - playerIndex = resp.PlayerIndex - isNewRoom := false - - log.Println("Starting game") - // If we are connecting to overlord, request serverID from roomID - if oclient != nil { - roomServerID := getServerIDOfRoom(oclient, roomID) - log.Println("Server of RoomID ", roomID, " is ", roomServerID) - if roomServerID != "" && wssession.ServerID != roomServerID { - // TODO: Re -register - go bridgeConnection(wssession, roomServerID, gameName, roomID, playerIndex) - return - } - } - - roomID, isNewRoom = startSession(wssession.peerconnection, gameName, roomID, playerIndex) - // Register room to overlord if we are connecting to overlord - if isNewRoom && oclient != nil { - oclient.send(WSPacket{ - ID: "registerRoom", - Data: roomID, - }, nil) - } - req.ID = "start" - req.RoomID = roomID - req.SessionID = sessionID - - return req - }) - - client.receive("candidate", func(resp WSPacket) (req WSPacket) { - // Unuse code - hi := pionRTC.ICECandidateInit{} - err = json.Unmarshal([]byte(resp.Data), &hi) - if err != nil { - log.Println("[!] Cannot parse candidate: ", err) - } else { - // webRTC.AddCandidate(hi) - } - req.ID = "candidate" - - return req - }) - - client.listen() -} - -// generateRoomID generate a unique room ID containing 16 digits -func generateRoomID() string { - roomID := strconv.FormatInt(rand.Int63(), 16) - //roomID := uuid.Must(uuid.NewV4()).String() - return roomID -} - -func (r *Room) startVideo() { - // fanout Screen - for { - select { - case <-r.Done: - r.remove() - return - case image := <-r.imageChannel: - //isRoomRunning := false - - 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 - } - //isRoomRunning = true - } - r.sessionsLock.Unlock() - } - } -} - -func (r *Room) startAudio() { - log.Println("Enter fan audio") - - enc, err := opus.NewEncoder(ui.SampleRate, ui.Channels, opus.AppAudio) - - maxBufferSize := ui.TimeFrame * ui.SampleRate / 1000 - pcm := make([]float32, maxBufferSize) // 640 * 1000 / 16000 == 40 ms - idx := 0 - - if err != nil { - log.Println("[!] Cannot create audio encoder") - return - } - - var count byte = 0 - - // fanout Audio - for { - select { - case <-r.Done: - r.remove() - return - case sample := <-r.audioChannel: - pcm[idx] = sample - idx++ - if idx == len(pcm) { - data := make([]byte, 640) - - n, err := enc.EncodeFloat32(pcm, data) - - 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 - - } - } - } -} - -func (r *Room) remove() { - log.Println("Closing room", r) - r.director.Done <- struct{}{} -} - -// startWebRTCSession fan-in of the same room to inputChannel -func startWebRTCSession(room *Room, webRTC *webrtc.WebRTC, playerIndex int) { - inputChannel := room.inputChannel - log.Println("room, inputChannel", room, inputChannel) - for { - select { - case <-webRTC.Done: - removeSession(webRTC, room) - default: - } - // Client stopped - if webRTC.IsClosed() { - return - } - - // encode frame - if webRTC.IsConnected() { - input := <-webRTC.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) * ui.NumKeys) - inputChannel <- input - } - } -} - -func cleanSession(w *webrtc.WebRTC) { - room, ok := rooms[w.RoomID] - if !ok { - return - } - removeSession(w, room) -} - -func removeSession(w *webrtc.WebRTC, room *Room) { - room.sessionsLock.Lock() - defer room.sessionsLock.Unlock() - for i, s := range room.rtcSessions { - if s == w { - room.rtcSessions = append(room.rtcSessions[:i], room.rtcSessions[i+1:]...) - break - } - } - // If room has no sessions, close room - if len(room.rtcSessions) == 0 { - room.Done <- struct{}{} - } -} - -func getServerIDOfRoom(oc *Client, roomID string) string { - log.Println("Request overlord roomID") - packet := oc.syncSend( - WSPacket{ - ID: "getRoom", - Data: roomID, - }, - ) - log.Println("Received roomID from overlord") - - return packet.Data -} - -func bridgeConnection(session *Session, serverID string, gameName string, roomID string, playerIndex int) { - log.Println("Bridging connection to other Host ", serverID) - client := session.client - // Ask client to init - - log.Println("Requesting offer to browser", serverID) - resp := client.syncSend(WSPacket{ - ID: "requestOffer", - Data: "", - }) - - log.Println("Sending offer to overlord to relay message to target host", resp.TargetHostID) - // Ask overlord to relay SDP packet to serverID - resp.TargetHostID = serverID - remoteTargetSDP := oclient.syncSend(resp) - log.Println("Got back remote host SDP, sending to browser") - // Send back remote SDP of remote server to browser - //client.syncSend(WSPacket{ - //ID: "sdp", - //Data: remoteTargetSDP.Data, - //}) - client.send(WSPacket{ - ID: "sdp", - Data: remoteTargetSDP.Data, - }, nil) - log.Println("Init session done, start game on target host") - - oclient.syncSend(WSPacket{ - ID: "start", - Data: gameName, - TargetHostID: serverID, - RoomID: roomID, - PlayerIndex: playerIndex, - }) - log.Println("Game is started on remote host") -} - -func createOverlordConnection() (*websocket.Conn, error) { - c, _, err := websocket.DefaultDialer.Dial(*config.OverlordHost, nil) - if err != nil { - return nil, err - } - - return c, nil -} - -func NewOverlordClient(oc *websocket.Conn) *Client { - oclient := NewClient(oc) - - // Received from overlord the serverID - oclient.receive( - "serverID", - func(response WSPacket) (request WSPacket) { - // Stick session with serverID got from overlord - log.Println("Received serverID ", response.Data) - serverID = response.Data - - return EmptyPacket - }, - ) - - // Received from overlord the sdp. This is happens when bridging - // TODO: refactor - oclient.receive( - "initwebrtc", - func(resp WSPacket) (req WSPacket) { - log.Println("Received a sdp request from overlord") - log.Println("Start peerconnection from the sdp") - peerconnection := webrtc.NewWebRTC() - // init new peerconnection from sessionID - localSession, err := peerconnection.StartClient(resp.Data, width, height) - peerconnections[resp.SessionID] = peerconnection - - if err != nil { - log.Fatalln(err) - } - - return WSPacket{ - ID: "sdp", - Data: localSession, - } - }, - ) - - // Received start from overlord. This is happens when bridging - // TODO: refactor - oclient.receive( - "start", - func(resp WSPacket) (req WSPacket) { - log.Println("Received a start request from overlord") - log.Println("Add the connection to current room on the host") - - peerconnection := peerconnections[resp.SessionID] - log.Println("start session") - roomID, isNewRoom := startSession(peerconnection, resp.Data, resp.RoomID, resp.PlayerIndex) - log.Println("Done, sending back") - // Bridge always access to old room - // TODO: log warn - if isNewRoom == true { - log.Fatal("Bridge should not spawn new room") - } - - req.ID = "start" - req.RoomID = roomID - return req - }, - ) - // heartbeat to keep pinging overlord. We not ping from server to browser, so we don't call heartbeat in browserClient - go oclient.heartbeat() - go oclient.listen() - - return oclient -} diff --git a/main_test.go b/main_test.go deleted file mode 100644 index 715d1a42..00000000 --- a/main_test.go +++ /dev/null @@ -1,151 +0,0 @@ -package main - -import ( - "fmt" - "net/http" - "net/http/httptest" - "strings" - "testing" - "time" - - gamertc "github.com/giongto35/cloud-game/webrtc" - "github.com/gorilla/websocket" - "github.com/pion/webrtc" -) - -var host = "http://localhost:8000" -var webrtcconfig = webrtc.Configuration{ICEServers: []webrtc.ICEServer{{URLs: []string{"stun:stun.l.google.com:19302"}}}} - -func initOverlord() *httptest.Server { - overlord := httptest.NewServer(http.HandlerFunc(wso)) - return overlord -} - -func initServer(t *testing.T, overlordURL string) *httptest.Server { - if overlordURL == "" { - oclient = nil - } else { - u := "ws" + strings.TrimPrefix(overlordURL, "http") - fmt.Println("connecting to overlord: ", u) - - oconn, _, err := websocket.DefaultDialer.Dial(u, nil) - if err != nil { - t.Fatalf("%v", err) - } - oclient = NewOverlordClient(oconn) - } - - server := httptest.NewServer(http.HandlerFunc(ws)) - return server -} - -func initClient(t *testing.T, host string) { - // Convert http://127.0.0.1 to ws://127.0.0. - u := "ws" + strings.TrimPrefix(host, "http") - - // Connect to the server - ws, _, err := websocket.DefaultDialer.Dial(u, nil) - if err != nil { - t.Fatalf("%v", err) - } - defer ws.Close() - - // Simulate peerconnection initialization from client - - peerConnection, err := webrtc.NewPeerConnection(webrtcconfig) - if err != nil { - t.Fatalf("%v", err) - } - - offer, err := peerConnection.CreateOffer(nil) - if err != nil { - t.Fatalf("%v", err) - } - - // Sets the LocalDescription, and starts our UDP listeners - err = peerConnection.SetLocalDescription(offer) - if err != nil { - panic(err) - } - - // Send offer to server - client := NewClient(ws) - go client.listen() - - fmt.Println("Sending offer...") - client.send(WSPacket{ - ID: "initwebrtc", - Data: gamertc.Encode(offer), - }, nil) - fmt.Println("Waiting sdp...") - - client.receive("sdp", func(resp WSPacket) WSPacket { - fmt.Println("received", resp.Data) - answer := webrtc.SessionDescription{} - gamertc.Decode(resp.Data, &answer) - // Apply the answer as the remote description - err = peerConnection.SetRemoteDescription(answer) - if err != nil { - panic(err) - } - - return EmptyPacket - }) - - time.Sleep(time.Second * 3) - fmt.Println("Sending start...") - - roomID := make(chan string) - client.send(WSPacket{ - ID: "start", - Data: "Contra.nes", - RoomID: "", - PlayerIndex: 1, - }, func(resp WSPacket) { - fmt.Println("Received response") - fmt.Println("RoomID:", resp.RoomID) - roomID <- resp.RoomID - }) - - respRoomID := <-roomID - if respRoomID == "" { - fmt.Println("RoomID should not be empty") - t.Fail() - } - fmt.Println("Done") - ws.Close() - // If receive roomID, the server is running correctly -} - -func TestSingleServerNoOverlord(t *testing.T) { - // Init slave server - oclient = nil - s := initServer(t, "") - defer s.Close() - - initClient(t, s.URL) -} - -func TestSingleServerOneOverlord(t *testing.T) { - o := initOverlord() - defer o.Close() - // Init slave server - s := initServer(t, o.URL) - defer s.Close() - - initClient(t, s.URL) - oclient.conn.Close() -} - -//func TestTwoServerOneOverlord(t *testing.T) { -//o := initOverlord() -//defer o.Close() -//// Init slave server -//s1 := initServer(t, o.URL) -//defer s1.Close() -//s2 := initServer(t, o.URL) -//defer s2.Close() - -//initClient(t, s1.URL) -//oclient.conn.Close() -//} diff --git a/overlord.go b/overlord.go deleted file mode 100644 index 4a9795f8..00000000 --- a/overlord.go +++ /dev/null @@ -1,123 +0,0 @@ -package main - -import ( - "fmt" - "log" - "math/rand" - "net/http" - "strconv" - - "github.com/giongto35/cloud-game/webrtc" -) - -var roomToServer = map[string]string{} - -// servers are the map serverID to server Client -var servers = map[string]*Client{} - -// If it's overlord, handle overlord connection (from host to overlord) -func wso(w http.ResponseWriter, r *http.Request) { - fmt.Println("Connected") - c, err := upgrader.Upgrade(w, r, nil) - if err != nil { - log.Print("[!] WS upgrade:", err) - return - } - defer c.Close() - - // register new server - serverID := strconv.Itoa(rand.Int()) - log.Println("A new server connected ", serverID) - - client := NewClient(c) - servers[serverID] = client - - wssession := &Session{ - client: client, - peerconnection: webrtc.NewWebRTC(), - // The server session is maintaining - } - - client.send( - WSPacket{ - ID: "serverID", - Data: serverID, - }, - nil, - ) - - client.receive("registerRoom", func(resp WSPacket) WSPacket { - log.Println("Received registerRoom ", resp.Data, serverID) - roomToServer[resp.Data] = serverID - return WSPacket{ - ID: "registerRoom", - } - }) - - client.receive("getRoom", func(resp WSPacket) WSPacket { - log.Println("Received a getroom request") - return WSPacket{ - ID: "getRoom", - Data: roomToServer[resp.Data], - } - }) - - client.receive("initwebrtc", func(resp WSPacket) WSPacket { - log.Println("Received a relay sdp request from a host") - // TODO: Abstract - if resp.TargetHostID != serverID { - log.Println("sending relay sdp to target host", resp) - // relay SDP to target host and get back sdp - // TODO: Async - sdp := servers[resp.TargetHostID].syncSend( - resp, - ) - - return sdp - } - log.Println("Target host is overlord itself: start peerconnection") - // If the target is in master - // start by its old - localSession, err := wssession.peerconnection.StartClient(resp.Data, width, height) - if err != nil { - log.Fatalln(err) - } - - return WSPacket{ - ID: "sdp", - Data: localSession, - } - }) - - // TODO: use relay ID type - // TODO: Merge sdp and start - client.receive("start", func(resp WSPacket) WSPacket { - log.Println("Received a relay start request from a host") - // TODO: Abstract - if resp.TargetHostID != serverID { - // relay SDP to target host and get back sdp - // TODO: Async - resp := servers[resp.TargetHostID].syncSend( - resp, - ) - - return resp - } - log.Println("Target host is overlord itself: start game") - // If the target is in master - // start by its old - roomID, isNewRoom := startSession(wssession.peerconnection, resp.Data, resp.RoomID, resp.PlayerIndex) - // Bridge always access to old room - // TODO: log warn - if isNewRoom == true { - log.Fatal("Bridge should not spawn new room") - } - - return WSPacket{ - ID: "start", - RoomID: roomID, - } - }) - - client.listen() -} diff --git a/overlord/overlord.go b/overlord/overlord.go index 8b137891..c73527a7 100644 --- a/overlord/overlord.go +++ b/overlord/overlord.go @@ -1 +1,143 @@ +package overlord +import ( + "fmt" + "log" + "math/rand" + "net/http" + "strconv" + + "github.com/giongto35/cloud-game/cws" + "github.com/gorilla/websocket" +) + +type Server struct { + roomToServer map[string]string + // servers are the map serverID to server Client + servers map[string]*cws.Client +} + +var upgrader = websocket.Upgrader{} + +func NewServer() *Server { + return &Server{ + servers: map[string]*cws.Client{}, + roomToServer: map[string]string{}, + } +} + +// If it's overlord, handle overlord connection (from host to overlord) +func (o *Server) WSO(w http.ResponseWriter, r *http.Request) { + fmt.Println("Connected") + c, err := upgrader.Upgrade(w, r, nil) + if err != nil { + 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) + + // 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 + //} + + // Sendback the ID to server + client.Send( + cws.WSPacket{ + ID: "serverID", + Data: serverID, + }, + nil, + ) + + // registerRoom event from a server, when server created a new room. + // RoomID is global so it is managed by overlord. + client.Receive("registerRoom", func(resp cws.WSPacket) cws.WSPacket { + log.Println("Overlord: Received registerRoom ", resp.Data, serverID) + o.roomToServer[resp.Data] = serverID + return cws.WSPacket{ + ID: "registerRoom", + } + }) + + // 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") + return cws.WSPacket{ + ID: "getRoom", + Data: o.roomToServer[resp.Data], + } + }) + + // Relay message from server to other target server + // TODO: Generalize + client.Receive("initwebrtc", func(resp cws.WSPacket) cws.WSPacket { + 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) + // relay SDP to target host and get back sdp + // TODO: Async + sdp := o.servers[resp.TargetHostID].SyncSend( + resp, + ) + + return sdp + } + log.Println("Overlord: Target host is overlord itself: start peerconnection") + // If the target is in master + // start by its old + //localSession, err := wssession.peerconnection.StartClient(resp.Data, width, height) + //if err != nil { + //log.Fatalln(err) + //} + + //return cws.WSPacket{ + //ID: "sdp", + //Data: localSession, + //} + return cws.EmptyPacket + }) + + // TODO: use relay ID type + // TODO: Merge sdp and start + client.Receive("start", func(resp cws.WSPacket) cws.WSPacket { + 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 + // TODO: Async + resp := o.servers[resp.TargetHostID].SyncSend( + resp, + ) + + return resp + } + log.Println("Overlord: Target host is overlord itself: start game") + //// If the target is in master + //// start by its old + //roomID, isNewRoom := startSession(wssession.peerconnection, resp.Data, resp.RoomID, resp.PlayerIndex) + //// Bridge always access to old room + //// TODO: log warn + //if isNewRoom == true { + //log.Fatal("Bridge should not spawn new room") + //} + + //return cws.WSPacket{ + //ID: "start", + //RoomID: roomID, + //} + return cws.EmptyPacket + }) + + client.Listen() +} diff --git a/webrtc/webrtc.go b/webrtc/webrtc.go index 2b52c4a1..a614ec55 100644 --- a/webrtc/webrtc.go +++ b/webrtc/webrtc.go @@ -16,6 +16,7 @@ import ( vpxEncoder "github.com/giongto35/cloud-game/vpx-encoder" "github.com/pion/webrtc" "github.com/pion/webrtc/pkg/media" + uuid "github.com/satori/go.uuid" ) var webrtcconfig = webrtc.Configuration{ICEServers: []webrtc.ICEServer{{URLs: []string{"stun:stun.l.google.com:19302"}}}} @@ -94,6 +95,8 @@ func Decode(in string, obj interface{}) { // NewWebRTC create func NewWebRTC() *WebRTC { w := &WebRTC{ + ID: uuid.Must(uuid.NewV4()).String(), + ImageChannel: make(chan []byte, 2), AudioChannel: make(chan []byte, 1000), InputChannel: make(chan int, 2), @@ -108,6 +111,8 @@ type InputDataPair struct { // WebRTC connection type WebRTC struct { + ID string + connection *webrtc.PeerConnection encoder *vpxEncoder.VpxEncoder isConnected bool @@ -167,20 +172,20 @@ func (w *WebRTC) StartClient(remoteSession string, width, height int) (string, e var d0 uint16 = 0 var d1 uint16 = 1 audioTrack, err := w.connection.CreateDataChannel("b", &webrtc.DataChannelInit{ - Ordered: &dfalse, + Ordered: &dfalse, MaxRetransmits: &d0, - Negotiated: &dtrue, - ID: &d1, + Negotiated: &dtrue, + ID: &d1, }) if err != nil { return "", err } // input channel - inputTrack, err := w.connection.CreateDataChannel("a", &webrtc.DataChannelInit{ - Ordered: &dtrue, + inputTrack, err := w.connection.CreateDataChannel("a", &webrtc.DataChannelInit{ + Ordered: &dtrue, Negotiated: &dtrue, - ID: &d0, + ID: &d0, }) inputTrack.OnOpen(func() { @@ -202,7 +207,6 @@ func (w *WebRTC) StartClient(remoteSession string, width, height int) (string, e w.Done <- struct{}{} close(w.Done) }) - // WebRTC state callback w.connection.OnICEConnectionStateChange(func(connectionState webrtc.ICEConnectionState) { @@ -225,8 +229,6 @@ func (w *WebRTC) StartClient(remoteSession string, width, height int) (string, e log.Println(iceCandidate) }) - - offer := webrtc.SessionDescription{} Decode(remoteSession, &offer)