update.

cloudwebrtc · cloudwebrtc · commit c8e522252abb · 2020-01-15T14:24:09.000+08:00
diff --git a/Makefile b/Makefile
@@ -1,6 +1,6 @@
 all: linux darwin windows
 
-release: all upx tar
+release: all tar
 
 clean:
 	rm -rf bin/*
diff --git a/go.sum b/go.sum
@@ -5,7 +5,6 @@ github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
 github.com/gorilla/websocket v1.4.1 h1:q7AeDBpnBk8AogcD4DSag/Ukw/KV+YhzLj2bP5HvKCM=
 github.com/gorilla/websocket v1.4.1/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
-github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo=
 github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
 github.com/pion/logging v0.2.2 h1:M9+AIj/+pxNsDfAT64+MAVgJO0rsyLnoJKCqf//DoeY=
 github.com/pion/logging v0.2.2/go.mod h1:k0/tDVsRCX2Mb2ZEmTqNa7CWsQPc+YYCB7Q+5pahoms=
@@ -20,9 +19,7 @@ github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZN
 github.com/rs/xid v1.2.1/go.mod h1:+uKXf+4Djp6Md1KODXJxgGQPKngRmWyn10oCKFzNHOQ=
 github.com/rs/zerolog v1.14.3 h1:4EGfSkR2hJDB0s3oFfrlPqjU1e4WLncergLil3nEKW0=
 github.com/rs/zerolog v1.14.3/go.mod h1:3WXPzbXEEliJ+a6UFE4vhIxV8qR1EML6ngzP9ug4eYg=
-github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d h1:zE9ykElWQ6/NYmHa3jpm/yHnI4xSofP+UP6SpjHcSeM=
 github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
-github.com/smartystreets/goconvey v1.6.4 h1:fv0U8FUIMPNf1L9lnHLvLhgicrIVChEkdzIKYqbNC9s=
 github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
diff --git a/pkg/signaler/signaler.go b/pkg/signaler/signaler.go
@@ -7,8 +7,8 @@ import (
 	"strings"
 
 	"github.com/cloudwebrtc/flutter-webrtc-server/pkg/logger"
-	"github.com/cloudwebrtc/flutter-webrtc-server/pkg/transport"
 	"github.com/cloudwebrtc/flutter-webrtc-server/pkg/turn"
+	"github.com/cloudwebrtc/flutter-webrtc-server/pkg/websocket"
 )
 
 func Marshal(m map[string]interface{}) string {
@@ -38,8 +38,8 @@ type PeerInfo struct {
 
 // Peer .
 type Peer struct {
-	info      PeerInfo
-	transport *transport.WebSocketTransport
+	info PeerInfo
+	conn *websocket.WebSocketConn
 }
 
 // Session info.
@@ -70,7 +70,7 @@ func (s Signaler) authHandler(username string, realm string, srcAddr net.Addr) (
 	return nil, false
 }
 
-func (s *Signaler) NotifyPeersUpdate(transport *transport.WebSocketTransport, peers map[string]Peer) {
+func (s *Signaler) NotifyPeersUpdate(conn *websocket.WebSocketConn, peers map[string]Peer) {
 	infos := []PeerInfo{}
 	for _, peer := range peers {
 		infos = append(infos, peer.info)
@@ -79,33 +79,33 @@ func (s *Signaler) NotifyPeersUpdate(transport *transport.WebSocketTransport, pe
 	request["type"] = "peers"
 	request["data"] = infos
 	for _, peer := range peers {
-		peer.transport.Send(Marshal(request))
+		peer.conn.Send(Marshal(request))
 	}
 }
 
 func (s *Signaler) HandleTurnServerCredentials(writer http.ResponseWriter, request *http.Request) {
 	// return turn credentials for client.
 }
 
-func (s *Signaler) HandleNewWebSocket(transport *transport.WebSocketTransport, request *http.Request) {
+func (s *Signaler) HandleNewWebSocket(conn *websocket.WebSocketConn, request *http.Request) {
 	logger.Infof("On Open %v", request)
-	transport.On("message", func(message []byte) {
+	conn.On("message", func(message []byte) {
 		request := Unmarshal(string(message))
 		data := request["data"].(map[string]interface{})
 		switch request["type"] {
 		case "new":
 			{
 
 				peer := Peer{
-					transport: transport,
+					conn: conn,
 					info: PeerInfo{
 						ID:        data["id"].(string),
 						Name:      data["name"].(string),
 						UserAgent: data["user_agent"].(string),
 					},
 				}
 				s.peers[peer.info.ID] = peer
-				s.NotifyPeersUpdate(transport, s.peers)
+				s.NotifyPeersUpdate(conn, s.peers)
 			}
 			break
 		case "leave":
@@ -128,10 +128,10 @@ func (s *Signaler) HandleNewWebSocket(transport *transport.WebSocketTransport, r
 							"reason":  "Peer [" + To + "] not found ",
 						},
 					}
-					transport.Send(Marshal(msg))
+					conn.Send(Marshal(msg))
 					return
 				} else {
-					peer.transport.Send(Marshal(request))
+					peer.conn.Send(Marshal(request))
 				}
 			}
 			break
@@ -147,7 +147,7 @@ func (s *Signaler) HandleNewWebSocket(transport *transport.WebSocketTransport, r
 							"reason":  "Invalid session [" + sessionID + "]",
 						},
 					}
-					transport.Send(Marshal(msg))
+					conn.Send(Marshal(msg))
 					return
 				}
 				if peer, ok := s.peers[ids[0]]; !ok {
@@ -158,7 +158,7 @@ func (s *Signaler) HandleNewWebSocket(transport *transport.WebSocketTransport, r
 							"reason":  "Peer [" + ids[0] + "] not found.",
 						},
 					}
-					transport.Send(Marshal(msg))
+					conn.Send(Marshal(msg))
 					return
 				} else {
 					bye := map[string]interface{}{
@@ -168,7 +168,7 @@ func (s *Signaler) HandleNewWebSocket(transport *transport.WebSocketTransport, r
 							"session_id": sessionID,
 						},
 					}
-					peer.transport.Send(Marshal(bye))
+					peer.conn.Send(Marshal(bye))
 				}
 
 				if peer, ok := s.peers[ids[1]]; !ok {
@@ -179,7 +179,7 @@ func (s *Signaler) HandleNewWebSocket(transport *transport.WebSocketTransport, r
 							"reason":  "Peer [" + ids[0] + "] not found ",
 						},
 					}
-					transport.Send(Marshal(msg))
+					conn.Send(Marshal(msg))
 					return
 				} else {
 					bye := map[string]interface{}{
@@ -189,7 +189,7 @@ func (s *Signaler) HandleNewWebSocket(transport *transport.WebSocketTransport, r
 							"session_id": sessionID,
 						},
 					}
-					peer.transport.Send(Marshal(bye))
+					peer.conn.Send(Marshal(bye))
 				}
 			}
 			break
@@ -198,7 +198,7 @@ func (s *Signaler) HandleNewWebSocket(transport *transport.WebSocketTransport, r
 				"type": "keepalive",
 				"data": map[string]interface{}{},
 			}
-			transport.Send(Marshal(keepalive))
+			conn.Send(Marshal(keepalive))
 			break
 		default:
 			{
@@ -208,15 +208,15 @@ func (s *Signaler) HandleNewWebSocket(transport *transport.WebSocketTransport, r
 		}
 	})
 
-	transport.On("close", func(code int, text string) {
-		logger.Infof("On Close %v", transport)
+	conn.On("close", func(code int, text string) {
+		logger.Infof("On Close %v", conn)
 		for _, peer := range s.peers {
-			if peer.transport == transport {
+			if peer.conn == conn {
 				logger.Infof("Remove peer %s", peer.info.ID)
 				delete(s.peers, peer.info.ID)
 				break
 			}
 		}
-		s.NotifyPeersUpdate(transport, s.peers)
+		s.NotifyPeersUpdate(conn, s.peers)
 	})
 }
diff --git a/pkg/transport/transport.go b/pkg/transport/transport.go
diff --git a/pkg/websocket/conn.go b/pkg/websocket/conn.go
@@ -0,0 +1,104 @@
+package websocket
+
+import (
+	"errors"
+	"sync"
+	"time"
+
+	"github.com/chuckpreslar/emission"
+	"github.com/cloudwebrtc/flutter-webrtc-server/pkg/logger"
+	"github.com/gorilla/websocket"
+)
+
+const pingPeriod = 5 * time.Second
+
+type WebSocketConn struct {
+	emission.Emitter
+	socket *websocket.Conn
+	mutex  *sync.Mutex
+	closed bool
+}
+
+func NewWebSocketConn(socket *websocket.Conn) *WebSocketConn {
+	var conn WebSocketConn
+	conn.Emitter = *emission.NewEmitter()
+	conn.socket = socket
+	conn.mutex = new(sync.Mutex)
+	conn.closed = false
+	conn.socket.SetCloseHandler(func(code int, text string) error {
+		logger.Warnf("%s [%d]", text, code)
+		conn.Emit("close", code, text)
+		conn.closed = true
+		return nil
+	})
+	return &conn
+}
+
+func (conn *WebSocketConn) ReadMessage() {
+	in := make(chan []byte)
+	stop := make(chan struct{})
+	pingTicker := time.NewTicker(pingPeriod)
+
+	var c = conn.socket
+	go func() {
+		for {
+			_, message, err := c.ReadMessage()
+			if err != nil {
+				logger.Warnf("Got error: %v", err)
+				if c, k := err.(*websocket.CloseError); k {
+					conn.Emit("error", c.Code, c.Text)
+				}
+				close(stop)
+				break
+			}
+			in <- message
+		}
+	}()
+
+	for {
+		select {
+		case _ = <-pingTicker.C:
+			logger.Infof("Send keepalive !!!")
+			if err := conn.Send("{}"); err != nil {
+				logger.Errorf("Keepalive has failed")
+				pingTicker.Stop()
+				return
+			}
+		case message := <-in:
+			{
+				logger.Infof("Recivied data: %s", message)
+				conn.Emit("message", []byte(message))
+			}
+		case <-stop:
+			return
+		}
+	}
+}
+
+/*
+* Send |message| to the connection.
+ */
+func (conn *WebSocketConn) Send(message string) error {
+	logger.Infof("Send data: %s", message)
+	conn.mutex.Lock()
+	defer conn.mutex.Unlock()
+	if conn.closed {
+		return errors.New("websocket: write closed")
+	}
+	return conn.socket.WriteMessage(websocket.TextMessage, []byte(message))
+}
+
+/*
+* Close conn.
+ */
+func (conn *WebSocketConn) Close() {
+	conn.mutex.Lock()
+	defer conn.mutex.Unlock()
+	if conn.closed == false {
+		logger.Infof("Close ws conn now : ", conn)
+		conn.socket.Close()
+		conn.closed = true
+	} else {
+		logger.Warnf("Transport already closed :", conn)
+	}
+}
diff --git a/pkg/websocket/server.go b/pkg/websocket/server.go

