server-go/socket/readwrite.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
		}
	}
}