rest/rest_websocket/socket.go

package rest_websocket

import (
	"context"
	"errors"
	"io"
	"log"
	"sync"
	"sync/atomic"
	"time"

	"git.ali33.ru/fcg-xvii/rest"
	"github.com/gorilla/websocket"
)

func NewSocket(conn *websocket.Conn, pingEnable bool) *Socket {
	ctx, cancel := context.WithCancel(context.Background())
	ws := &Socket{
		conn:        conn,
		wait:        new(sync.Map),
		ctx:         ctx,
		cancel:      cancel,
		writeLocker: &sync.Mutex{},
		chIn:        make(chan *rest.RequestStream, 10),
		pingEnable:  pingEnable,
	}
	ws.lastWrite.Store(time.Now().Unix())
	ws.read()
	return ws
}

type Socket struct {
	ctx         context.Context
	cancel      context.CancelFunc
	conn        *websocket.Conn
	wait        *sync.Map
	chIn        chan *rest.RequestStream
	writeLocker *sync.Mutex
	idCounter   atomic.Int64
	lastWrite   atomic.Int64
	pingEnable  bool
}

func (s *Socket) Context() context.Context {
	return s.ctx
}

// MessagesIn возвращает канал, в который будут переданы все входящие сообщения (rest.RequestTypeMessage и rest.RequestTypeEvent)
func (s *Socket) MessagesIn() <-chan *rest.RequestStream {
	return s.chIn
}

// getID увеличивает счётчик сообщений на единицу и возвраащает результат. используется для маркирования идентификаторами исходящих сообщений
func (s *Socket) getID() int64 {
	return s.idCounter.Add(1)
}

// read реализует чтение входящих сообщений
func (s *Socket) read() {
	//defer log.Println("work close...")
	// контекст
	s.ctx, s.cancel = context.WithCancel(context.Background())
	// создаем канал для обработки входящих сообщений
	chIn := s.exec()
	go func() {
		defer func() {
			s.cancel()
			s.conn.Close()
		}()
		for {
			// Read message from server
			mType, r, err := s.conn.NextReader()
			if err != nil {
				s.cancel()
				log.Println(err)
				return
			}
			switch mType {
			case websocket.TextMessage, websocket.BinaryMessage:
				// Обработка текстового или бинарного сообщения
				req, err := rest.ReadRequestStream(r)
				if err != nil {
					log.Println("data error: ", err)
					return
				}
				log.Println("REQUEST", req.Request.Command, req.Request.Data.JSONPrettyString())
				chIn <- req
			}
		}
	}()
}

// exec реализует обработку сообщений.
func (s *Socket) exec() chan<- *rest.RequestStream {
	ch := make(chan *rest.RequestStream)
	go func() {
		//defer log.Println("exec close...")
		for {
			select {
			// закрытие контекста
			case <-s.ctx.Done():
				return
			// новое сообщение
			case req, ok := <-ch:
				if !ok {
					return
				}
				//log.Println("OOOOOOOOOOOOOOOOOOOOOOOOOOO")
				switch req.Type {
				case rest.RequestTypeIn:
					s.chIn <- req
				case rest.RequestTypeEvent:
					s.chIn <- req
				case rest.RequestTypeOut:
					log.Println("answer in", req.ID)
					ir, check := s.wait.Load(req.ID)
					if check {
						rreq := ir.(*waitRequest)
						s.wait.Delete(rreq.id)
						rreq.answerIn <- req
					}
				}
			// чистка просроченных сообщений и отправка пинга (при необходимости)
			case <-time.After(time.Second * 10):
				// чистим сообщения без ответа по дедлайну
				now := time.Now()
				s.wait.Range(func(key, val any) bool {
					if val.(*waitRequest).timeout.Before(now) {
						log.Println("CLEAN...", key)
						s.wait.Delete(key)
					}
					return true
				})
				// отправляем пинг для проверки, живое соединение или нет
				if s.pingEnable && now.Unix()-s.lastWrite.Load() > 10 {
					log.Println("PING")
					//s.writeLocker.Lock()
					err := s.conn.WriteControl(websocket.PingMessage, nil, time.Now().Add(time.Second))
					//s.writeLocker.Unlock()
					if err != nil {
						log.Println("ping send error")
						s.conn.Close()
						return
					}
					s.lastWrite.Store(time.Now().Unix())
				}
			}
		}
	}()
	return ch
}

func (s *Socket) nextWriter(messageType int) (io.WriteCloser, error) {
	s.writeLocker.Lock()
	res, err := s.conn.NextWriter(messageType)
	s.writeLocker.Unlock()
	s.lastWrite.Store(time.Now().Unix())
	return res, err
}

func (s *Socket) SendMessage(req *rest.RequestStream) (ch <-chan *rest.RequestStream, err error) {
	switch req.Type {
	case rest.RequestTypeIn:
		req.ID = s.getID()
		clReq := newWaitRequest(req.ID, req.Timeout)
		ch = clReq.answer
		s.wait.Store(req.ID, clReq)
	case rest.RequestTypeEvent, rest.RequestTypeOut:
	default:
		return nil, errors.New("unexpected request type")
	}
	var writer io.WriteCloser
	if writer, err = s.nextWriter(websocket.BinaryMessage); err != nil {
		return
	}
	err = req.Write(writer)
	writer.Close()
	return
}

func (s *Socket) Close() {
	s.cancel()
	s.conn.Close()
}