Original file line number	Diff line number	Diff line change
`@@ -7,8 +7,8 @@ import (`
`7`	`7`	`"strings"`
`8`	`8`
`9`	`9`	`"github.com/cloudwebrtc/flutter-webrtc-server/pkg/logger"`
`10`		`- "github.com/cloudwebrtc/flutter-webrtc-server/pkg/transport"`
`11`	`10`	`"github.com/cloudwebrtc/flutter-webrtc-server/pkg/turn"`
	`11`	`+ "github.com/cloudwebrtc/flutter-webrtc-server/pkg/websocket"`
`12`	`12`	`)`
`13`	`13`
`14`	`14`	`func Marshal(m map[string]interface{}) string {`
`@@ -38,8 +38,8 @@ type PeerInfo struct {`
`38`	`38`
`39`	`39`	`// Peer .`
`40`	`40`	`type Peer struct {`
`41`		`- info PeerInfo`
`42`		`- transport *transport.WebSocketTransport`
	`41`	`+ info PeerInfo`
	`42`	`+ conn *websocket.WebSocketConn`
`43`	`43`	`}`
`44`	`44`
`45`	`45`	`// Session info.`
`@@ -70,7 +70,7 @@ func (s Signaler) authHandler(username string, realm string, srcAddr net.Addr) (`
`70`	`70`	`return nil, false`
`71`	`71`	`}`
`72`	`72`
`73`		`-func (s Signaler) NotifyPeersUpdate(transport transport.WebSocketTransport, peers map[string]Peer) {`
	`73`	`+func (s Signaler) NotifyPeersUpdate(conn websocket.WebSocketConn, peers map[string]Peer) {`
