From d9febbf0a98f690b21e58323134a07cf54d00eb9 Mon Sep 17 00:00:00 2001 From: giongto35 Date: Thu, 2 May 2019 03:01:58 +0800 Subject: [PATCH] Fix websocket --- cmd/main_test.go | 170 +++++++++++++++++++++++++++++++++++++------ cws/cws.go | 37 +++++++--- handler/browser.go | 11 ++- handler/handlers.go | 31 ++++---- handler/overlord.go | 3 +- overlord/overlord.go | 18 ++--- webrtc/webrtc.go | 15 ++-- 7 files changed, 212 insertions(+), 73 deletions(-) diff --git a/cmd/main_test.go b/cmd/main_test.go index 1ae03a1f..6f43b944 100644 --- a/cmd/main_test.go +++ b/cmd/main_test.go @@ -2,6 +2,7 @@ package main import ( "fmt" + "log" "net/http" "net/http/httptest" "strings" @@ -51,7 +52,7 @@ func connectTestOverlordServer(t *testing.T, overlordURL string) *websocket.Conn return oconn } -func initClient(t *testing.T, host string) { +func initClient(t *testing.T, host string) (conn *websocket.Conn, roomID chan string) { // Convert http://127.0.0.1 to ws://127.0.0. u := "ws" + strings.TrimPrefix(host, "http") @@ -61,10 +62,8 @@ func initClient(t *testing.T, host string) { if err != nil { t.Fatalf("%v", err) } - defer ws.Close() // Simulate peerconnection initialization from client - fmt.Println("Simulating PeerConnection") peerConnection, err := webrtc.NewPeerConnection(webrtcconfig) if err != nil { @@ -83,6 +82,7 @@ func initClient(t *testing.T, host string) { } // Send offer to server + log.Println("Browser Client") client := cws.NewClient(ws) go client.Listen() @@ -109,7 +109,7 @@ func initClient(t *testing.T, host string) { time.Sleep(time.Second * 3) fmt.Println("Sending start...") - roomID := make(chan string) + roomID = make(chan string) client.Send(cws.WSPacket{ ID: "start", Data: "Contra.nes", @@ -120,13 +120,7 @@ func initClient(t *testing.T, host string) { roomID <- resp.RoomID }) - respRoomID := <-roomID - if respRoomID == "" { - fmt.Println("RoomID should not be empty") - t.Fail() - } - fmt.Println("Done") - ws.Close() + return ws, roomID // If receive roomID, the server is running correctly } @@ -135,7 +129,16 @@ func TestSingleServerNoOverlord(t *testing.T) { s := initServer(t, nil) defer s.Close() - initClient(t, s.URL) + conn, roomID := initClient(t, s.URL) + defer conn.Close() + + respRoomID := <-roomID + if respRoomID == "" { + fmt.Println("RoomID should not be empty") + t.Fail() + } + fmt.Println("Done") + conn.Close() } func TestSingleServerOneOverlord(t *testing.T) { @@ -143,22 +146,141 @@ func TestSingleServerOneOverlord(t *testing.T) { defer o.Close() oconn := connectTestOverlordServer(t, o.URL) + defer oconn.Close() // Init slave server s := initServer(t, oconn) defer s.Close() - initClient(t, s.URL) + conn, roomID := initClient(t, s.URL) + respRoomID := <-roomID + if respRoomID == "" { + fmt.Println("RoomID should not be empty") + t.Fail() + } + fmt.Println("Done") + 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() +func TestTwoServerOneOverlord(t *testing.T) { + o := initOverlord() + defer o.Close() -//initClient(t, s1.URL) -//oclient.conn.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() + + conn1, roomID := initClient(t, s1.URL) + respRoomID := <-roomID + if respRoomID == "" { + fmt.Println("RoomID should not be empty") + t.Fail() + } + fmt.Println("Done create a room in server 1") + + fmt.Println("Request the room from server 2", respRoomID) + conn2, roomID := initClient2(t, s2.URL, respRoomID) + respRoomID = <-roomID + if respRoomID == "" { + fmt.Println("RoomID should not be empty") + t.Fail() + } + + fmt.Println("Done") + conn1.Close() + conn2.Close() +} + +func initClient2(t *testing.T, host string, remoteRoomID string) (conn *websocket.Conn, roomID chan string) { + // 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) + } + //defer ws.Close() + + // 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 + }) + + time.Sleep(time.Second * 3) + fmt.Println("Sending start...") + + // Doing the same create local room. + localRoomID := make(chan string) + client.Send(cws.WSPacket{ + ID: "start", + Data: "Contra.nes", + RoomID: "", + PlayerIndex: 1, + }, func(resp cws.WSPacket) { + fmt.Println("RoomID:", resp.RoomID) + localRoomID <- resp.RoomID + }) + + <-localRoomID + + log.Println("Server2 trying to join server1 room") + // After trying loging in to one session, login to other with the roomID + client.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 + return ws, roomID +} diff --git a/cws/cws.go b/cws/cws.go index b943ad55..f4a821de 100644 --- a/cws/cws.go +++ b/cws/cws.go @@ -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, @@ -56,20 +61,23 @@ 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 @@ -79,6 +87,7 @@ func (c *Client) Receive(id string, f func(response WSPacket) (request WSPacket) // 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 { @@ -125,6 +134,8 @@ func (c *Client) Listen() { } wspacket := WSPacket{} err = json.Unmarshal(rawMsg, &wspacket) + //log.Println( "Received: ", wspacket) + if err != nil { continue } @@ -132,16 +143,20 @@ 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) diff --git a/handler/browser.go b/handler/browser.go index 4d4fbde3..7b7817c5 100644 --- a/handler/browser.go +++ b/handler/browser.go @@ -78,22 +78,29 @@ func (s *Session) RegisterBrowserClient() { s.PlayerIndex = resp.PlayerIndex log.Println("Starting game") - // If we are connecting to overlord, request serverID from roomID + // 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 ", 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.createNewRoom(s.GameName, s.RoomID, s.PlayerIndex) + // Attach peerconnection to room room.addConnectionToRoom(s.peerconnection, s.PlayerIndex) s.RoomID = room.ID // Register room to overlord if we are connecting to overlord + log.Println("Try Registering room", room, "Client: ", s.OverlordClient) if room != nil && s.OverlordClient != nil { + log.Println("Registering room", s.RoomID) s.OverlordClient.Send(cws.WSPacket{ ID: "registerRoom", Data: s.RoomID, diff --git a/handler/handlers.go b/handler/handlers.go index 7e0de6cf..ed50effe 100644 --- a/handler/handlers.go +++ b/handler/handlers.go @@ -5,7 +5,6 @@ import ( "io/ioutil" "log" "net/http" - "time" "github.com/giongto35/cloud-game/cws" "github.com/giongto35/cloud-game/handler/gamelist" @@ -19,30 +18,28 @@ const ( debugIndex = "./static/index_ws.html" ) -// Time allowed to write a message to the peer. -var readWait = 30 * time.Second -var writeWait = 30 * time.Second - // Flag to determine if the server is overlord or not var upgrader = websocket.Upgrader{} type Handler struct { - oClient *OverlordClient - rooms map[string]*Room + // Client that connects to overlord + oClient *OverlordClient + // Rooms map : RoomID -> Room + rooms map[string]*Room + // ID of the current server globalwise serverID string // isDebug determines the mode handler is running - isDebug bool - isOverlord bool - gamePath string - - // ID to peerconnection + isDebug bool + // Path to game list + gamePath string + // All webrtc peerconnections are handled by the server + // ID -> peerconnections peerconnections map[string]*webrtc.WebRTC - // Session - wssession Session } // 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{}, @@ -90,15 +87,15 @@ func (h *Handler) WS(w http.ResponseWriter, r *http.Request) { peerconnection: webrtc.NewWebRTC(), handler: h, } - wssession.RegisterBrowserClient() - fmt.Println("oclient : ", h.oClient) - 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), diff --git a/handler/overlord.go b/handler/overlord.go index c6fd13c3..08de552e 100644 --- a/handler/overlord.go +++ b/handler/overlord.go @@ -79,6 +79,7 @@ func (s *Session) RegisterOverlordClient() { 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) @@ -109,7 +110,7 @@ func getServerIDOfRoom(oc *OverlordClient, roomID string) string { Data: roomID, }, ) - log.Println("Received roomID from overlord") + log.Println("Received roomID from overlord ", packet.Data) return packet.Data } diff --git a/overlord/overlord.go b/overlord/overlord.go index b1e459a2..c73527a7 100644 --- a/overlord/overlord.go +++ b/overlord/overlord.go @@ -31,14 +31,14 @@ 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("[!] WS upgrade:", err) + log.Print("Overlord: [!] WS upgrade:", err) return } defer c.Close() // Register new server serverID := strconv.Itoa(rand.Int()) - log.Println("A new server connected ", serverID) + log.Println("Overlord: A new server connected to Overlord", serverID) // Register to servers map the client connection client := cws.NewClient(c) @@ -62,7 +62,7 @@ func (o *Server) WSO(w http.ResponseWriter, r *http.Request) { // 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("Received registerRoom ", resp.Data, serverID) + log.Println("Overlord: Received registerRoom ", resp.Data, serverID) o.roomToServer[resp.Data] = serverID return cws.WSPacket{ ID: "registerRoom", @@ -71,7 +71,7 @@ func (o *Server) WSO(w http.ResponseWriter, r *http.Request) { // getRoom returns the server ID based on requested roomID. client.Receive("getRoom", func(resp cws.WSPacket) cws.WSPacket { - log.Println("Received a getroom request") + log.Println("Overlord: Received a getroom request") return cws.WSPacket{ ID: "getRoom", Data: o.roomToServer[resp.Data], @@ -81,10 +81,10 @@ func (o *Server) WSO(w http.ResponseWriter, r *http.Request) { // Relay message from server to other target server // TODO: Generalize client.Receive("initwebrtc", func(resp cws.WSPacket) cws.WSPacket { - log.Println("Received a relay sdp request from a host") + log.Println("Overlord: Received a relay sdp request from a host") // TODO: Abstract if resp.TargetHostID != serverID { - log.Println("sending relay sdp to target host", resp) + 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( @@ -93,7 +93,7 @@ func (o *Server) WSO(w http.ResponseWriter, r *http.Request) { return sdp } - log.Println("Target host is overlord itself: start peerconnection") + 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) @@ -111,7 +111,7 @@ func (o *Server) WSO(w http.ResponseWriter, r *http.Request) { // TODO: use relay ID type // TODO: Merge sdp and start client.Receive("start", func(resp cws.WSPacket) cws.WSPacket { - log.Println("Received a relay start request from a host") + 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 @@ -122,7 +122,7 @@ func (o *Server) WSO(w http.ResponseWriter, r *http.Request) { return resp } - log.Println("Target host is overlord itself: start game") + 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) diff --git a/webrtc/webrtc.go b/webrtc/webrtc.go index 2b52c4a1..1005c03a 100644 --- a/webrtc/webrtc.go +++ b/webrtc/webrtc.go @@ -167,20 +167,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 +202,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 +224,6 @@ func (w *WebRTC) StartClient(remoteSession string, width, height int) (string, e log.Println(iceCandidate) }) - - offer := webrtc.SessionDescription{} Decode(remoteSession, &offer)