Merge pull request #15 from giongto35/access-overlord

Access overlord

main.go

@@ -36,15 +36,9 @@ var indexFN = gameboyIndex
 var readWait = 30 * time.Second
 var writeWait = 30 * time.Second
 
+var IsOverlord = false
 var upgrader = websocket.Upgrader{}
 
-type WSPacket struct {
-	ID          string `json:"id"`
-	Data        string `json:"data"`
-	RoomID      string `json:"room_id"`
-	PlayerIndex int    `json:"player_index"`
-}
-
 // 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 {
@@ -60,6 +54,7 @@ type Room struct {
 }
 
 var rooms = map[string]*Room{}
+var port string = "8000"
 
 func main() {
 	fmt.Println("Usage: ./game [debug]")
@@ -68,18 +63,35 @@ func main() {
 		indexFN = debugIndex
 		fmt.Println("Use debug version")
 	}
+	if len(os.Args) >= 3 {
+		if os.Args[2] == "overlord" {
+			IsOverlord = true
+		}
+		fmt.Println("Running as overlord ")
+	}
+	if len(os.Args) >= 4 {
+		port = os.Args[3]
+	}
 
 	rand.Seed(time.Now().UTC().UnixNano())
-	fmt.Println("http://localhost:8000")
 	rooms = map[string]*Room{}
 
 	// ignore origin
 	upgrader.CheckOrigin = func(r *http.Request) bool { return true }
-	http.HandleFunc("/ws", ws)
 
 	http.HandleFunc("/", getWeb)
 	http.Handle("/static/", http.StripPrefix("/static/", http.FileServer(http.Dir("./static"))))
-	http.ListenAndServe(":8000", nil)
+	http.HandleFunc("/ws", ws)
+
+	if !IsOverlord {
+		fmt.Println("http://localhost:8000")
+		http.ListenAndServe(":"+port, nil)
+	} else {
+		fmt.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) {
@@ -137,13 +149,15 @@ func isRoomRunning(roomID string) bool {
 }
 
 // startSession handles one session call
-func startSession(webRTC *webrtc.WebRTC, gameName string, roomID string, playerIndex int) string {
+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
@@ -153,9 +167,18 @@ func startSession(webRTC *webrtc.WebRTC, gameName string, roomID string, playerI
 	webRTC.AttachRoomID(roomID)
 	go startWebRTCSession(room, webRTC, playerIndex)
 
-	return roomID
+	return roomID, isNewRoom
 }
 
+// Session represents a session connected from the browser to the current server
+type Session struct {
+	client         *Client
+	oclient        *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 {
@@ -163,119 +186,114 @@ func ws(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	defer c.Close()
-
-	log.Println("New ws connection")
-	webRTC := webrtc.NewWebRTC()
-
-	// streaming game
-
-	// start new games and webrtc stuff?
-	//isDone := false
-
 	var gameName string
 	var roomID string
 	var playerIndex int
 
-	for {
-		c.SetReadDeadline(time.Now().Add(readWait))
-		mt, message, err := c.ReadMessage()
+	// Create connection to overlord
+	client := NewClient(c)
+
+	wssession := &Session{
+		client:         client,
+		peerconnection: webrtc.NewWebRTC(),
+		// The server session is maintaining
+	}
+
+	if !IsOverlord {
+		wssession.NewOverlordClient()
+	}
+
+	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.Println("[!] read:", err)
-			break
+			log.Fatalln(err)
 		}
 
-		req := WSPacket{}
-		res := WSPacket{}
+		return WSPacket{
+			ID:   "sdp",
+			Data: localSession,
+		}
+	})
 
-		err = json.Unmarshal(message, &req)
-		if err != nil {
-			log.Println("[!] json unmarshal:", err)
-			break
+	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"
 		}
 
-		// SDP connection initializations follows WebRTC convention
-		// https://developer.mozilla.org/en-US/docs/Web/API/WebRTC_API/Protocols
-		switch req.ID {
-		//case "ping":
-		//gameName = req.Data
-		//roomID = req.RoomID
-		//playerIndex = req.PlayerIndex
+		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("Ping from server with game:", gameName)
 		//res.ID = "pong"
-
-		case "pingpong":
-			res.ID = "pingpong"
-			res.Data = req.Data
-
-		case "initwebrtc":
-			log.Println("Received user SDP")
-			localSession, err := webRTC.StartClient(req.Data, width, height)
-			if err != nil {
-				log.Fatalln(err)
-			}
-
-			res.ID = "sdp"
-			res.Data = localSession
-
-		case "candidate":
-			// Unuse code
-			hi := pionRTC.ICECandidateInit{}
-			err = json.Unmarshal([]byte(req.Data), &hi)
-			if err != nil {
-				log.Println("[!] Cannot parse candidate: ", err)
-			} else {
-				// webRTC.AddCandidate(hi)
-			}
-			res.ID = "candidate"
-
-		case "start":
-			gameName = req.Data
-			roomID = req.RoomID
-			playerIndex = req.PlayerIndex
-			//log.Println("Ping from server with game:", gameName)
-			//res.ID = "pong"
-			log.Println("Starting game")
-			roomID = startSession(webRTC, gameName, roomID, playerIndex)
-			res.ID = "start"
-			res.RoomID = roomID
-
-		case "save":
-			log.Println("Saving game state")
-			res.ID = "save"
-			res.Data = "ok"
-			if roomID != "" {
-				err = rooms[roomID].director.SaveGame()
-				if err != nil {
-					log.Println("[!] Cannot save game state: ", err)
-					res.Data = "error"
-				}
-			} else {
-				res.Data = "error"
-			}
-
-		case "load":
-			log.Println("Loading game state")
-			res.ID = "load"
-			res.Data = "ok"
-			if roomID != "" {
-				err = rooms[roomID].director.LoadGame()
-				if err != nil {
-					log.Println("[!] Cannot load game state: ", err)
-					res.Data = "error"
-				}
-			} else {
-				res.Data = "error"
-			}
+		log.Println("Starting game")
+		roomServerID := getServerIDOfRoom(wssession.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
 		}
 
-		stRes, err := json.Marshal(res)
+		roomID, isNewRoom = startSession(wssession.peerconnection, gameName, roomID, playerIndex)
+		if isNewRoom {
+			wssession.oclient.send(WSPacket{
+				ID:   "registerRoom",
+				Data: roomID,
+			}, nil)
+		}
+		req.ID = "start"
+		req.RoomID = roomID
+
+		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("json marshal:", err)
+			log.Println("[!] Cannot parse candidate: ", err)
+		} else {
+			// webRTC.AddCandidate(hi)
 		}
+		req.ID = "candidate"
 
-		c.SetWriteDeadline(time.Now().Add(writeWait))
-		err = c.WriteMessage(mt, []byte(stRes))
-	}
+		return req
+	})
+
+	client.listen()
 }
 
 // generateRoomID generate a unique room ID containing 16 digits
