From deff2bb7354bf9dc0830342ff331ab87865ef6ee Mon Sep 17 00:00:00 2001 From: giongto35 Date: Sat, 20 Apr 2019 01:03:19 +0800 Subject: [PATCH] refactor with callback --- main.go | 362 ++++++++++++++++++++++++++++++------------- overlord.go | 28 ++++ overlord/overlord.go | 1 + static/js/ws.js | 12 +- ws.go | 93 +++++++++++ 5 files changed, 391 insertions(+), 105 deletions(-) create mode 100644 overlord.go create mode 100644 overlord/overlord.go create mode 100644 ws.go diff --git a/main.go b/main.go index 07ebb72f..57160d18 100644 --- a/main.go +++ b/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 { @@ -68,6 +62,11 @@ func main() { indexFN = debugIndex fmt.Println("Use debug version") } + if len(os.Args) == 3 { + if os.Args[3] == "overlord" { + IsOverlord = true + } + } rand.Seed(time.Now().UTC().UnixNano()) fmt.Println("http://localhost:8000") @@ -164,116 +163,261 @@ func ws(w http.ResponseWriter, r *http.Request) { } 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() + client := NewClient(c, webrtc.NewWebRTC()) + //&Client{ + //conn: c, + ////wsOverlord: createOverlordClient(), + //peerconnection: webrtc.NewWebRTC(), + //} + + client.syncReceive("initwebrtc", func(req WSPacket) WSPacket { + log.Println("Received user SDP") + localSession, err := client.peerconnection.StartClient(req.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.syncReceive("save", func(req WSPacket) (res WSPacket) { + 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" } - // 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 res + }) + + client.syncReceive("load", func(req WSPacket) (res WSPacket) { + 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" + } + + return res + }) + + client.syncReceive("start", func(req WSPacket) (res WSPacket) { + 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(client.peerconnection, gameName, roomID, playerIndex) + res.ID = "start" + res.RoomID = roomID - case "initwebrtc": - log.Println("Received user SDP") - localSession, err := webRTC.StartClient(req.Data, width, height) - if err != nil { - log.Fatalln(err) - } + return res + }) - 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" - } - } - - stRes, err := json.Marshal(res) + client.syncReceive("candidate", func(req WSPacket) (res WSPacket) { + // Unuse code + hi := pionRTC.ICECandidateInit{} + err = json.Unmarshal([]byte(req.Data), &hi) if err != nil { - log.Println("json marshal:", err) + log.Println("[!] Cannot parse candidate: ", err) + } else { + // webRTC.AddCandidate(hi) } + res.ID = "candidate" - c.SetWriteDeadline(time.Now().Add(writeWait)) - err = c.WriteMessage(mt, []byte(stRes)) - } + return res + }) + + client.listen() } +//func wsold(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() + +//client := &Client{ +//ws: c, +//wsOverlord: createOverlordClient(), +//peerconnection: webrtc.NewWebRTC(), +//} + +//if IsOverlord { + +//} + +//log.Println("New ws connection") + +//// 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() +//if err != nil { +//log.Println("[!] read:", err) +//break +//} + +//req := WSPacket{} +//res := WSPacket{} + +//err = json.Unmarshal(message, &req) +//if err != nil { +//log.Println("[!] json unmarshal:", err) +//break +//} + +//// 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 +////log.Println("Ping from server with game:", gameName) +////res.ID = "pong" + +//case "initwebrtc": +//log.Println("Received user SDP") +//localSession, err := client.peerconnection.StartClient(req.Data, width, height) +//if err != nil { +//log.Fatalln(err) +//} + +//res.ID = "sdp" +//res.Data = localSession + +//// Master received request Offer +////case "requestOffer": +//// Master ask the host connecting to master to request offer from browser + +////case "receivedRemoteSDP": +////log.Println("Received remote rtc") +//// Relay message to overlord, so overlord will send to the correct host +//// TODO: Include room to data +////res.id = "overlordrelaysdp" +////res.Data = req.Data + +////case "receivedRelaySDPFromOverlord": +////// initwebrtc +////log.Println("Received user SDP") +////localSession, err := client.peerconnection.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 "remoteStart": +//gameName = req.Data +//roomID = req.RoomID +//log.Println("Received Remote start", gameName, roomID) + +//roomID = startSession(client.peerconnection, gameName, roomID, playerIndex) +//res.ID = "start" +//res.RoomID = roomID + +//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") +//if overlord.isRemoteRoom(roomID) { +//res.ID = "overlordRequestOffer" +//} else { +//roomID = startSession(client.peerconnnection, 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" +//} +//} + +//stRes, err := json.Marshal(res) +//if err != nil { +//log.Println("json marshal:", err) +//} + +//c.SetWriteDeadline(time.Now().Add(writeWait)) +//if strings.HasPrefix(res.ID, "overlord") { +//overlord.ws.WriteMessage(res.Data) +//} +//c.WriteMessage(res.Data) +//err = c.WriteMessage(mt, []byte(stRes)) +//} +//} + // generateRoomID generate a unique room ID containing 16 digits func generateRoomID() string { roomID := strconv.FormatInt(rand.Int63(), 16) @@ -324,7 +468,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 @@ -344,19 +488,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 } @@ -366,3 +510,13 @@ func removeSession(room *Room, webrtc *webrtc.WebRTC) { room.Done <- struct{}{} } } + +//func (o *Overlord) isRemoteRoom(roomID string) bool { +//err := c.WriteMessage(websocket.TextMessage, []byte(stRes)) +//o.ws.WriteMessage() +//_, message, err := o.ws.ReadMessage() +//if message == "isRemoteRoom" { +//return true +//} +//return false +//} diff --git a/overlord.go b/overlord.go new file mode 100644 index 00000000..72b7f2a3 --- /dev/null +++ b/overlord.go @@ -0,0 +1,28 @@ +package main + +import ( + "github.com/gorilla/websocket" +) + +const overlordHost = "http://localhost:9000" + +type Overlord struct { + ws websocket.Conn +} + +//func createOverlordClient() websocket.Conn { +//signal.Notify(interrupt, os.Interrupt) + +////u := url.URL{Scheme: "ws", Host: *addr, Path: "/echo"} +////log.Printf("connecting to %s", u.String()) + +//c, _, err := websocket.DefaultDialer.Dial(overlordHost, nil) +//if err != nil { +//log.Fatal("dial:", err) +//} +//overlord := &Overlord{ +//ws: c, +//} + +//return overlord +//} diff --git a/overlord/overlord.go b/overlord/overlord.go new file mode 100644 index 00000000..8b137891 --- /dev/null +++ b/overlord/overlord.go @@ -0,0 +1 @@ + diff --git a/static/js/ws.js b/static/js/ws.js index afcf04c3..b50cd19e 100644 --- a/static/js/ws.js +++ b/static/js/ws.js @@ -27,6 +27,16 @@ conn.onmessage = e => { log("Got remote sdp"); pc.setRemoteDescription(new RTCSessionDescription(JSON.parse(atob(d["data"])))); break; + //case "requestOffer": + //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"); @@ -51,7 +61,7 @@ conn.onmessage = e => { function sendPing() { // TODO: format the package with time - conn.send(JSON.stringify({"id": "pingpong", "data": "pingpong"})); + //conn.send(JSON.stringify({"id": "pingpong", "data": "pingpong"})); } function startWebRTC() { diff --git a/ws.go b/ws.go new file mode 100644 index 00000000..e971c410 --- /dev/null +++ b/ws.go @@ -0,0 +1,93 @@ +package main + +import ( + "encoding/json" + "log" + "time" + + "github.com/giongto35/cloud-game/webrtc" + "github.com/gorilla/websocket" +) + +type Client struct { + conn *websocket.Conn + wsoverlord *websocket.Conn + peerconnection *webrtc.WebRTC + + // 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 +} + +func NewClient(conn *websocket.Conn, webrtc *webrtc.WebRTC) *Client { + sendCallback := map[string]func(WSPacket){} + recvCallback := map[string]func(WSPacket){} + return &Client{ + conn: conn, + peerconnection: webrtc, + sendCallback: sendCallback, + recvCallback: recvCallback, + } +} + +// syncSend sends a packet and trigger callback when the packet comes back +func (c *Client) syncSend(packet WSPacket, callback func(msg WSPacket)) { + data, err := json.Marshal(packet) + if err != nil { + return + } + + c.conn.WriteMessage(0, data) + c.sendCallback[packet.PacketID] = callback +} + +// syncReceive receive and response back +func (c *Client) syncReceive(id string, f func(request WSPacket) (response WSPacket)) { + c.recvCallback[id] = func(request WSPacket) { + packet := f(request) + + 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) + } +} + +func (c *Client) listen() { + for { + _, 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) + } + // Check if some receiver with the ID is registered + if callback, ok := c.recvCallback[wspacket.ID]; ok { + callback(wspacket) + } + } +}