commons/server-go
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ing

Browse Source
This commit is contained in:
crusader 2018-04-04 17:36:23 +09:00
parent 17223f4bb9
commit 99bac0d66b
6 changed files with 116 additions and 77 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
fasthttp/websocket/client.go
View File

@ -166,30 +166,28 @@ func (c *Client) handleConnection(conn *server.Conn) {
select {
case <-readerDoneChan:
close(stopChan)
conn.Close()
<-writerDoneChan
conn = nil
case <-writerDoneChan:
close(stopChan)
conn.Close()
<-readerDoneChan
conn = nil
case <-c.stopChan:
close(stopChan)
conn.Close()
<-readerDoneChan
<-writerDoneChan
conn = nil
}
}
func handleClientRead(c *Client, conn *server.Conn, doneChan chan<- struct{}, stopChan <-chan struct{}) {
func handleClientRead(c *Client, conn *server.Conn, stopChan <-chan struct{}, doneChan chan<- struct{}) {
defer func() {
close(doneChan)
doneChan <- struct{}{}
}()
conn.SetReadLimit(c.MaxMessageSize)
conn.SetReadDeadline(time.Now().Add(c.ReadTimeout))
if 0 < c.MaxMessageSize {
conn.SetReadLimit(c.MaxMessageSize)
}
if 0 < c.ReadTimeout {
conn.SetReadDeadline(time.Now().Add(c.ReadTimeout))
}
conn.SetPongHandler(func(string) error {
conn.SetReadDeadline(time.Now().Add(c.PongTimeout))
return nil
@ -205,37 +203,30 @@ func handleClientRead(c *Client, conn *server.Conn, doneChan chan<- struct{}, st
go func() {
_, message, err = conn.ReadMessage()
if err != nil {
if server.IsUnexpectedCloseError(err, server.CloseGoingAway, server.CloseAbnormalClosure) {
logging.Logger().Debugf(c.clientMessage(fmt.Sprintf("Read error %v", err)))
}
}
close(readMessageChan)
}()
select {
case <-c.stopChan:
case <-stopChan:
<-readMessageChan
break
return
case <-readMessageChan:
}
if nil != err {
select {
case <-c.stopChan:
break
case <-time.After(time.Second):
if server.IsUnexpectedCloseError(err, server.CloseGoingAway, server.CloseAbnormalClosure) {
logging.Logger().Debugf(c.clientMessage(fmt.Sprintf("Read error %v", err)))
}
continue
return
}
c.readChan <- message
}
}
func handleClientWrite(c *Client, conn *server.Conn, doneChan chan<- struct{}, stopChan <-chan struct{}) {
func handleClientWrite(c *Client, conn *server.Conn, stopChan <-chan struct{}, doneChan chan<- struct{}) {
defer func() {
close(doneChan)
doneChan <- struct{}{}
}()
ticker := time.NewTicker(c.PingPeriod)
@ -265,8 +256,8 @@ func handleClientWrite(c *Client, conn *server.Conn, doneChan chan<- struct{}, s
if err := conn.WriteMessage(server.PingMessage, nil); nil != err {
return
}
case <-c.stopChan:
break
case <-stopChan:
return
}
}
}

49
fasthttp/websocket/server.go
View File

@ -212,6 +212,10 @@ func (s *Server) handleConnection(servlet Servlet, servletCtx server.ServletCtx,
defer s.connections.Delete(conn)
servlet.OnConnect(servletCtx, conn)
conn.SetCloseHandler(func(code int, text string) error {
logging.Logger().Debugf("close")
return nil
})
stopChan := make(chan struct{})
servletDoneChan := make(chan struct{})
@ -229,29 +233,21 @@ func (s *Server) handleConnection(servlet Servlet, servletCtx server.ServletCtx,
select {
case <-readerDoneChan:
close(stopChan)
conn.Close()
<-writerDoneChan
<-servletDoneChan
conn = nil
case <-writerDoneChan:
close(stopChan)
conn.Close()
<-readerDoneChan
<-servletDoneChan
conn = nil
case <-servletDoneChan:
close(stopChan)
conn.Close()
<-readerDoneChan
<-writerDoneChan
conn = nil
case <-s.stopChan:
close(stopChan)
conn.Close()
<-readerDoneChan
<-writerDoneChan
<-servletDoneChan
conn = nil
}
}
@ -259,13 +255,17 @@ func (s *Server) onError(ctx *fasthttp.RequestCtx, status int, reason error) {
s.ServerHandler.OnError(s.ctx, ctx, status, reason)
}
func handleRead(s *Server, conn *server.Conn, doneChan chan<- struct{}, stopChan <-chan struct{}, readChan chan []byte) {
func handleRead(s *Server, conn *server.Conn, stopChan <-chan struct{}, doneChan chan<- struct{}, readChan chan []byte) {
defer func() {
close(doneChan)
doneChan <- struct{}{}
}()
conn.SetReadLimit(s.ServerHandler.GetMaxMessageSize())
conn.SetReadDeadline(time.Now().Add(s.ServerHandler.GetReadTimeout()))
if 0 < s.ServerHandler.GetMaxMessageSize() {
conn.SetReadLimit(s.ServerHandler.GetMaxMessageSize())
}
if 0 < s.ServerHandler.GetReadTimeout() {
conn.SetReadDeadline(time.Now().Add(s.ServerHandler.GetReadTimeout()))
}
conn.SetPongHandler(func(string) error {
conn.SetReadDeadline(time.Now().Add(s.ServerHandler.GetPongTimeout()))
return nil
@ -281,37 +281,30 @@ func handleRead(s *Server, conn *server.Conn, doneChan chan<- struct{}, stopChan
go func() {
_, message, err = conn.ReadMessage()
if err != nil {
if server.IsUnexpectedCloseError(err, server.CloseGoingAway, server.CloseAbnormalClosure) {
logging.Logger().Debugf(s.serverMessage(fmt.Sprintf("Read error %v", err)))
}
}
close(readMessageChan)
}()
select {
case <-s.stopChan:
<-readMessageChan
break
return
case <-readMessageChan:
}
if nil != err {
select {
case <-s.stopChan:
break
case <-time.After(time.Second):
if server.IsUnexpectedCloseError(err, server.CloseGoingAway, server.CloseAbnormalClosure) {
logging.Logger().Debugf(s.serverMessage(fmt.Sprintf("Read error %v", err)))
}
continue
return
}
readChan <- message
}
}
func handleWrite(s *Server, conn *server.Conn, doneChan chan<- struct{}, stopChan <-chan struct{}, writeChan chan []byte) {
func handleWrite(s *Server, conn *server.Conn, stopChan <-chan struct{}, doneChan chan<- struct{}, writeChan chan []byte) {
defer func() {
close(doneChan)
doneChan <- struct{}{}
}()
ticker := time.NewTicker(s.ServerHandler.GetPingPeriod())
@ -321,7 +314,9 @@ func handleWrite(s *Server, conn *server.Conn, doneChan chan<- struct{}, stopCha
for {
select {
case message, ok := <-writeChan:
conn.SetWriteDeadline(time.Now().Add(s.ServerHandler.GetWriteTimeout()))
if 0 < s.ServerHandler.GetWriteTimeout() {
conn.SetWriteDeadline(time.Now().Add(s.ServerHandler.GetWriteTimeout()))
}
if !ok {
conn.WriteMessage(server.CloseMessage, []byte{})
return
@ -342,7 +337,7 @@ func handleWrite(s *Server, conn *server.Conn, doneChan chan<- struct{}, stopCha
return
}
case <-s.stopChan:
break
return
}
}
}

32
fasthttp/websocket/servlet.go
View File

@ -10,3 +10,35 @@ type Servlet interface {
Handshake(servletCtx server.ServletCtx, ctx *fasthttp.RequestCtx) (*fasthttp.ResponseHeader, error)
}
type Servlets struct {
Servlet
}
func (s *Servlets) ServletCtx(serverCtx server.ServerCtx) server.ServletCtx {
return server.NewServletContext(nil, serverCtx)
}
func (s *Servlets) Init(serverCtx server.ServerCtx) error {
return nil
}
func (s *Servlets) Destroy(serverCtx server.ServerCtx) {
//
}
func (s *Servlets) OnConnect(servletCtx server.ServletCtx, conn *server.Conn) {
//
}
func (s *Servlets) OnDisconnect(servletCtx server.ServletCtx) {
//
}
func (s *Servlets) Handshake(servletCtx server.ServletCtx, ctx *fasthttp.RequestCtx) (*fasthttp.ResponseHeader, error) {
return nil, nil
}
func (s *Servlets) Handle(servletCtx server.ServletCtx, stopChan <-chan struct{}, doneChan chan<- struct{}, readChan <-chan []byte, writeChan chan<- []byte) {
}

35
net/client.go
View File

@ -141,26 +141,20 @@ func (c *Client) handleConnection(conn *server.Conn) {
select {
case <-readerDoneChan:
close(stopChan)
conn.Close()
<-writerDoneChan
conn = nil
case <-writerDoneChan:
close(stopChan)
conn.Close()
<-readerDoneChan
conn = nil
case <-c.stopChan:
close(stopChan)
conn.Close()
<-readerDoneChan
<-writerDoneChan
conn = nil
}
}
func handleClientRead(c *Client, conn *server.Conn, doneChan chan<- struct{}, stopChan <-chan struct{}) {
func handleClientRead(c *Client, conn *server.Conn, stopChan <-chan struct{}, doneChan chan<- struct{}) {
defer func() {
close(doneChan)
doneChan <- struct{}{}
}()
if 0 < c.MaxMessageSize {
@ -184,37 +178,30 @@ func handleClientRead(c *Client, conn *server.Conn, doneChan chan<- struct{}, st
go func() {
_, message, err = conn.ReadMessage()
if err != nil {
if server.IsUnexpectedCloseError(err, server.CloseGoingAway, server.CloseAbnormalClosure) {
logging.Logger().Debugf(c.clientMessage(fmt.Sprintf("Read error %v", err)))
}
}
close(readMessageChan)
}()
select {
case <-c.stopChan:
case <-stopChan:
<-readMessageChan
break
return
case <-readMessageChan:
}
if nil != err {
select {
case <-c.stopChan:
break
case <-time.After(time.Second):
if server.IsUnexpectedCloseError(err, server.CloseGoingAway, server.CloseAbnormalClosure) {
logging.Logger().Debugf(c.clientMessage(fmt.Sprintf("Read error %v", err)))
}
continue
return
}
c.readChan <- message
}
}
func handleClientWrite(c *Client, conn *server.Conn, doneChan chan<- struct{}, stopChan <-chan struct{}) {
func handleClientWrite(c *Client, conn *server.Conn, stopChan <-chan struct{}, doneChan chan<- struct{}) {
defer func() {
close(doneChan)
doneChan <- struct{}{}
}()
ticker := time.NewTicker(c.PingPeriod)
@ -246,8 +233,8 @@ func handleClientWrite(c *Client, conn *server.Conn, doneChan chan<- struct{}, s
if err := conn.WriteMessage(server.PingMessage, nil); nil != err {
return
}
case <-c.stopChan:
break
case <-stopChan:
return
}
}
}

2
net/server.go
View File

@ -178,6 +178,7 @@ func (s *Server) handleConnection(servlet Servlet, servletCtx server.ServletCtx,
if nil != conn {
conn.Close()
}
servlet.OnDisconnect(servletCtx)
logging.Logger().Infof(s.serverMessage(fmt.Sprintf("Client[%s] has been disconnected", addr)))
s.stopWg.Done()
}()
@ -290,6 +291,7 @@ func handleWrite(s *Server, conn *server.Conn, stopChan <-chan struct{}, doneCha
if 0 < s.ServerHandler.GetWriteTimeout() {
conn.SetWriteDeadline(time.Now().Add(s.ServerHandler.GetWriteTimeout()))
}
if !ok {
conn.WriteMessage(server.CloseMessage, []byte{})
return

32
net/servlet.go
View File

@ -11,3 +11,35 @@ type Servlet interface {
Handshake(servletCtx server.ServletCtx, conn net.Conn) error
}
type Servlets struct {
Servlet
}
func (s *Servlets) ServletCtx(serverCtx server.ServerCtx) server.ServletCtx {
return server.NewServletContext(nil, serverCtx)
}
func (s *Servlets) Init(serverCtx server.ServerCtx) error {
return nil
}
func (s *Servlets) Destroy(serverCtx server.ServerCtx) {
//
}
func (s *Servlets) OnConnect(servletCtx server.ServletCtx, conn *server.Conn) {
//
}
func (s *Servlets) OnDisconnect(servletCtx server.ServletCtx) {
//
}
func (s *Servlets) Handshake(servletCtx server.ServletCtx, conn net.Conn) error {
return nil
}
func (s *Servlets) Handle(servletCtx server.ServletCtx, stopChan <-chan struct{}, doneChan chan<- struct{}, readChan <-chan []byte, writeChan chan<- []byte) {
}