tecamino-driver-artNet/websocket/client.go

package websocket

import (
	"encoding/json"
	"fmt"
	"log"
	"time"

	"github.com/gorilla/websocket"
	json_dataModels "github.com/tecamino/tecamino-json_data/models"
)

const (
	// Time allowed to write a message to the peer.
	writeWait = 10 * time.Second

	// Time allowed to read the next pong message from the peer.
	pongWait = 10 * time.Second

	// Send pings to peer with this period. Must be less than pongWait.
	pingPeriod = (pongWait * 9) / 10
)

type Client struct {
	ip         string
	port       uint
	Connected  bool
	conn       *websocket.Conn
	OnOpen     func()
	OnMessage  func(data []byte)
	OnClose    func(code int, reason string)
	OnError    func(err error)
	OnPing     func()
	OnPong     func()
	sendPong   chan string
	send       chan []byte
	unregister chan []byte
}

// Connect to websocket server
// ip: ip address of server
func NewClient(ip, id string, port uint) (*Client, error) {
	url := fmt.Sprintf("ws://%s:%d/ws?id=%s", ip, port, id)
	c := &Client{
		ip:         ip,
		port:       port,
		Connected:  true,
		sendPong:   make(chan string),
		send:       make(chan []byte),
		unregister: make(chan []byte),
	}

	dialer := websocket.DefaultDialer
	conn, resp, err := dialer.Dial(url, nil)
	if err != nil {
		if c.OnError != nil {
			c.OnError(err)
		}
		return nil, fmt.Errorf("dial error %v (status %v)", err, resp)
	}
	c.conn = conn

	// Setup control handlers
	conn.SetPingHandler(func(appData string) error {
		if c.OnPing != nil {
			c.OnPing()
		}
		conn.SetWriteDeadline(time.Now().Add(writeWait))
		conn.SetReadDeadline(time.Now().Add(writeWait))
		c.sendPong <- appData
		return nil
	})

	conn.SetPongHandler(func(string) error {
		conn.SetReadDeadline(time.Now().Add(pongWait))
		if c.OnPong != nil {
			c.OnPong()
		}
		return nil
	})

	conn.SetCloseHandler(func(code int, text string) error {
		if c.OnClose != nil {
			c.OnClose(code, text)
		}
		return nil
	})

	if c.OnOpen != nil {
		c.OnOpen()
	}

	// Start reading messages from client
	go c.Read()
	go c.Write()
	return c, nil
}

func (c *Client) Read() {
	if c.OnOpen != nil {
		c.OnOpen()
	}

	c.conn.SetReadDeadline(time.Now().Add(writeWait))
	for c.Connected {
		msgType, msg, err := c.conn.ReadMessage()
		if err != nil {
			c.handleError(fmt.Errorf("read error (id:%s:%d): %w", c.ip, c.port, err))
			return
		}
		switch msgType {
		case websocket.CloseMessage:
			c.Close(websocket.CloseNormalClosure, "Client closed")
			return
		case websocket.TextMessage:
			if c.OnMessage != nil {
				c.OnMessage(msg)
			} else {
				log.Printf("Received message but no handler set (id:%s:%d): %s", c.ip, c.port, string(msg))
			}
		default:
			log.Printf("Unhandled message type %d (id:%s:%d)", msgType, c.ip, c.port)
		}
	}
}

func (c *Client) Write() {
	ticker := time.NewTicker(pingPeriod)
	defer func() {
		ticker.Stop()
		c.conn.Close()
	}()
	for {

		select {
		case message, ok := <-c.send:
			c.conn.SetWriteDeadline(time.Now().Add(writeWait))
			if !ok {
				// The hub closed the channel.
				if err := c.conn.WriteMessage(websocket.CloseMessage, []byte{}); err != nil {
					c.handleError(err)
					return
				}
				c.handleError(fmt.Errorf("server %s:%d closed channel", c.ip, c.port))
				return
			} else {
				if err := c.conn.WriteMessage(websocket.TextMessage, message); err != nil {
					c.handleError(err)
					return
				}
			}
		case <-ticker.C:
			c.conn.SetWriteDeadline(time.Now().Add(writeWait))
			if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
				c.handleError(err)
				return
			}

			if c.OnPing != nil {
				c.OnPing()
			}

		case message, ok := <-c.sendPong:
			if ok {
				c.conn.WriteMessage(websocket.PongMessage, []byte(message))
			}
		case message := <-c.unregister:
			c.conn.WriteMessage(websocket.CloseMessage, message)
			c.Connected = false
			close(c.send)
			close(c.sendPong)
			close(c.unregister)
			return
		}

	}
}

func (c *Client) SendRequest(req *json_dataModels.Request) error {
	if !c.Connected {
		return nil
	}

	data, err := json.Marshal(*req)
	if err != nil {
		return err
	}
	c.send <- data
	return nil
}

func (c *Client) ReadJsonData(data []byte) (json_dataModels.Response, error) {
	var resp json_dataModels.Response
	err := json.Unmarshal(data, &resp)
	return resp, err
}

// Close connection to websocket server
func (c *Client) Close(code int, reason string) error {
	closeMsg := websocket.FormatCloseMessage(code, reason)
	select {
	case c.unregister <- closeMsg: // Attempt to send
	default: // If the channel is full, this runs
		return fmt.Errorf("attempt close client socket failed")
	}
	if c.OnClose != nil {
		c.OnClose(code, reason)
	}
	return nil
}

func (c *Client) handleError(err error) {
	if c.OnError != nil {
		c.OnError(err)
	}
	if err := c.Close(websocket.CloseInternalServerErr, err.Error()); err != nil {
		if c.OnError != nil {
			c.OnError(err)
		} else {
			fmt.Println("error: ", err)
		}
	}
}