@@ -328,7 +346,7 @@ func startWebRTCSession(room *Room, webRTC *webrtc.WebRTC, playerIndex int) {
 		select {
 		case <-webRTC.Done:
 			fmt.Println("One session closed")
-			removeSession(room, webRTC)
+			removeSession(webRTC, room)
 		default:
 		}
 		// Client stopped
@@ -348,19 +366,19 @@ func startWebRTCSession(room *Room, webRTC *webrtc.WebRTC, playerIndex int) {
 	}
 }
 
-func cleanSession(webrtc *webrtc.WebRTC) {
-	room, ok := rooms[webrtc.RoomID]
+func cleanSession(w *webrtc.WebRTC) {
+	room, ok := rooms[w.RoomID]
 	if !ok {
 		return
 	}
-	removeSession(room, webrtc)
+	removeSession(w, room)
 }
 
-func removeSession(room *Room, webrtc *webrtc.WebRTC) {
+func removeSession(w *webrtc.WebRTC, room *Room) {
 	room.sessionsLock.Lock()
 	defer room.sessionsLock.Unlock()
 	for i, s := range room.rtcSessions {
-		if s == webrtc {
+		if s == w {
 			room.rtcSessions = append(room.rtcSessions[:i], room.rtcSessions[i+1:]...)
 			break
 		}
@@ -370,3 +388,138 @@ func removeSession(room *Room, webrtc *webrtc.WebRTC) {
 		room.Done <- struct{}{}
 	}
 }
+
+func getServerIDOfRoom(oc *Client, roomID string) string {
+	packet := oc.syncSend(
+		WSPacket{
+			ID:   "getRoom",
+			Data: roomID,
+		},
+	)
+
+	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
+	oclient := session.oclient
+	// 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")
+}
+
+const overlordHost = "ws://localhost:9000/wso"
+
+func createOverlordConnection() (*websocket.Conn, error) {
+	c, _, err := websocket.DefaultDialer.Dial(overlordHost, nil)
+	if err != nil {
+		log.Fatal("dial:", err)
+		return nil, err
+	}
+
+	return c, nil
+}
+
+func (s *Session) NewOverlordClient() {
+	oc, err := createOverlordConnection()
+	if err != nil {
+		log.Println("Cannot connect to overlord")
+	}
+	oclient := NewClient(oc)
+	oclient.send(
+		WSPacket{
+			ID: "ping",
+		},
+		func(resp WSPacket) {
+			log.Println("Received pong full flow")
+		},
+	)
+
+	// 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)
+			s.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")
+			localSession, err := s.peerconnection.StartClient(resp.Data, width, height)
+			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")
+
+			roomID, isNewRoom := startSession(s.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")
+			}
+
+			req.ID = "start"
+			req.RoomID = roomID
+			return req
+		},
+	)
+	go oclient.listen()
+
+	s.oclient = oclient
+	// TODO: return oclient
+
+	return
+}

overlord.go

@@ -0,0 +1,129 @@
+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("ping", func(resp WSPacket) WSPacket {
+		log.Println("received Ping, sending Pong")
+		return WSPacket{
+			ID: "pong",
+		}
+	})
+
+	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 {
+		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()
+}

overlord/overlord.go

@@ -0,0 +1 @@
+

static/js/ws.js

@@ -1,4 +1,5 @@
 var pc;
+var curPacketID = "";
 // web socket
 
 conn = new WebSocket(`ws://${location.host}/ws`);
@@ -26,7 +27,21 @@ conn.onmessage = e => {
     case "sdp":
         log("Got remote sdp");
         pc.setRemoteDescription(new RTCSessionDescription(JSON.parse(atob(d["data"]))));
+        //conn.send(JSON.stringify({"id": "sdpdon", "packet_id": d["packet_id"]}));
         break;
+    case "requestOffer":
+        curPacketID = d["packet_id"];
+        log("Received request offer ", curPacketID)
+        startWebRTC();
+        //pc.createOffer({offerToReceiveVideo: true, offerToReceiveAudio: false}).then(d => {
+            //pc.setLocalDescription(d).catch(log);
+        //})
+
+    //case "sdpremote":
+        //log("Got remote sdp");
+        //pc.setRemoteDescription(new RTCSessionDescription(JSON.parse(atob(d["data"]))));
+        //conn.send(JSON.stringify({"id": "remotestart", "data": GAME_LIST[gameIdx].nes, "room_id": roomID.value, "player_index": parseInt(playerIndex.value, 10)}));inputTimer
+        //break;
     case "pong":
         // TODO: Change name use one session
         log("Recv pong. Start webrtc");
@@ -99,7 +114,8 @@ function startWebRTC() {
             session = btoa(JSON.stringify(pc.localDescription));
             localSessionDescription = session;
             log("Send SDP to remote peer");
-            conn.send(JSON.stringify({"id": "initwebrtc", "data": session}));
+            // TODO: Fix curPacketID
+            conn.send(JSON.stringify({"id": "initwebrtc", "data": session, "packet_id": curPacketID}));
         } else {
             console.log(JSON.stringify(event.candidate));
         }

ws.go

@@ -0,0 +1,117 @@
+package main
+
+import (
+	"encoding/json"
+	"log"
+	"math/rand"
+	"strconv"
+	"time"
+
+	"github.com/gorilla/websocket"
+)
+
+type Client struct {
+	conn *websocket.Conn
+
+	// sendCallback is callback based on packetID
+	sendCallback map[string]func(req WSPacket)
+	// recvCallback is callback when receive based on ID of the packet
+	recvCallback map[string]func(req WSPacket)
+}
+
+type WSPacket struct {
+	ID   string `json:"id"`
+	Data string `json:"data"`
+
+	RoomID      string `json:"room_id"`
+	PlayerIndex int    `json:"player_index"`
+
+	TargetHostID string `json:"target_id"`
+	PacketID     string `json:"packet_id"`
+}
+
+var EmptyPacket = WSPacket{}
+
+func NewClient(conn *websocket.Conn) *Client {
+	sendCallback := map[string]func(WSPacket){}
+	recvCallback := map[string]func(WSPacket){}
+	return &Client{
+		conn: conn,
+
+		sendCallback: sendCallback,
+		recvCallback: recvCallback,
+	}
+}
+
+// send sends a packet and trigger callback when the packet comes back
+func (c *Client) send(request WSPacket, callback func(response WSPacket)) {
+	request.PacketID = strconv.Itoa(rand.Int())
+	data, err := json.Marshal(request)
+	if err != nil {
+		return
+	}
+
+	c.conn.WriteMessage(websocket.TextMessage, data)
+	if callback == nil {
+		return
+	}
+	c.sendCallback[request.PacketID] = callback
+}
+
+// receive receive and response back
+func (c *Client) receive(id string, f func(response WSPacket) (request WSPacket)) {
+	c.recvCallback[id] = func(response WSPacket) {
+		packet := f(response)
+		// Add Meta data
+		packet.PacketID = response.PacketID
+
+		// Skip rqeuest if it is EmptyPacket
+		if packet == EmptyPacket {
+			return
+		}
+		resp, err := json.Marshal(packet)
+		if err != nil {
+			log.Println("[!] json marshal error:", err)
+		}
+		c.conn.SetWriteDeadline(time.Now().Add(writeWait))
+		c.conn.WriteMessage(websocket.TextMessage, resp)
+	}
+}
+
+// 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)
+	return <-res
+}
+
+func (c *Client) listen() {
+	for {
+		log.Println("Waiting for message")
+		_, rawMsg, err := c.conn.ReadMessage()
+		if err != nil {
+			log.Println("[!] read:", err)
+			break
+		}
+		wspacket := WSPacket{}
+		err = json.Unmarshal(rawMsg, &wspacket)
+		if err != nil {
+			continue
+		}
+
+		// Check if some async send is waiting for the response based on packetID
+		if callback, ok := c.sendCallback[wspacket.PacketID]; ok {
+			callback(wspacket)
+			delete(c.sendCallback, wspacket.PacketID)
+			// Skip receiveCallback to avoid duplication
+			continue
+		}
+		// Check if some receiver with the ID is registered
+		if callback, ok := c.recvCallback[wspacket.ID]; ok {
+			callback(wspacket)
+		}
+	}
+}