From 401940b7172bdf10e4e179b4e8efffd5b9bd3fcb Mon Sep 17 00:00:00 2001 From: giongto35 Date: Sat, 4 May 2019 01:52:05 +0800 Subject: [PATCH] Remove session after rejoin few times --- cmd/main.go | 25 +++++++++------ emulator/gameview.go | 2 +- handler/browser.go | 2 ++ handler/cloud-storage/storage.go | 55 ++++++++++++++++++++++++++++++++ handler/handlers.go | 10 ++++++ handler/overlord.go | 1 + handler/room/media.go | 4 +-- handler/room/room.go | 21 +++++++----- webrtc/webrtc.go | 5 +++ 9 files changed, 105 insertions(+), 20 deletions(-) create mode 100644 handler/cloud-storage/storage.go diff --git a/cmd/main.go b/cmd/main.go index 57ab7fca..c1974ff4 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -3,9 +3,11 @@ package main import ( "flag" "log" + "math/rand" "net/http" _ "net/http/pprof" "strings" + "time" "github.com/giongto35/cloud-game/config" "github.com/giongto35/cloud-game/handler" @@ -22,6 +24,15 @@ const ( // Time allowed to write a message to the peer. var upgrader = websocket.Upgrader{} +func createOverlordConnection() (*websocket.Conn, error) { + c, _, err := websocket.DefaultDialer.Dial(*config.OverlordHost, nil) + if err != nil { + return nil, err + } + + return c, nil +} + // initilizeOverlord setup an overlord server func initilizeOverlord() { overlord := overlord.NewServer() @@ -33,15 +44,6 @@ func initilizeOverlord() { http.ListenAndServe(":9000", nil) } -func createOverlordConnection() (*websocket.Conn, error) { - c, _, err := websocket.DefaultDialer.Dial(*config.OverlordHost, nil) - if err != nil { - return nil, err - } - - return c, nil -} - // initializeServer setup a server func initializeServer() { conn, err := createOverlordConnection() @@ -67,6 +69,11 @@ func main() { flag.Parse() log.Println("Usage: ./game [-debug]") + rand.Seed(time.Now().UTC().UnixNano()) + + // There are two server mode + // Overlord is coordinator. If the OvelordHost Param is `overlord`, we spawn a new host as Overlord. + // else we spawn new server as normal server connecting to OverlordHost. if *config.OverlordHost == "overlord" { log.Println("Running as overlord ") initilizeOverlord() diff --git a/emulator/gameview.go b/emulator/gameview.go index de429f2b..a6100d62 100644 --- a/emulator/gameview.go +++ b/emulator/gameview.go @@ -94,7 +94,7 @@ func (view *GameView) Enter() { view.console.SetAudioSampleRate(SampleRate) view.console.SetAudioChannel(view.audioChannel) - // load state + // load state if the hash file existed (Join the old room) if err := view.console.LoadState(savePath(view.hash)); err == nil { return } else { diff --git a/handler/browser.go b/handler/browser.go index e0f701de..d591aefd 100644 --- a/handler/browser.go +++ b/handler/browser.go @@ -94,11 +94,13 @@ func (s *Session) RegisterBrowserClient() { // Create new room // TODO: check if roomID is in the current server room := s.handler.getRoom(s.RoomID) + log.Println("Got Room from local ", room, " ID: ", s.RoomID) if room == nil { room = s.handler.createNewRoom(s.GameName, s.RoomID, s.PlayerIndex) } // Attach peerconnection to room + s.handler.detachPeerConn(s.peerconnection) room.AddConnectionToRoom(s.peerconnection, s.PlayerIndex) s.RoomID = room.ID diff --git a/handler/cloud-storage/storage.go b/handler/cloud-storage/storage.go new file mode 100644 index 00000000..15afdeed --- /dev/null +++ b/handler/cloud-storage/storage.go @@ -0,0 +1,55 @@ +package storage + +import ( + "context" + "fmt" + "io" + "log" + + "cloud.google.com/go/storage" +) + +type Client struct { + bucket string + gclient storage.Client +} + +func NewClient() *Client { + ctx := context.Background() + + // Sets your Google Cloud Platform project ID. + projectID := "YOUR_PROJECT_ID" + + // Creates a client. + client, err := storage.NewClient(ctx) + if err != nil { + log.Fatalf("Failed to create client: %v", err) + } + + // Sets the name for the new bucket. + bucketName := "my-new-bucket" + + // Creates a Bucket instance. + bucket := client.Bucket(bucketName) + + // Creates the new bucket. + if err := bucket.Create(ctx, projectID, nil); err != nil { + log.Fatalf("Failed to create bucket: %v", err) + } + + fmt.Printf("Bucket %v created.\n", bucketName) +} + +func (c *Client) SaveFile(name string, data string) (err error) { + wc := c.gclient.Bucket(c.bucket).Object(name).NewWriter(nil) + if _, err = io.Copy(wc, f); err != nil { + return err + } + if err := wc.Close(); err != nil { + return err + } +} + +func (h *Helper) LoadFile(name string) []byte { + +} diff --git a/handler/handlers.go b/handler/handlers.go index 675fdb2b..60d57e90 100644 --- a/handler/handlers.go +++ b/handler/handlers.go @@ -105,6 +105,16 @@ func (h *Handler) WS(w http.ResponseWriter, r *http.Request) { wssession.BrowserClient.Listen() } +// Detach peerconnection detach/remove a peerconnection from current room +func (h *Handler) detachPeerConn(pc *webrtc.WebRTC) { + roomID := pc.RoomID + room := h.getRoom(roomID) + if room == nil { + return + } + room.CleanSession(pc) +} + // getRoom returns room from roomID func (h *Handler) getRoom(roomID string) *room.Room { room, ok := h.rooms[roomID] diff --git a/handler/overlord.go b/handler/overlord.go index 0ed7c4c5..2e1ef59e 100644 --- a/handler/overlord.go +++ b/handler/overlord.go @@ -86,6 +86,7 @@ func (s *Session) RegisterOverlordClient() { log.Println("Room not found ", s.RoomID) return cws.EmptyPacket } + s.handler.detachPeerConn(s.peerconnection) room.AddConnectionToRoom(peerconnection, s.PlayerIndex) //roomID, isNewRoom := startSession(peerconnection, resp.Data, resp.RoomID, resp.PlayerIndex) log.Println("Done, sending back") diff --git a/handler/room/media.go b/handler/room/media.go index 8c9b79cf..3a3581ab 100644 --- a/handler/room/media.go +++ b/handler/room/media.go @@ -28,7 +28,7 @@ func (r *Room) startAudio() { for { select { case <-r.Done: - r.remove() + r.Close() return case sample := <-r.audioChannel: pcm[idx] = sample @@ -75,7 +75,7 @@ func (r *Room) startVideo() { for { select { case <-r.Done: - r.remove() + r.Close() return case image := <-r.imageChannel: diff --git a/handler/room/room.go b/handler/room/room.go index 4a33368a..78067e8f 100644 --- a/handler/room/room.go +++ b/handler/room/room.go @@ -1,6 +1,7 @@ package room import ( + "fmt" "image" "log" "math/rand" @@ -28,7 +29,7 @@ type Room struct { director *emulator.Director } -// init initilizes a room returns roomID +// NewRoom creates a new room func NewRoom(roomID, gamepath, gameName string) *Room { // if no roomID is given, generate it if roomID == "" { @@ -64,22 +65,21 @@ func NewRoom(roomID, gamepath, gameName string) *Room { // generateRoomID generate a unique room ID containing 16 digits func generateRoomID() string { roomID := strconv.FormatInt(rand.Int63(), 16) + log.Println("Generate Room ID", roomID) //roomID := uuid.Must(uuid.NewV4()).String() return roomID } func (r *Room) AddConnectionToRoom(peerconnection *webrtc.WebRTC, playerIndex int) { - r.cleanSession(peerconnection) peerconnection.AttachRoomID(r.ID) - go r.startWebRTCSession(peerconnection, playerIndex) - r.rtcSessions = append(r.rtcSessions, peerconnection) + + go r.startWebRTCSession(peerconnection, playerIndex) } // startWebRTCSession fan-in of the same room to inputChannel func (r *Room) startWebRTCSession(peerconnection *webrtc.WebRTC, playerIndex int) { inputChannel := r.inputChannel - log.Println("room, inputChannel", r, inputChannel) for { select { case <-peerconnection.Done: @@ -103,20 +103,25 @@ func (r *Room) startWebRTCSession(peerconnection *webrtc.WebRTC, playerIndex int } } -func (r *Room) cleanSession(peerconnection *webrtc.WebRTC) { +func (r *Room) CleanSession(peerconnection *webrtc.WebRTC) { r.removeSession(peerconnection) // TODO: Clean all channels } func (r *Room) removeSession(w *webrtc.WebRTC) { + fmt.Println("Cleaning session: ", w) r.sessionsLock.Lock() defer r.sessionsLock.Unlock() + fmt.Println("Sessions list", r.rtcSessions) for i, s := range r.rtcSessions { - if s == w { + fmt.Println("found session: ", s, w) + if s.ID == w.ID { r.rtcSessions = append(r.rtcSessions[:i], r.rtcSessions[i+1:]...) + fmt.Println("found session: ", len(r.rtcSessions)) // If room has no sessions, close room if len(r.rtcSessions) == 0 { + log.Println("No session in room") r.Done <- struct{}{} } break @@ -124,7 +129,7 @@ func (r *Room) removeSession(w *webrtc.WebRTC) { } } -func (r *Room) remove() { +func (r *Room) Close() { log.Println("Closing room", r) r.director.Done <- struct{}{} } diff --git a/webrtc/webrtc.go b/webrtc/webrtc.go index 1005c03a..a614ec55 100644 --- a/webrtc/webrtc.go +++ b/webrtc/webrtc.go @@ -16,6 +16,7 @@ import ( vpxEncoder "github.com/giongto35/cloud-game/vpx-encoder" "github.com/pion/webrtc" "github.com/pion/webrtc/pkg/media" + uuid "github.com/satori/go.uuid" ) var webrtcconfig = webrtc.Configuration{ICEServers: []webrtc.ICEServer{{URLs: []string{"stun:stun.l.google.com:19302"}}}} @@ -94,6 +95,8 @@ func Decode(in string, obj interface{}) { // NewWebRTC create func NewWebRTC() *WebRTC { w := &WebRTC{ + ID: uuid.Must(uuid.NewV4()).String(), + ImageChannel: make(chan []byte, 2), AudioChannel: make(chan []byte, 1000), InputChannel: make(chan int, 2), @@ -108,6 +111,8 @@ type InputDataPair struct { // WebRTC connection type WebRTC struct { + ID string + connection *webrtc.PeerConnection encoder *vpxEncoder.VpxEncoder isConnected bool