`74`	`74`	`infos := []PeerInfo{}`
`75`	`75`	`for _, peer := range peers {`
`76`	`76`	`infos = append(infos, peer.info)`
`@@ -79,33 +79,33 @@ func (s Signaler) NotifyPeersUpdate(transport transport.WebSocketTransport, pe`
`79`	`79`	`request["type"] = "peers"`
`80`	`80`	`request["data"] = infos`
`81`	`81`	`for _, peer := range peers {`
`82`		`- peer.transport.Send(Marshal(request))`
	`82`	`+ peer.conn.Send(Marshal(request))`
`83`	`83`	`}`
`84`	`84`	`}`
`85`	`85`
`86`	`86`	`func (s Signaler) HandleTurnServerCredentials(writer http.ResponseWriter, request http.Request) {`
`87`	`87`	`// return turn credentials for client.`
`88`	`88`	`}`
`89`	`89`
`90`		`-func (s Signaler) HandleNewWebSocket(transport transport.WebSocketTransport, request *http.Request) {`
	`90`	`+func (s Signaler) HandleNewWebSocket(conn websocket.WebSocketConn, request *http.Request) {`
`91`	`91`	`logger.Infof("On Open %v", request)`
`92`		`- transport.On("message", func(message []byte) {`
	`92`	`+ conn.On("message", func(message []byte) {`
