package socket import ( "fmt" "io" "time" logging "git.loafle.net/commons/logging-go" ) func connReadHandler(readWriteHandler ReadWriteHandler, conn Conn, stopChan <-chan struct{}, doneChan chan<- error, readChan chan<- []byte) { var ( err error ) defer func() { doneChan <- err }() if 0 < readWriteHandler.GetMaxMessageSize() { conn.SetReadLimit(readWriteHandler.GetMaxMessageSize()) } if 0 < readWriteHandler.GetReadTimeout() { conn.SetReadDeadline(time.Now().Add(readWriteHandler.GetReadTimeout())) } else { conn.SetReadDeadline(time.Time{}) } conn.SetPongHandler(func(string) error { if 0 < readWriteHandler.GetPongTimeout() { conn.SetReadDeadline(time.Now().Add(readWriteHandler.GetPongTimeout())) } else { conn.SetReadDeadline(time.Time{}) } return nil }) for { var message []byte readMessageChan := make(chan struct{}) go func() { _, message, err = conn.ReadMessage() close(readMessageChan) }() select { case <-stopChan: return case <-readMessageChan: } if nil != err { if IsUnexpectedCloseError(err, CloseGoingAway, CloseAbnormalClosure) { err = fmt.Errorf("Read error %v", err) } logging.Logger().Debug(err) return } readChan <- message } } func connWriteHandler(readWriteHandler ReadWriteHandler, conn Conn, stopChan <-chan struct{}, doneChan chan<- error, writeChan <-chan []byte) { var ( wc io.WriteCloser message []byte ok bool err error ) defer func() { doneChan <- err }() ticker := time.NewTicker(readWriteHandler.GetPingPeriod()) defer func() { ticker.Stop() }() for { select { case message, ok = <-writeChan: if 0 < readWriteHandler.GetWriteTimeout() { conn.SetWriteDeadline(time.Now().Add(readWriteHandler.GetWriteTimeout())) } else { conn.SetWriteDeadline(time.Time{}) } if !ok { conn.WriteMessage(CloseMessage, []byte{}) return } wc, err = conn.NextWriter(TextMessage) if err != nil { logging.Logger().Debug(err) return } wc.Write(message) if err = wc.Close(); nil != err { logging.Logger().Debug(err) return } case <-ticker.C: if 0 < readWriteHandler.GetPingTimeout() { conn.SetWriteDeadline(time.Now().Add(readWriteHandler.GetPingTimeout())) } else { conn.SetWriteDeadline(time.Time{}) } if err = conn.WriteMessage(PingMessage, nil); nil != err { logging.Logger().Debug(err) return } case <-stopChan: return } } }