124 lines
2.7 KiB
Go
124 lines
2.7 KiB
Go
package socket
|
|
|
|
import (
|
|
"fmt"
|
|
"time"
|
|
|
|
olog "git.loafle.net/overflow/log-go"
|
|
)
|
|
|
|
type SocketMessage func() (int, []byte)
|
|
|
|
func MakeSocketMessage(messageType int, message []byte) SocketMessage {
|
|
return func() (int, []byte) { return messageType, message }
|
|
}
|
|
|
|
func connReadHandler(readWriteHandler ReadWriteHandler, conn Conn, stopChan <-chan struct{}, doneChan chan<- error, readChan chan<- SocketMessage) {
|
|
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
|
|
var messageType int
|
|
readMessageChan := make(chan struct{})
|
|
|
|
go func() {
|
|
messageType, 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)
|
|
}
|
|
olog.Logger().Debug(err.Error())
|
|
return
|
|
}
|
|
|
|
readChan <- MakeSocketMessage(messageType, message)
|
|
}
|
|
}
|
|
|
|
func connWriteHandler(readWriteHandler ReadWriteHandler, conn Conn, stopChan <-chan struct{}, doneChan chan<- error, writeChan <-chan SocketMessage) {
|
|
var (
|
|
socketMessage SocketMessage
|
|
message []byte
|
|
messageType int
|
|
ok bool
|
|
err error
|
|
)
|
|
|
|
defer func() {
|
|
doneChan <- err
|
|
}()
|
|
|
|
ticker := time.NewTicker(readWriteHandler.GetPingPeriod())
|
|
defer func() {
|
|
ticker.Stop()
|
|
}()
|
|
for {
|
|
select {
|
|
case socketMessage, 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
|
|
}
|
|
|
|
messageType, message = socketMessage()
|
|
|
|
err = conn.WriteMessage(messageType, message)
|
|
if err != nil {
|
|
olog.Logger().Debug(err.Error())
|
|
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 {
|
|
olog.Logger().Debug(err.Error())
|
|
return
|
|
}
|
|
case <-stopChan:
|
|
return
|
|
}
|
|
}
|
|
}
|