`93`	`93`	`request := Unmarshal(string(message))`
`94`	`94`	`data := request["data"].(map[string]interface{})`
`95`	`95`	`switch request["type"] {`
`96`	`96`	`case "new":`
`97`	`97`	`{`
`98`	`98`
`99`	`99`	`peer := Peer{`
`100`		`- transport: transport,`
	`100`	`+ conn: conn,`
`101`	`101`	`info: PeerInfo{`
`102`	`102`	`ID: data["id"].(string),`
`103`	`103`	`Name: data["name"].(string),`
`104`	`104`	`UserAgent: data["user_agent"].(string),`
`105`	`105`	`},`
`106`	`106`	`}`
`107`	`107`	`s.peers[peer.info.ID] = peer`
`108`		`- s.NotifyPeersUpdate(transport, s.peers)`
	`108`	`+ s.NotifyPeersUpdate(conn, s.peers)`
`109`	`109`	`}`
`110`	`110`	`break`
`111`	`111`	`case "leave":`
`@@ -128,10 +128,10 @@ func (s Signaler) HandleNewWebSocket(transport transport.WebSocketTransport, r`
`128`	`128`	`"reason": "Peer [" + To + "] not found ",`
`129`	`129`	`},`
`130`	`130`	`}`
`131`		`- transport.Send(Marshal(msg))`
	`131`	`+ conn.Send(Marshal(msg))`
`132`	`132`	`return`
`133`	`133`	`} else {`
`134`		`- peer.transport.Send(Marshal(request))`
	`134`	`+ peer.conn.Send(Marshal(request))`
