diff --git a/assets/emulator/libretro/cores/mednafen_psx_hw_libretro.so b/assets/emulator/libretro/cores/mednafen_psx_hw_libretro.so old mode 100755 new mode 100644 diff --git a/assets/emulator/libretro/cores/mednafen_psx_libretro.dll b/assets/emulator/libretro/cores/mednafen_psx_libretro.dll old mode 100755 new mode 100644 diff --git a/assets/emulator/libretro/cores/mednafen_psx_libretro.dylib b/assets/emulator/libretro/cores/mednafen_psx_libretro.dylib old mode 100755 new mode 100644 diff --git a/assets/emulator/libretro/cores/mednafen_psx_libretro.so b/assets/emulator/libretro/cores/mednafen_psx_libretro.so old mode 100755 new mode 100644 diff --git a/assets/emulator/libretro/cores/mednafen_snes_libretro.dll b/assets/emulator/libretro/cores/mednafen_snes_libretro.dll old mode 100755 new mode 100644 diff --git a/assets/emulator/libretro/cores/mednafen_snes_libretro.dylib b/assets/emulator/libretro/cores/mednafen_snes_libretro.dylib old mode 100755 new mode 100644 diff --git a/assets/emulator/libretro/cores/mednafen_snes_libretro.so b/assets/emulator/libretro/cores/mednafen_snes_libretro.so old mode 100755 new mode 100644 diff --git a/assets/emulator/libretro/cores/mgba_libretro.dll b/assets/emulator/libretro/cores/mgba_libretro.dll old mode 100755 new mode 100644 diff --git a/assets/emulator/libretro/cores/mgba_libretro.dylib b/assets/emulator/libretro/cores/mgba_libretro.dylib old mode 100755 new mode 100644 diff --git a/assets/emulator/libretro/cores/nestopia_libretro.dll b/assets/emulator/libretro/cores/nestopia_libretro.dll old mode 100755 new mode 100644 diff --git a/assets/emulator/libretro/cores/nestopia_libretro.dylib b/assets/emulator/libretro/cores/nestopia_libretro.dylib old mode 100755 new mode 100644 diff --git a/assets/games/anguna.gba b/assets/games/anguna.gba old mode 100755 new mode 100644 diff --git a/hack/scripts/install_tools.sh b/hack/scripts/install_tools.sh old mode 100755 new mode 100644 diff --git a/pkg/coordinator/browser.go b/pkg/coordinator/browser.go index de04f3eb..7fb3064a 100644 --- a/pkg/coordinator/browser.go +++ b/pkg/coordinator/browser.go @@ -1,6 +1,7 @@ package coordinator import ( + "fmt" "log" "github.com/giongto35/cloud-game/pkg/cws" @@ -9,21 +10,81 @@ import ( type BrowserClient struct { *cws.Client + SessionID string + RoomID string + WorkerID string // TODO: how about pointer to workerClient? } -// RouteBrowser are all routes server accepts for browser -func (s *Session) RouteBrowser() { - browserClient := s.BrowserClient +// NewCoordinatorClient returns a client connecting to browser. +// This connection exchanges information between browser and coordinator. +func NewBrowserClient(c *websocket.Conn, browserID string) *BrowserClient { + return &BrowserClient{ + Client: cws.NewClient(c), + SessionID: browserID, + } +} - browserClient.Receive("heartbeat", func(resp cws.WSPacket) cws.WSPacket { +// Register new log +func (bc *BrowserClient) Printf(format string, args ...interface{}) { + newFmt := fmt.Sprintf("Browser %s] %s", bc.SessionID, format) + log.Printf(newFmt, args...) +} + +func (bc *BrowserClient) Println(args ...interface{}) { + msg := fmt.Sprintf("Browser %s] %s", bc.SessionID, fmt.Sprint(args...)) + log.Println(msg) +} + +// RouteBrowser +// Register callbacks for connection from browser -> coordinator +func (o *Server) RouteBrowser(client *BrowserClient) { + /* WebSocket */ + client.Receive("heartbeat", func(resp cws.WSPacket) cws.WSPacket { return resp }) - browserClient.Receive("icecandidate", func(resp cws.WSPacket) cws.WSPacket { - log.Println("Coordinator: Received icecandidate from a browser", resp.Data) - log.Println("Coordinator: Relay icecandidate from a browser to worker") + /* WebRTC */ + client.Receive("initwebrtc", func(resp cws.WSPacket) cws.WSPacket { + // initwebrtc now only sends signal to worker, asks it to createOffer + client.Printf("Received initwebrtc request -> relay to worker: %s", client.WorkerID) - wc, ok := s.handler.workerClients[s.ServerID] + // relay request to target worker + // worker creates a PeerConnection, and createOffer + // send SDP back to browser + resp.SessionID = client.SessionID + wc, ok := o.workerClients[client.WorkerID] + if !ok { + return cws.EmptyPacket + } + sdp := wc.SyncSend(resp) + + client.Println("Received SDP from worker -> sending back to browser") + return sdp + }) + + client.Receive("answer", func(resp cws.WSPacket) cws.WSPacket { + // contains SDP of browser createAnswer + // forward to worker + client.Println("Received browser answered SDP -> relay to worker") + + resp.SessionID = client.SessionID + wc, ok := o.workerClients[client.WorkerID] + if !ok { + return cws.EmptyPacket + } + wc.Send(resp, nil) + + // no need to response + return cws.EmptyPacket + }) + + client.Receive("candidate", func(resp cws.WSPacket) cws.WSPacket { + // contains ICE candidate of browser + // forward to worker + client.Println("Received IceCandidate from browser -> relay to worker") + + resp.SessionID = client.SessionID + wc, ok := o.workerClients[client.WorkerID] if !ok { return cws.EmptyPacket } @@ -32,119 +93,82 @@ func (s *Session) RouteBrowser() { return cws.EmptyPacket }) - browserClient.Receive("initwebrtc", func(resp cws.WSPacket) cws.WSPacket { - log.Println("Coordinator: Received sdp request from a browser") - log.Println("Coordinator: Relay sdp request from a browser to worker") + /* GameLogic */ + client.Receive("quit", func(resp cws.WSPacket) (req cws.WSPacket) { + client.Println("Received quit request from a browser -> relay to worker") - // relay SDP to target worker and get back SDP of the worker // TODO: Async - log.Println("Coordinator: serverID: ", s.ServerID, resp.SessionID) - resp.SessionID = s.ID - wc, ok := s.handler.workerClients[s.ServerID] + resp.SessionID = client.SessionID + wc, ok := o.workerClients[client.WorkerID] if !ok { return cws.EmptyPacket } - sdp := wc.SyncSend( - resp, - ) - - log.Println("Coordinator: Received sdp request from a worker") - log.Println("Coordinator: Sending back sdp to browser") - - return sdp - }) - - browserClient.Receive("quit", func(resp cws.WSPacket) (req cws.WSPacket) { - log.Println("Coordinator: Received quit request from a browser") - log.Println("Coordinator: Relay quit request from a browser to worker") - - // TODO: Async - resp.SessionID = s.ID - wc, ok := s.handler.workerClients[s.ServerID] - if !ok { - return cws.EmptyPacket - } - resp = wc.SyncSend( - resp, - ) + // Send but, waiting + wc.SyncSend(resp) return cws.EmptyPacket }) - browserClient.Receive("start", func(resp cws.WSPacket) cws.WSPacket { - log.Println("Coordinator: Received start request from a browser") - log.Println("Coordinator: Relay start request from a browser to worker") + client.Receive("start", func(resp cws.WSPacket) cws.WSPacket { + client.Println("Received start request from a browser -> relay to worker") + // TODO: Async - resp.SessionID = s.ID - wc, ok := s.handler.workerClients[s.ServerID] + resp.SessionID = client.SessionID + wc, ok := o.workerClients[client.WorkerID] if !ok { return cws.EmptyPacket } - workerResp := wc.SyncSend( - resp, - ) + workerResp := wc.SyncSend(resp) + // Response from worker contains initialized roomID. Set roomID to the session - s.RoomID = workerResp.RoomID - log.Println("Received room response from browser: ", workerResp.RoomID) + client.RoomID = workerResp.RoomID + client.Println("Received room response from browser: ", workerResp.RoomID) return workerResp }) - browserClient.Receive("save", func(resp cws.WSPacket) (req cws.WSPacket) { - log.Println("Coordinator: Received save request from a browser") - log.Println("Coordinator: Relay save request from a browser to worker") + client.Receive("save", func(resp cws.WSPacket) (req cws.WSPacket) { + client.Println("Received save request from a browser -> relay to worker") + // TODO: Async - resp.SessionID = s.ID - resp.RoomID = s.RoomID - wc, ok := s.handler.workerClients[s.ServerID] + resp.SessionID = client.SessionID + resp.RoomID = client.RoomID + wc, ok := o.workerClients[client.WorkerID] if !ok { return cws.EmptyPacket } - resp = wc.SyncSend( - resp, - ) + resp = wc.SyncSend(resp) return resp }) - browserClient.Receive("load", func(resp cws.WSPacket) (req cws.WSPacket) { - log.Println("Coordinator: Received load request from a browser") - log.Println("Coordinator: Relay load request from a browser to worker") + client.Receive("load", func(resp cws.WSPacket) (req cws.WSPacket) { + client.Println("Received load request from a browser -> relay to worker") + // TODO: Async - resp.SessionID = s.ID - resp.RoomID = s.RoomID - wc, ok := s.handler.workerClients[s.ServerID] + resp.SessionID = client.SessionID + resp.RoomID = client.RoomID + wc, ok := o.workerClients[client.WorkerID] if !ok { return cws.EmptyPacket } - resp = wc.SyncSend( - resp, - ) + resp = wc.SyncSend(resp) return resp }) - browserClient.Receive("playerIdx", func(resp cws.WSPacket) (req cws.WSPacket) { - log.Println("Coordinator: Received update player index request from a browser") - log.Println("Coordinator: Relay update player index request from a browser to worker") + client.Receive("playerIdx", func(resp cws.WSPacket) (req cws.WSPacket) { + client.Println("Received update player index request from a browser -> relay to worker") + // TODO: Async - resp.SessionID = s.ID - resp.RoomID = s.RoomID - wc, ok := s.handler.workerClients[s.ServerID] + resp.SessionID = client.SessionID + resp.RoomID = client.RoomID + wc, ok := o.workerClients[client.WorkerID] if !ok { return cws.EmptyPacket } - resp = wc.SyncSend( - resp, - ) + resp = wc.SyncSend(resp) return resp }) } - -// NewCoordinatorClient 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/pkg/coordinator/handlers.go b/pkg/coordinator/handlers.go index 143e627c..3ded9905 100644 --- a/pkg/coordinator/handlers.go +++ b/pkg/coordinator/handlers.go @@ -28,6 +28,8 @@ type Server struct { roomToWorker map[string]string // workerClients are the map workerID to worker Client workerClients map[string]*WorkerClient + // browserClients are the map sessionID to browser Client + browserClients map[string]*BrowserClient } const pingServerTemp = "https://%s.%s/echo" @@ -39,10 +41,12 @@ var errNotFound = errors.New("Not found") func NewServer(cfg Config) *Server { return &Server{ cfg: cfg, - // Mapping serverID to client - workerClients: map[string]*WorkerClient{}, // Mapping roomID to server roomToWorker: map[string]string{}, + // Mapping workerID to worker + workerClients: map[string]*WorkerClient{}, + // Mapping sessionID to browser + browserClients: map[string]*BrowserClient{}, } } @@ -80,74 +84,114 @@ func (o *Server) getPingServer(zone string) string { // WSO handles all connections from a new worker to coordinator func (o *Server) WSO(w http.ResponseWriter, r *http.Request) { - fmt.Println("Connected") + log.Println("Coordinator: A worker is connecting...") + + // be aware of ReadBufferSize, WriteBufferSize (default 4096) + // https://pkg.go.dev/github.com/gorilla/websocket?tab=doc#Upgrader c, err := upgrader.Upgrade(w, r, nil) if err != nil { - log.Print("Coordinator: [!] WS upgrade:", err) + log.Println("Coordinator: [!] WS upgrade:", err) return } - // Register new server - serverID := uuid.Must(uuid.NewV4()).String() - log.Println("Coordinator: A new server connected to Coordinator", serverID) + + // Generate workerID + var workerID string + for { + workerID = uuid.Must(uuid.NewV4()).String() + // check duplicate + if _, ok := o.workerClients[workerID]; !ok { + break + } + } + + // Create a workerClient instance + wc := NewWorkerClient(c, workerID) + wc.Println("Generated worker ID") // Register to workersClients map the client connection address := util.GetRemoteAddress(c) + wc.Println("Address:", address) // Zone of the worker zone := r.URL.Query().Get("zone") + wc.Printf("Is public: %v zone: %v", util.IsPublicIP(address), zone) pingServer := o.getPingServer(zone) - fmt.Printf("Is public: %v zone: %v\n", util.IsPublicIP(address), zone) - // In case worker and coordinator in the same host if !util.IsPublicIP(address) && *config.Mode == config.ProdEnv { // Don't accept private IP for worker's address in prod mode // However, if the worker in the same host with coordinator, we can get public IP of worker - log.Printf("Error: address %s is invalid", address) + wc.Printf("[!] Address %s is invalid", address) + address = util.GetHostPublicIP() - log.Println("Find public address:", address) + wc.Printf("Find public address: %s", address) + if address == "" || !util.IsPublicIP(address) { // Skip this worker because we cannot find public IP + wc.Println("[!] Unable to find public address, reject worker") return } } - client := NewWorkerClient(c, serverID, address, fmt.Sprintf(config.StunTurnTemplate, address, address), zone, pingServer) - o.workerClients[serverID] = client - defer o.cleanConnection(client, serverID) - // Sendback the ID to server - client.Send( - cws.WSPacket{ - ID: "serverID", - Data: serverID, - }, - nil, - ) - o.RouteWorker(client) + // Create a workerClient instance + wc.Address = address + wc.StunTurnServer = fmt.Sprintf(config.StunTurnTemplate, address, address) + wc.Zone = zone + wc.PingServer = pingServer - client.Listen() + // Eveything is cool + // Attach to Server instance with workerID, add defer + o.workerClients[workerID] = wc + defer o.cleanWorker(wc, workerID) + + // Sendback the ID to worker + // TODO: do we need this packet? + wc.Send(cws.WSPacket{ + ID: "serverID", + Data: workerID, + }, nil) + + // Add receiver callbacks, and listen + o.RouteWorker(wc) + wc.Listen() } // WSO handles all connections from user/frontend to coordinator func (o *Server) WS(w http.ResponseWriter, r *http.Request) { - log.Println("A user connected to coordinator ", r.URL) + log.Println("Coordinator: A user is connecting...") defer func() { if r := recover(); r != nil { log.Println("Warn: Something wrong. Recovered in ", r) } }() + // be aware of ReadBufferSize, WriteBufferSize (default 4096) + // https://pkg.go.dev/github.com/gorilla/websocket?tab=doc#Upgrader c, err := upgrader.Upgrade(w, r, nil) if err != nil { - log.Print("[!] WS upgrade:", err) + log.Println("Coordinator: [!] WS upgrade:", err) return } - defer c.Close() - client := NewBrowserClient(c) - go client.Listen() + // Generate sessionID for browserClient + var sessionID string + for { + sessionID = uuid.Must(uuid.NewV4()).String() + // check duplicate + if _, ok := o.browserClients[sessionID]; !ok { + break + } + } - var workerClient *WorkerClient + // Create browserClient instance + bc := NewBrowserClient(c, sessionID) + bc.Println("Generated worker ID") + + // Run browser listener first (to capture ping) + go bc.Listen() + + /* Create a session - mapping browserClient with workerClient */ + var wc *WorkerClient // get roomID if it is embeded in request. Server will pair the frontend with the server running the room. It only happens when we are trying to access a running room over share link. // TODO: Update link to the wiki @@ -155,76 +199,71 @@ func (o *Server) WS(w http.ResponseWriter, r *http.Request) { // zone param is to pick worker in that zone only // if there is no zone param, we can pic userZone := r.URL.Query().Get("zone") - log.Printf("Get Room %s Zone %s From URL %v", roomID, userZone, r.URL) + + bc.Printf("Get Room %s Zone %s From URL %v", roomID, userZone, r.URL) if roomID != "" { - log.Printf("Detected roomID %v from URL", roomID) + bc.Printf("Detected roomID %v from URL", roomID) if workerID, ok := o.roomToWorker[roomID]; ok { - workerClient = o.workerClients[workerID] - if userZone != "" && workerClient.Zone != userZone { + wc = o.workerClients[workerID] + if userZone != "" && wc.Zone != userZone { // if there is zone param, we need to ensure ther worker in that zone // if not we consider the room is missing - workerClient = nil + wc = nil } else { - log.Printf("Found running server with id=%v client=%v", workerID, workerClient) + bc.Printf("Found running server with id=%v client=%v", workerID, wc) } } } // If there is no existing server to connect to, we find the best possible worker for the frontend - if workerClient == nil { + if wc == nil { // Get best server for frontend to connect to - workerClient, err = o.getBestWorkerClient(client, userZone) + wc, err = o.getBestWorkerClient(bc, userZone) if err != nil { return } } - // SessionID will be the unique per frontend connection - sessionID := uuid.Must(uuid.NewV4()).String() - // Setup session - wssession := &Session{ - ID: sessionID, - handler: o, - BrowserClient: client, - WorkerClient: workerClient, - ServerID: workerClient.ServerID, - } - // TODO:? - // defer wssession.Close() - log.Println("New client will conect to server", wssession.ServerID) - wssession.WorkerClient.IsAvailable = false + // Assign available worker to browserClient + bc.WorkerID = wc.WorkerID + wc.IsAvailable = false - wssession.RouteBrowser() + // Everything is cool + // Attach to Server instance with sessionID + o.browserClients[sessionID] = bc + defer o.cleanBrowser(bc, sessionID) - wssession.BrowserClient.Send(cws.WSPacket{ + // Routing browserClient message + o.RouteBrowser(bc) + + bc.Send(cws.WSPacket{ ID: "init", - Data: createInitPackage(workerClient.StunTurnServer), + Data: createInitPackage(wc.StunTurnServer), }, nil) // If peerconnection is done (client.Done is signalled), we close peerconnection - <-client.Done + <-bc.Done + // Notify worker to clean session - wssession.WorkerClient.Send( - cws.WSPacket{ - ID: "terminateSession", - SessionID: sessionID, - }, - nil, - ) + wc.Send(cws.WSPacket{ + ID: "terminateSession", + SessionID: sessionID, + }, nil) + // WorkerClient become available again - wssession.WorkerClient.IsAvailable = true + wc.IsAvailable = true } func (o *Server) getBestWorkerClient(client *BrowserClient, zone string) (*WorkerClient, error) { if o.cfg.DebugHost != "" { - log.Println("Connecting to debug host instead prod servers", o.cfg.DebugHost) + client.Println("Connecting to debug host instead prod servers", o.cfg.DebugHost) wc := o.getWorkerFromAddress(o.cfg.DebugHost) if wc != nil { return wc, nil } // if there is not debugHost, continue usual flow - log.Println("Not found, connecting to all available servers") + client.Println("Not found, connecting to all available servers") } workerClients := o.getAvailableWorkers() @@ -270,7 +309,7 @@ func (o *Server) findBestServerFromBrowser(workerClients map[string]*WorkerClien } latencies := o.getLatencyMapFromBrowser(workerClients, client) - log.Println("Latency map", latencies) + client.Println("Latency map", latencies) if len(latencies) == 0 { return "", errors.New("No server found") @@ -292,7 +331,7 @@ func (o *Server) findBestServerFromBrowser(workerClients map[string]*WorkerClien } } - return bestWorker.ServerID, nil + return bestWorker.WorkerID, nil } // getLatencyMapFromBrowser get all latencies from worker to user @@ -308,7 +347,7 @@ func (o *Server) getLatencyMapFromBrowser(workerClients map[string]*WorkerClient } // send this address to user and get back latency - log.Println("Send sync", addressList, strings.Join(addressList, ",")) + client.Println("Send sync", addressList, strings.Join(addressList, ",")) data := client.SyncSend(cws.WSPacket{ ID: "checkLatency", Data: strings.Join(addressList, ","), @@ -329,20 +368,28 @@ func (o *Server) getLatencyMapFromBrowser(workerClients map[string]*WorkerClient return latencyMap } -// cleanConnection is called when a worker is disconnected -// connection from worker (client) to server is also closed -func (o *Server) cleanConnection(client *WorkerClient, serverID string) { - log.Println("Unregister server from coordinator") - // Remove serverID from servers - delete(o.workerClients, serverID) +// cleanBrowser is called when a browser is disconnected +func (o *Server) cleanBrowser(bc *BrowserClient, sessionID string) { + bc.Println("Disconnect from coordinator") + delete(o.browserClients, sessionID) + bc.Close() +} + +// cleanWorker is called when a worker is disconnected +// connection from worker to coordinator is also closed +func (o *Server) cleanWorker(wc *WorkerClient, workerID string) { + wc.Println("Unregister worker from coordinator") + // Remove workerID from workerClients + delete(o.workerClients, workerID) // Clean all rooms connecting to that server for roomID, roomServer := range o.roomToWorker { - if roomServer == serverID { + if roomServer == workerID { + wc.Printf("Remove room %s", roomID) delete(o.roomToWorker, roomID) } } - client.Close() + wc.Close() } // createInitPackage returns serverhost + game list in encoded wspacket format diff --git a/pkg/coordinator/session.go b/pkg/coordinator/session.go deleted file mode 100644 index fabf5e58..00000000 --- a/pkg/coordinator/session.go +++ /dev/null @@ -1,19 +0,0 @@ -package coordinator - -// Session represents a session connected from the browser to the current server -// It requires one connection to browser and one connection to the coordinator -// connection to browser is 1-1. connection to coordinator is n - 1 -// Peerconnection can be from other server to ensure better latency -type Session struct { - ID string - BrowserClient *BrowserClient - WorkerClient *WorkerClient - // CoordinatorClient *CoordinatorClient - // peerconnection *webrtc.WebRTC - - // TODO: Decouple this - handler *Server - - ServerID string - RoomID string -} diff --git a/pkg/coordinator/worker.go b/pkg/coordinator/worker.go index b26dc37a..f354a22d 100644 --- a/pkg/coordinator/worker.go +++ b/pkg/coordinator/worker.go @@ -1,6 +1,7 @@ package coordinator import ( + "fmt" "log" "github.com/giongto35/cloud-game/pkg/cws" @@ -11,7 +12,7 @@ const pingServer = "%s://%s/echo" type WorkerClient struct { *cws.Client - ServerID string + WorkerID string Address string // ip address of worker // public server used for ping check (Cannot use worker address because they are not publicly exposed) PingServer string @@ -20,13 +21,33 @@ type WorkerClient struct { Zone string } +// NewWorkerClient returns a client connecting to worker. This connection exchanges information between workers and server +func NewWorkerClient(c *websocket.Conn, workerID string) *WorkerClient { + return &WorkerClient{ + Client: cws.NewClient(c), + WorkerID: workerID, + IsAvailable: true, + } +} + +// Register new log +func (wc *WorkerClient) Printf(format string, args ...interface{}) { + newFmt := fmt.Sprintf("Worker %s] %s", wc.WorkerID, format) + log.Printf(newFmt, args...) +} + +func (wc *WorkerClient) Println(args ...interface{}) { + msg := fmt.Sprintf("Worker %s] %s", wc.WorkerID, fmt.Sprint(args...)) + log.Println(msg) +} + // RouteWorker are all routes server received from worker -func (o *Server) RouteWorker(workerClient *WorkerClient) { +func (o *Server) RouteWorker(wc *WorkerClient) { // registerRoom event from a worker, when worker created a new room. // RoomID is global so it is managed by coordinator. - workerClient.Receive("registerRoom", func(resp cws.WSPacket) cws.WSPacket { - log.Printf("Coordinator: Received registerRoom room %s from worker %s", resp.Data, workerClient.ServerID) - o.roomToWorker[resp.Data] = workerClient.ServerID + wc.Receive("registerRoom", func(resp cws.WSPacket) cws.WSPacket { + log.Printf("Coordinator: Received registerRoom room %s from worker %s", resp.Data, wc.WorkerID) + o.roomToWorker[resp.Data] = wc.WorkerID log.Printf("Coordinator: Current room list is: %+v", o.roomToWorker) return cws.WSPacket{ @@ -35,8 +56,8 @@ func (o *Server) RouteWorker(workerClient *WorkerClient) { }) // closeRoom event from a worker, when worker close a room - workerClient.Receive("closeRoom", func(resp cws.WSPacket) cws.WSPacket { - log.Printf("Coordinator: Received closeRoom room %s from worker %s", resp.Data, workerClient.ServerID) + wc.Receive("closeRoom", func(resp cws.WSPacket) cws.WSPacket { + log.Printf("Coordinator: Received closeRoom room %s from worker %s", resp.Data, wc.WorkerID) delete(o.roomToWorker, resp.Data) log.Printf("Coordinator: Current room list is: %+v", o.roomToWorker) @@ -46,7 +67,7 @@ func (o *Server) RouteWorker(workerClient *WorkerClient) { }) // getRoom returns the server ID based on requested roomID. - workerClient.Receive("getRoom", func(resp cws.WSPacket) cws.WSPacket { + wc.Receive("getRoom", func(resp cws.WSPacket) cws.WSPacket { log.Println("Coordinator: Received a getroom request") log.Println("Result: ", o.roomToWorker[resp.Data]) return cws.WSPacket{ @@ -55,20 +76,22 @@ func (o *Server) RouteWorker(workerClient *WorkerClient) { } }) - workerClient.Receive("heartbeat", func(resp cws.WSPacket) cws.WSPacket { + wc.Receive("heartbeat", func(resp cws.WSPacket) cws.WSPacket { return resp }) -} -// NewWorkerClient returns a client connecting to worker. This connection exchanges information between workers and server -func NewWorkerClient(c *websocket.Conn, serverID string, address string, stunturn string, zone, pingServer string) *WorkerClient { - return &WorkerClient{ - Client: cws.NewClient(c), - ServerID: serverID, - PingServer: pingServer, - Address: address, - StunTurnServer: stunturn, - IsAvailable: true, - Zone: zone, - } + /* WebRTC */ + wc.Receive("candidate", func(resp cws.WSPacket) cws.WSPacket { + wc.Println("Received IceCandidate from worker -> relay to browser") + bc, ok := o.browserClients[resp.SessionID] + if ok { + // Remove SessionID while sending back to browser + resp.SessionID = "" + bc.Send(resp, nil) + } else { + wc.Println("Error: unknown SessionID:", resp.SessionID) + } + + return cws.EmptyPacket + }) } diff --git a/pkg/cws/cws.go b/pkg/cws/cws.go index 547f43dd..7f700b0d 100644 --- a/pkg/cws/cws.go +++ b/pkg/cws/cws.go @@ -35,9 +35,8 @@ type WSPacket struct { RoomID string `json:"room_id"` PlayerIndex int `json:"player_index"` - TargetHostID string `json:"target_id"` - PacketID string `json:"packet_id"` - // Globally ID of a session + PacketID string `json:"packet_id"` + // Globally ID of a browser session SessionID string `json:"session_id"` } diff --git a/pkg/util/netutil.go b/pkg/util/netutil.go index 5132b90d..855a7e78 100644 --- a/pkg/util/netutil.go +++ b/pkg/util/netutil.go @@ -1,7 +1,6 @@ package util import ( - "log" "net" "strings" @@ -11,7 +10,7 @@ import ( // GetRemoteAddress returns public address of websocket connection func GetRemoteAddress(conn *websocket.Conn) string { var remoteAddr string - log.Println("Address :", conn.RemoteAddr().String()) + // log.Println("Address :", conn.RemoteAddr().String()) if parts := strings.Split(conn.RemoteAddr().String(), ":"); len(parts) == 2 { remoteAddr = parts[0] } diff --git a/pkg/webrtc/webrtc.go b/pkg/webrtc/webrtc.go index a71aa034..09ab9eec 100644 --- a/pkg/webrtc/webrtc.go +++ b/pkg/webrtc/webrtc.go @@ -17,42 +17,32 @@ import ( "github.com/pion/webrtc/v2/pkg/media" ) +// TODO: double check if no need TURN server here var webrtcconfig = webrtc.Configuration{ICEServers: []webrtc.ICEServer{{URLs: []string{"stun:stun.l.google.com:19302"}}}} -// Allows compressing offer/answer to bypass terminal input limits. -const compress = false - // Encode encodes the input in base64 -// It can optionally zip the input before encoding -func Encode(obj interface{}) string { +func Encode(obj interface{}) (string, error) { b, err := json.Marshal(obj) if err != nil { - panic(err) + return "", err } - if compress { - b = zip(b) - } - - return base64.StdEncoding.EncodeToString(b) + return base64.StdEncoding.EncodeToString(b), nil } // Decode decodes the input from base64 -// It can optionally unzip the input after decoding -func Decode(in string, obj interface{}) { +func Decode(in string, obj interface{}) error { b, err := base64.StdEncoding.DecodeString(in) if err != nil { - panic(err) - } - - if compress { - b = unzip(b) + return err } err = json.Unmarshal(b, obj) if err != nil { - panic(err) + return err } + + return nil } // NewWebRTC create @@ -99,8 +89,10 @@ type GameMeta struct { PlayerIndex int } +type OnIceCallback func(candidate string) + // StartClient start webrtc -func (w *WebRTC) StartClient(remoteSession string, isMobile bool, iceCandidates []string) (string, error) { +func (w *WebRTC) StartClient(isMobile bool, iceCB OnIceCallback) (string, error) { defer func() { if err := recover(); err != nil { log.Println(err) @@ -122,10 +114,11 @@ func (w *WebRTC) StartClient(remoteSession string, isMobile bool, iceCandidates return "", err } + // add video track if util.GetVideoEncoder(isMobile) == config.CODEC_H264 { - videoTrack, err = w.connection.NewTrack(webrtc.DefaultPayloadTypeH264, rand.Uint32(), "video", "pion2") + videoTrack, err = w.connection.NewTrack(webrtc.DefaultPayloadTypeH264, rand.Uint32(), "video", "game-video") } else { - videoTrack, err = w.connection.NewTrack(webrtc.DefaultPayloadTypeVP8, rand.Uint32(), "video", "pion2") + videoTrack, err = w.connection.NewTrack(webrtc.DefaultPayloadTypeVP8, rand.Uint32(), "video", "game-video") } if err != nil { return "", err @@ -135,9 +128,10 @@ func (w *WebRTC) StartClient(remoteSession string, isMobile bool, iceCandidates if err != nil { return "", err } + log.Println("Add video track") - // audio track - opusTrack, err := w.connection.NewTrack(webrtc.DefaultPayloadTypeOpus, rand.Uint32(), "audio", "pion2b") + // add audio track + opusTrack, err := w.connection.NewTrack(webrtc.DefaultPayloadTypeOpus, rand.Uint32(), "audio", "game-audio") if err != nil { return "", err } @@ -145,17 +139,11 @@ func (w *WebRTC) StartClient(remoteSession string, isMobile bool, iceCandidates if err != nil { return "", err } + log.Println("Add audio track") - dfalse := false - dtrue := true - var d0 uint16 = 0 - - // input channel - inputTrack, err := w.connection.CreateDataChannel("a", &webrtc.DataChannelInit{ - Ordered: &dfalse, - Negotiated: &dtrue, - ID: &d0, - }) + // create data channel for input, and register callbacks + // order: true, negotiated: false, id: random + inputTrack, err := w.connection.CreateDataChannel("game-input", nil) inputTrack.OnOpen(func() { log.Printf("Data channel '%s'-'%d' open.\n", inputTrack.Label(), inputTrack.ID()) @@ -188,44 +176,39 @@ func (w *WebRTC) StartClient(remoteSession string, isMobile bool, iceCandidates } }) - // TODO: take a look at this w.connection.OnICECandidate(func(iceCandidate *webrtc.ICECandidate) { - log.Println(iceCandidate) + if iceCandidate != nil { + log.Println("OnIceCandidate:", iceCandidate.ToJSON().Candidate) + candidate, err := Encode(iceCandidate.ToJSON()) + if err != nil { + log.Println("Encode IceCandidate failed: " + iceCandidate.ToJSON().Candidate) + return + } + iceCB(candidate) + } else { + // finish, send null + iceCB("") + } + }) - offer := webrtc.SessionDescription{} + // Stream provider supposes to send offer + offer, err := w.connection.CreateOffer(nil) + if err != nil { + return "", err + } + log.Println("Created Offer") - Decode(remoteSession, &offer) - - err = w.connection.SetRemoteDescription(offer) + err = w.connection.SetLocalDescription(offer) if err != nil { return "", err } - // Parse candidates list - // This logic is wrong - for _, bcandidate := range iceCandidates { - iceCandidate := webrtc.ICECandidateInit{} - if err := json.Unmarshal([]byte(bcandidate), &iceCandidate); err != nil { - log.Println("Cannot parse ", bcandidate) - continue - } - log.Println("Add iceCandidate: ", iceCandidate) - w.connection.AddICECandidate(iceCandidate) - } - - answer, err := w.connection.CreateAnswer(nil) + localSession, err := Encode(offer) if err != nil { return "", err } - err = w.connection.SetLocalDescription(answer) - if err != nil { - return "", err - } - - // Sendback answer from server - localSession := Encode(answer) return localSession, nil } @@ -233,12 +216,41 @@ func (w *WebRTC) AttachRoomID(roomID string) { w.RoomID = roomID } -// TODO: Take a look at this -func (w *WebRTC) AddCandidate(candidate webrtc.ICECandidateInit) { - err := w.connection.AddICECandidate(candidate) +func (w *WebRTC) SetRemoteSDP(remoteSDP string) error { + var answer webrtc.SessionDescription + err := Decode(remoteSDP, &answer) if err != nil { - log.Println("Cannot add candidate: ", err) + log.Println("Decode remote sdp from peer failed") + return err } + + err = w.connection.SetRemoteDescription(answer) + if err != nil { + log.Println("Set remote description from peer failed") + return err + } + + log.Println("Set Remote Description") + return nil +} + +func (w *WebRTC) AddCandidate(candidate string) error { + var iceCandidate webrtc.ICECandidateInit + err := Decode(candidate, &iceCandidate) + if err != nil { + log.Println("Decode Ice candidate from peer failed") + return err + } + log.Println("Decoded Ice: " + iceCandidate.Candidate) + + err = w.connection.AddICECandidate(iceCandidate) + if err != nil { + log.Println("Add Ice candidate from peer failed") + return err + } + + log.Println("Add Ice Candidate: " + iceCandidate.Candidate) + return nil } // StopClient disconnect @@ -266,7 +278,6 @@ func (w *WebRTC) IsConnected() bool { return w.isConnected } -// func (w *WebRTC) startStreaming(vp8Track *webrtc.Track, opusTrack *webrtc.Track) { func (w *WebRTC) startStreaming(vp8Track *webrtc.Track, opusTrack *webrtc.Track) { log.Println("Start streaming") // receive frame buffer diff --git a/pkg/worker/coordinator.go b/pkg/worker/coordinator.go index c7dbe98a..c7d6eaa6 100644 --- a/pkg/worker/coordinator.go +++ b/pkg/worker/coordinator.go @@ -32,9 +32,11 @@ func NewCoordinatorClient(oc *websocket.Conn) *CoordinatorClient { // RouteCoordinator are all routes server received from coordinator func (h *Handler) RouteCoordinator() { - iceCandidates := map[string][]string{} + // iceCandidates := map[string][]string{} oClient := h.oClient + /* Coordinator */ + // Received from coordinator the serverID oClient.Receive( "serverID", @@ -47,21 +49,36 @@ func (h *Handler) RouteCoordinator() { }, ) + /* WebRTC Connection */ + oClient.Receive( "initwebrtc", func(resp cws.WSPacket) (req cws.WSPacket) { - log.Println("Received relay SDP of a browser from coordinator") + log.Println("Received a request to createOffer from browser via coordinator") peerconnection := webrtc.NewWebRTC() var initPacket struct { - SDP string `json:"sdp"` - IsMobile bool `json:"is_mobile"` + IsMobile bool `json:"is_mobile"` } err := json.Unmarshal([]byte(resp.Data), &initPacket) if err != nil { - panic(err) + log.Println("Error: Cannot decode json:", err) + return cws.EmptyPacket } - localSession, err := peerconnection.StartClient(initPacket.SDP, initPacket.IsMobile, iceCandidates[resp.SessionID]) + + localSession, err := peerconnection.StartClient( + initPacket.IsMobile, + func(candidate string) { + // send back candidate string to browser + oClient.Send(cws.WSPacket{ + ID: "candidate", + Data: candidate, + SessionID: resp.SessionID, + }, nil) + }, + ) + + // localSession, err := peerconnection.StartClient(initPacket.IsMobile, iceCandidates[resp.SessionID]) // h.peerconnections[resp.SessionID] = peerconnection // Create new sessions when we have new peerconnection initialized @@ -69,26 +86,73 @@ func (h *Handler) RouteCoordinator() { peerconnection: peerconnection, } h.sessions[resp.SessionID] = session - log.Println("Start peerconnection", resp.SessionID) + if err != nil { log.Println("Error: Cannot create new webrtc session", err) return cws.EmptyPacket } return cws.WSPacket{ - ID: "sdp", - Data: localSession, - SessionID: resp.SessionID, + ID: "offer", + Data: localSession, } }, ) + oClient.Receive( + "answer", + func(resp cws.WSPacket) (req cws.WSPacket) { + log.Println("Received answer SDP from browser") + session := h.getSession(resp.SessionID) + + if session != nil { + peerconnection := session.peerconnection + + err := peerconnection.SetRemoteSDP(resp.Data) + if err != nil { + log.Println("Error: Cannot set RemoteSDP of client: " + resp.SessionID) + } + } else { + log.Printf("Error: No session for ID: %s\n", resp.SessionID) + } + + return cws.EmptyPacket + }, + ) + + oClient.Receive( + "candidate", + func(resp cws.WSPacket) (req cws.WSPacket) { + log.Println("Received remote Ice Candidate from browser") + session := h.getSession(resp.SessionID) + + if session != nil { + peerconnection := session.peerconnection + + err := peerconnection.AddCandidate(resp.Data) + if err != nil { + log.Println("Error: Cannot add IceCandidate of client: " + resp.SessionID) + } + } else { + log.Printf("Error: No session for ID: %s\n", resp.SessionID) + } + + return cws.EmptyPacket + }, + ) + + /* Game Logic */ + oClient.Receive( "start", func(resp cws.WSPacket) (req cws.WSPacket) { log.Println("Received a start request from coordinator") - session, _ := h.sessions[resp.SessionID] + session := h.getSession(resp.SessionID) + if session == nil { + log.Printf("Error: No session for ID: %s\n", resp.SessionID) + return cws.EmptyPacket + } peerconnection := session.peerconnection // TODO: Standardize for all types of packet. Make WSPacket generic @@ -99,7 +163,8 @@ func (h *Handler) RouteCoordinator() { err := json.Unmarshal([]byte(resp.Data), &startPacket) if err != nil { - panic(err) + log.Println("Error: Cannot decode json:", err) + return cws.EmptyPacket } room := h.startGameHandler(startPacket.GameName, resp.RoomID, resp.PlayerIndex, peerconnection, util.GetVideoEncoder(startPacket.IsMobile)) @@ -118,13 +183,16 @@ func (h *Handler) RouteCoordinator() { "quit", func(resp cws.WSPacket) (req cws.WSPacket) { log.Println("Received a quit request from coordinator") - session, ok := h.sessions[resp.SessionID] - log.Println("Find ", resp.SessionID, session, ok) + session := h.getSession(resp.SessionID) - room := h.getRoom(session.RoomID) - // Defensive coding, check if the peerconnection is in room - if room.IsPCInRoom(session.peerconnection) { - h.detachPeerConn(session.peerconnection) + if session != nil { + room := h.getRoom(session.RoomID) + // Defensive coding, check if the peerconnection is in room + if room.IsPCInRoom(session.peerconnection) { + h.detachPeerConn(session.peerconnection) + } + } else { + log.Printf("Error: No session for ID: %s\n", resp.SessionID) } return cws.EmptyPacket @@ -197,26 +265,17 @@ func (h *Handler) RouteCoordinator() { return req }) - oClient.Receive( - "icecandidate", - func(resp cws.WSPacket) (req cws.WSPacket) { - log.Println("Received a icecandidate from coordinator: ", resp.Data) - iceCandidates[resp.SessionID] = append(iceCandidates[resp.SessionID], resp.Data) - - return cws.EmptyPacket - }, - ) - oClient.Receive( "terminateSession", func(resp cws.WSPacket) (req cws.WSPacket) { log.Println("Received a terminate session ", resp.SessionID) - session, ok := h.sessions[resp.SessionID] - log.Println("Find ", session, ok) - if ok { + session := h.getSession(resp.SessionID) + if session != nil { session.Close() delete(h.sessions, resp.SessionID) h.detachPeerConn(session.peerconnection) + } else { + log.Printf("Error: No session for ID: %s\n", resp.SessionID) } return cws.EmptyPacket diff --git a/web/fonts/6809 chargen.ttf b/web/fonts/6809 chargen.ttf old mode 100755 new mode 100644 diff --git a/web/game.html b/web/game.html index f0198376..d5a2f1fa 100644 --- a/web/game.html +++ b/web/game.html @@ -80,6 +80,7 @@ Fork me on GitHub +