server-go/socket/net/client.go

package net

import (
	"crypto/tls"
	"fmt"
	"net"
	"sync"
	"time"

	"git.loafle.net/commons/logging-go"
	"git.loafle.net/commons/server-go/socket"
)

type Client struct {
	socket.Client

	socket.ClientConnHandlers
	socket.ReadWriteHandlers

	Name string `json:"name"`

	Network      string `json:"network"`
	Address      string `json:"address"`
	LocalAddress net.Addr

	stopChan chan struct{}
	stopWg   sync.WaitGroup

	readChan  chan []byte
	writeChan chan []byte

	disconnectedChan chan struct{}
	reconnectedChan  chan socket.Conn

	crw socket.ClientReadWriter
}

func (c *Client) Connect() (readChan <-chan []byte, writeChan chan<- []byte, err error) {
	var (
		conn socket.Conn
	)

	if c.stopChan != nil {
		return nil, nil, fmt.Errorf(c.clientMessage("already running. Stop it before starting it again"))
	}

	err = c.Validate()
	if nil != err {
		return nil, nil, err
	}

	conn, err = c.connect()
	if nil != err {
		return nil, nil, err
	}

	c.readChan = make(chan []byte, 256)
	c.writeChan = make(chan []byte, 256)
	c.disconnectedChan = make(chan struct{})
	c.reconnectedChan = make(chan socket.Conn)
	c.stopChan = make(chan struct{})

	c.crw.ReadwriteHandler = c
	c.crw.ReadChan = c.readChan
	c.crw.WriteChan = c.writeChan
	c.crw.ClientStopChan = c.stopChan
	c.crw.ClientStopWg = &c.stopWg
	c.crw.DisconnectedChan = c.disconnectedChan
	c.crw.ReconnectedChan = c.reconnectedChan

	c.stopWg.Add(2)
	go c.handleReconnect()
	go c.crw.HandleConnection(conn)

	return c.readChan, c.writeChan, nil
}

func (c *Client) Disconnect() error {
	if c.stopChan == nil {
		return fmt.Errorf(c.clientMessage("must be started before stopping it"))
	}
	close(c.stopChan)
	c.stopWg.Wait()

	c.stopChan = nil

	return nil
}

func (c *Client) clientMessage(msg string) string {
	return fmt.Sprintf("Client[%s]: %s", c.Name, msg)
}

func (c *Client) handleReconnect() {
	defer func() {
		c.stopWg.Done()
	}()

RC_LOOP:
	for {
		select {
		case <-c.disconnectedChan:
		case <-c.stopChan:
			return
		}

		if 0 >= c.ReconnectTryTime {
			c.reconnectedChan <- nil
			continue RC_LOOP
		}

		logging.Logger().Debugf("connection lost")

		for indexI := 0; indexI < c.ReconnectTryTime; indexI++ {
			logging.Logger().Debugf("trying reconnect[%d]", indexI)

			conn, err := c.connect()
			if nil == err {
				logging.Logger().Debugf("reconnected")
				c.reconnectedChan <- conn
				continue RC_LOOP
			}
			time.Sleep(c.ReconnectInterval)
		}
		logging.Logger().Debugf("reconnecting has been failed")
	}
}

func (c *Client) connect() (socket.Conn, error) {
	netConn, err := c.dial()
	if nil != err {
		return nil, err
	}

	conn := socket.NewConn(netConn, false, c.ReadBufferSize, c.WriteBufferSize)
	conn.SetCloseHandler(func(code int, text string) error {
		logging.Logger().Debugf("close")
		return nil
	})
	return conn, nil
}

func (c *Client) dial() (net.Conn, error) {
	if err := c.Validate(); nil != err {
		return nil, err
	}

	var deadline time.Time
	if 0 != c.HandshakeTimeout {
		deadline = time.Now().Add(c.HandshakeTimeout)
	}

	d := &net.Dialer{
		KeepAlive: c.KeepAlive,
		Deadline:  deadline,
		LocalAddr: c.LocalAddress,
	}

	conn, err := d.Dial(c.Network, c.Address)
	if nil != err {
		return nil, err
	}

	if nil != c.TLSConfig {
		cfg := c.TLSConfig.Clone()
		tlsConn := tls.Client(conn, cfg)
		if err := tlsConn.Handshake(); err != nil {
			tlsConn.Close()
			return nil, err
		}
		if !cfg.InsecureSkipVerify {
			if err := tlsConn.VerifyHostname(cfg.ServerName); err != nil {
				return nil, err
			}
		}
		conn = tlsConn
	}

	return conn, nil
}

func (c *Client) Validate() error {
	if err := c.ClientConnHandlers.Validate(); nil != err {
		return err
	}
	if err := c.ReadWriteHandlers.Validate(); nil != err {
		return err
	}

	if "" == c.Name {
		c.Name = "Client"
	}

	if "" == c.Network {
		return fmt.Errorf("Client: Network is not valid")
	}

	if "" == c.Address {
		return fmt.Errorf("Client: Address is not valid")
	}

	return nil
}