`135`	`135`	`}`
`136`	`136`	`}`
`137`	`137`	`break`
`@@ -147,7 +147,7 @@ func (s Signaler) HandleNewWebSocket(transport transport.WebSocketTransport, r`
`147`	`147`	`"reason": "Invalid session [" + sessionID + "]",`
`148`	`148`	`},`
`149`	`149`	`}`
`150`		`- transport.Send(Marshal(msg))`
	`150`	`+ conn.Send(Marshal(msg))`
`151`	`151`	`return`
`152`	`152`	`}`
`153`	`153`	`if peer, ok := s.peers[ids[0]]; !ok {`
`@@ -158,7 +158,7 @@ func (s Signaler) HandleNewWebSocket(transport transport.WebSocketTransport, r`
`158`	`158`	`"reason": "Peer [" + ids[0] + "] not found.",`
`159`	`159`	`},`
`160`	`160`	`}`
`161`		`- transport.Send(Marshal(msg))`
	`161`	`+ conn.Send(Marshal(msg))`
`162`	`162`	`return`
`163`	`163`	`} else {`
`164`	`164`	`bye := map[string]interface{}{`
`@@ -168,7 +168,7 @@ func (s Signaler) HandleNewWebSocket(transport transport.WebSocketTransport, r`
`168`	`168`	`"session_id": sessionID,`
`169`	`169`	`},`
`170`	`170`	`}`
`171`		`- peer.transport.Send(Marshal(bye))`
	`171`	`+ peer.conn.Send(Marshal(bye))`
`172`	`172`	`}`
`173`	`173`
`174`	`174`	`if peer, ok := s.peers[ids[1]]; !ok {`
`@@ -179,7 +179,7 @@ func (s Signaler) HandleNewWebSocket(transport transport.WebSocketTransport, r`
`179`	`179`	`"reason": "Peer [" + ids[0] + "] not found ",`
`180`	`180`	`},`
`181`	`181`	`}`
`182`		`- transport.Send(Marshal(msg))`
	`182`	`+ conn.Send(Marshal(msg))`
`183`	`183`	`return`
`184`	`184`	`} else {`
`185`	`185`	`bye := map[string]interface{}{`
`@@ -189,7 +189,7 @@ func (s Signaler) HandleNewWebSocket(transport transport.WebSocketTransport, r`
`189`	`189`	`"session_id": sessionID,`
`190`	`190`	`},`
`191`	`191`	`}`
`192`		`- peer.transport.Send(Marshal(bye))`
	`192`	`+ peer.conn.Send(Marshal(bye))`
`193`	`193`	`}`
`194`	`194`	`}`
`195`	`195`	`break`
`@@ -198,7 +198,7 @@ func (s Signaler) HandleNewWebSocket(transport transport.WebSocketTransport, r`
`198`	`198`	`"type": "keepalive",`
`199`	`199`	`"data": map[string]interface{}{},`
`200`	`200`	`}`
`201`		`- transport.Send(Marshal(keepalive))`
	`201`	`+ conn.Send(Marshal(keepalive))`
`202`	`202`	`break`
`203`	`203`	`default:`
`204`	`204`	`{`
`@@ -208,15 +208,15 @@ func (s Signaler) HandleNewWebSocket(transport transport.WebSocketTransport, r`
`208`	`208`	`}`
`209`	`209`	`})`
`210`	`210`
`211`		`- transport.On("close", func(code int, text string) {`
`212`		`- logger.Infof("On Close %v", transport)`
	`211`	`+ conn.On("close", func(code int, text string) {`
	`212`	`+ logger.Infof("On Close %v", conn)`
`213`	`213`	`for _, peer := range s.peers {`
`214`		`- if peer.transport == transport {`
	`214`	`+ if peer.conn == conn {`
`215`	`215`	`logger.Infof("Remove peer %s", peer.info.ID)`
`216`	`216`	`delete(s.peers, peer.info.ID)`
`217`	`217`	`break`
`218`	`218`	`}`
`219`	`219`	`}`
`220`		`- s.NotifyPeersUpdate(transport, s.peers)`
	`220`	`+ s.NotifyPeersUpdate(conn, s.peers)`
`221`	`221`	`})`
`222`	`222`	`}`