Add support firefox safari ios (#175)

* change rtc offer/answer direction

* remove overlay log

* merge browser+session, fix log, add browser map

* add candidate transfer on worker + web

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),
-	}
-}

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

pkg/coordinator/session.go

@@ -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
-}

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
+	})
 }

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"`
 }
 

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]
 	}

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

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

web/game.html

@@ -80,6 +80,7 @@
 
     <a id="ribbon" style="position: fixed; right: 0; top: 0;" href="https://github.com/giongto35/cloud-game"><img width="149" height="149" src="https://github.blog/wp-content/uploads/2008/12/forkme_right_gray_6d6d6d.png?resize=149%2C149" class="attachment-full size-full" alt="Fork me on GitHub" data-recalc-dims="1"></a>
 
+
     <script>
         DEBUG = true;
         STUNTURN = {{.STUNTURN}};

web/js/controller.js

@@ -46,6 +46,7 @@
     const onLatencyCheckRequest = (data) => {
         popup('Ping check...');
         const timeoutMs = 2000;
+        // TODO: why we use maximum timeout
         const maxTimeoutMs = timeoutMs > ajax.defaultTimeoutMs() ? timeoutMs : ajax.defaultTimeoutMs();
 
         Promise.all((data.addresses || []).map(address => {
@@ -358,6 +359,8 @@
         gameList.set(data.games);
     });
     event.sub(MEDIA_STREAM_SDP_AVAILABLE, (data) => rtcp.setRemoteDescription(data.sdp, gameScreen[0]));
+    event.sub(MEDIA_STREAM_CANDIDATE_ADD, (data) => rtcp.addCandidate(data.candidate));
+    event.sub(MEDIA_STREAM_CANDIDATE_FLUSH, () => rtcp.flushCandidate());
     event.sub(MEDIA_STREAM_READY, () => rtcp.start());
     event.sub(CONNECTION_READY, onConnectionReady);
     event.sub(CONNECTION_CLOSED, () => input.poll().disable());

web/js/event/event.js

@@ -69,6 +69,8 @@ const CONNECTION_CLOSED = 'connectionClosed';
 
 const MEDIA_STREAM_INITIALIZED = 'mediaStreamInitialized';
 const MEDIA_STREAM_SDP_AVAILABLE = 'mediaStreamSdpAvailable';
+const MEDIA_STREAM_CANDIDATE_ADD = 'mediaStreamCandidateAdd';
+const MEDIA_STREAM_CANDIDATE_FLUSH = 'mediaStreamCandidateFlush';
 const MEDIA_STREAM_READY = 'mediaStreamReady';
 
 const GAMEPAD_CONNECTED = 'gamepadConnected';

web/js/network/rtcp.js

@@ -6,6 +6,9 @@ const rtcp = (() => {
     let connection;
     let inputChannel;
     let mediaStream;
+    let candidates = Array();
+    let isAnswered = false;
+    let isFlushing = false;
 
     let connected = false;
     let inputReady = false;
@@ -20,13 +23,18 @@ const rtcp = (() => {
         mediaStream = new MediaStream();
 
         // input channel, ordered + reliable, id 0
-        inputChannel = connection.createDataChannel('a', {ordered: true, negotiated: true, id: 0,});
-        inputChannel.onopen = () => {
-            log.debug('[rtcp] the input channel has opened');
-            inputReady = true;
-            event.pub(CONNECTION_READY)
-        };
-        inputChannel.onclose = () => log.debug('[rtcp] the input channel has closed');
+        // inputChannel = connection.createDataChannel('a', {ordered: true, negotiated: true, id: 0,});
+        // recv dataChannel from worker
+        connection.ondatachannel = e => {
+            log.debug(`[rtcp] ondatachannel: ${e.channel.label}`)
+            inputChannel = e.channel;
+            inputChannel.onopen = () => {
+                log.debug('[rtcp] the input channel has opened');
+                inputReady = true;
+                event.pub(CONNECTION_READY)
+            };
+            inputChannel.onclose = () => log.debug('[rtcp] the input channel has closed');
+        }
 
         connection.addTransceiver('video', {'direction': 'sendrecv'});
         connection.addTransceiver('audio', {'direction': 'sendrecv'});
@@ -36,11 +44,10 @@ const rtcp = (() => {
         connection.onicecandidate = ice.onIcecandidate;
         connection.ontrack = event => mediaStream.addTrack(event.track);
 
-        connection.createOffer({offerToReceiveVideo: true, offerToReceiveAudio: true})
-            .then(offer => {
-                log.info(offer.sdp);
-                connection.setLocalDescription(offer).catch(log.error);
-            });
+        socket.send({
+            'id': 'initwebrtc',
+            'data': JSON.stringify({'is_mobile': env.isMobileDevice()}),
+        });
     };
 
     const ice = (() => {
@@ -49,22 +56,18 @@ const rtcp = (() => {
 
         const ICE_TIMEOUT = 2000;
 
-        const sendCandidates = () => {
-            if (isGatheringDone) return;
-            const session = btoa(JSON.stringify(connection.localDescription));
-            const data = JSON.stringify({"sdp": session, "is_mobile": env.isMobileDevice()});
-            socket.send({"id": "initwebrtc", "data": data});
-            isGatheringDone = true;
-        };
-
         return {
             onIcecandidate: event => {
-                if (event.candidate && !isGatheringDone) {
-                    log.info(JSON.stringify(event.candidate));
-                } else {
-                    sendCandidates()
+                // this trigger when setRemoteDesc success
+                // send any candidate to worker
+                if (event.candidate != null) {
+                    candidate = JSON.stringify(event.candidate);
+                    log.info(`[rtcp] got ice candidate: ${candidate}`);
+                    socket.send({
+                        'id': 'candidate',
+                        'data': btoa(candidate),
+                    })
                 }
-                // TODO: Fix curPacketID
             },
             onIceStateChange: event => {
                 switch (event.target.iceGatheringState) {
@@ -72,7 +75,7 @@ const rtcp = (() => {
                         log.info('[rtcp] ice gathering');
                         timeForIceGathering = setTimeout(() => {
                             log.info(`[rtcp] ice gathering was aborted due to timeout ${ICE_TIMEOUT}ms`);
-                            sendCandidates();
+                            // sendCandidates();
                         }, ICE_TIMEOUT);
                         break;
                     case 'complete':
@@ -111,12 +114,40 @@ const rtcp = (() => {
 
     return {
         start: start,
-        setRemoteDescription: (data, media) => {
-            connection.setRemoteDescription(new RTCSessionDescription(JSON.parse(atob(data))))
-            // set media object stream
-                .then(() => {
-                    media.srcObject = mediaStream;
-                })
+        setRemoteDescription: async (data, media) => {
+            offer = new RTCSessionDescription(JSON.parse(atob(data)));
+            await connection.setRemoteDescription(offer);
+            
+            answer = await connection.createAnswer({});
+            await connection.setLocalDescription(answer);
+
+            isAnswered = true;
+            event.pub(MEDIA_STREAM_CANDIDATE_FLUSH);
+
+            socket.send({
+                'id': 'answer',
+                'data': btoa(JSON.stringify(answer)),
+            });
+
+            media.srcObject = mediaStream;
+        },
+        addCandidate: (data) => {
+            if (data == '') {
+                event.pub(MEDIA_STREAM_CANDIDATE_FLUSH);
+            } else {
+                candidates.push(data);
+            }
+        },
+        flushCandidate: () => {
+            if (isFlushing || !isAnswered) return;
+            isFlushing = true;
+            candidates.forEach(data => {
+                d = atob(data);
+                candidate = new RTCIceCandidate(JSON.parse(d));
+                log.debug('[rtcp] add candidate: ' + d);
+                connection.addIceCandidate(candidate);
+            });
+            isFlushing = false;
         },
         input: (data) => inputChannel.send(data),
         isConnected: () => connected,

web/js/network/socket.js

@@ -6,7 +6,25 @@
  * @version 1
  */
 const socket = (() => {
-    const pingIntervalMs = 1000 / 5;
+    // TODO: this ping is for maintain websocket state
+    /*
+        https://tools.ietf.org/html/rfc6455#section-5.5.2
+
+        Chrome doesn't support
+            https://groups.google.com/a/chromium.org/forum/#!topic/net-dev/2RAm-ZYAIYY
+            https://bugs.chromium.org/p/chromium/issues/detail?id=706002
+
+        Firefox has option but not enable 'network.websocket.timeout.ping.request'
+
+        Suppose ping message must be sent from WebSocket Server.
+        Gorilla WS doesnot support it.
+        https://github.com/gorilla/websocket/blob/5ed622c449da6d44c3c8329331ff47a9e5844f71/examples/filewatch/main.go#L104
+
+        Below is high level implementation of ping.
+        // TODO: find the best ping time, currently 2 seconds works well in Chrome+Firefox
+    */
+    const pingIntervalMs = 2000; // 2 secs
+    // const pingIntervalMs = 1000 / 5; // too much
 
     let conn;
     let curPacketId = '';
@@ -46,13 +64,12 @@ const socket = (() => {
                     let serverData = JSON.parse(data.data);
                     event.pub(MEDIA_STREAM_INITIALIZED, {stunturn: serverData.shift(), games: serverData});
                     break;
-                case 'sdp':
+                case 'offer':
+                    // this is offer from worker
                     event.pub(MEDIA_STREAM_SDP_AVAILABLE, {sdp: data.data});
                     break;
-                case 'requestOffer':
-                    // !to remove? wtf
-                    curPacketId = data.packet_id;
-                    event.pub(MEDIA_STREAM_READY);
+                case 'candidate':
+                    event.pub(MEDIA_STREAM_CANDIDATE_ADD, {candidate: data.data});
                     break;
                 case 'heartbeat':
                     event.pub(PING_RESPONSE);