rpc-go/client/client.go

package client

import (
	"context"
	"fmt"
	"reflect"
	"sync"
	"time"

	logging "git.loafle.net/commons/logging-go"
	"git.loafle.net/commons/rpc-go/protocol"
	"git.loafle.net/commons/rpc-go/registry"
	csc "git.loafle.net/commons/server-go/client"
)

var uint64Type = reflect.TypeOf(uint64(0))

type Client struct {
	Connector       csc.Connector
	Codec           protocol.ClientCodec
	RPCInvoker      registry.RPCInvoker
	Name            string
	PendingRequests int
	RequestTimeout  time.Duration

	stopChan chan struct{}
	stopWg   sync.WaitGroup

	requestID        uint64
	requestQueueChan chan *requestState

	pendingRequests sync.Map
}

func (c *Client) Start() error {
	if c.stopChan != nil {
		return fmt.Errorf("%s already running. Stop it before starting it again", c.logHeader())
	}

	if nil == c.Connector {
		return fmt.Errorf("%s Connector is not valid", c.logHeader())
	}

	if err := c.Connector.Validate(); nil != err {
		return err
	}

	if nil == c.Codec {
		return fmt.Errorf("%s Codec is not valid", c.logHeader())
	}

	if 0 >= c.PendingRequests {
		c.PendingRequests = 1024
	}

	if 0 >= c.RequestTimeout {
		c.RequestTimeout = 20 * time.Second
	}

	readChan, writeChan, err := c.Connector.Connect()
	if nil != err {
		return err
	}

	c.requestQueueChan = make(chan *requestState, c.PendingRequests)
	c.stopChan = make(chan struct{})

	c.stopWg.Add(1)
	go c.handleClient(readChan, writeChan)

	return nil
}

func (c *Client) Stop(ctx context.Context) error {
	if c.stopChan == nil {
		return fmt.Errorf("%s must be started before stopping it", c.logHeader())
	}
	close(c.stopChan)
	c.stopWg.Wait()
	c.stopChan = nil

	return nil
}

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

func (c *Client) Send(method string, params ...interface{}) error {
	_, err := c.internalSend(false, nil, method, params...)

	return err
}

func (c *Client) Call(result interface{}, method string, params ...interface{}) error {
	return c.CallTimeout(c.RequestTimeout, result, method, params...)
}

func (c *Client) CallTimeout(timeout time.Duration, result interface{}, method string, params ...interface{}) error {
	rs, err := c.internalSend(true, result, method, params...)
	if nil != err {
		return err
	}

	t := retainTimer(timeout)
	defer func() {
		releaseRequestState(rs)
		releaseTimer(t)
	}()

	select {
	case <-rs.doneChan:
		result = rs.result
		return rs.clientError
	case <-t.C:
		rs.cancel()
		return newError(method, params, fmt.Errorf("%s Timeout", c.logHeader()))
	}
}

func (c *Client) getRequestID() uint64 {
	c.requestID++
	return c.requestID
}

func (c *Client) internalSend(hasResponse bool, result interface{}, method string, params ...interface{}) (*requestState, error) {
	rs := retainRequestState()

	rs.method = method
	rs.params = params
	if hasResponse {
		rs.id = c.getRequestID()
		rs.result = result
		rs.doneChan = make(chan *requestState, 1)
	}

	select {
	case c.requestQueueChan <- rs:
		return rs, nil
	default:
		if !hasResponse {
			releaseRequestState(rs)
			return nil, newError(method, params, fmt.Errorf("%s Request Queue overflow", c.logHeader()))
		}
		select {
		case oldRS := <-c.requestQueueChan:
			if nil != oldRS.doneChan {
				oldRS.setError(fmt.Errorf("%s Request Queue overflow", c.logHeader()))
				oldRS.done()
			} else {
				releaseRequestState(oldRS)
			}
		default:
		}
		select {
		case c.requestQueueChan <- rs:
			return rs, nil
		default:
			releaseRequestState(rs)
			return nil, newError(method, params, fmt.Errorf("%s Request Queue overflow", c.logHeader()))
		}
	}
}

func (c *Client) handleClient(readChan <-chan []byte, writeChan chan<- []byte) {
	defer func() {

		if err := c.Connector.Disconnect(); nil != err {
			logging.Logger().Warn(err)
		}

		logging.Logger().Infof("%s Stopped", c.logHeader())
		c.stopWg.Done()
	}()

	stopChan := make(chan struct{})
	sendDoneChan := make(chan error)
	receiveDoneChan := make(chan error)

	go c.handleSend(stopChan, sendDoneChan, writeChan)
	go c.handleReceive(stopChan, receiveDoneChan, readChan)

	logging.Logger().Infof("%s Started", c.logHeader())

	select {
	case <-sendDoneChan:
		close(stopChan)
		<-receiveDoneChan
	case <-receiveDoneChan:
		close(stopChan)
		<-sendDoneChan
	case <-c.stopChan:
		close(stopChan)
		<-sendDoneChan
		<-receiveDoneChan
	}
}

func (c *Client) handleSend(stopChan <-chan struct{}, doneChan chan<- error, writeChan chan<- []byte) {
	var (
		rs      *requestState
		id      interface{}
		message []byte
		err     error
		ok      bool
	)

	defer func() {
		doneChan <- err
	}()

LOOP:
	for {
		select {
		case rs, ok = <-c.requestQueueChan:
			if !ok {
				return
			}
			if rs.isCanceled() {
				if nil != rs.doneChan {
					rs.done()
				} else {
					releaseRequestState(rs)
				}
				continue LOOP
			}

			id = nil
			if 0 < rs.id {
				id = rs.id
			}
			message, err = c.Codec.NewRequest(rs.method, rs.params, id)
			if nil != err {
				rs.setError(err)
				rs.done()
				continue LOOP
			}

			select {
			case writeChan <- message:
			default:
				rs.setError(fmt.Errorf("%s cannot send request", c.logHeader()))
				rs.done()
				continue LOOP
			}

			if 0 < rs.id {
				c.pendingRequests.Store(rs.id, rs)
			}
		case <-stopChan:
			return
		}
	}
}

func (c *Client) handleReceive(stopChan <-chan struct{}, doneChan chan<- error, readChan <-chan []byte) {
	var (
		message []byte
		err     error
		ok      bool
	)

	defer func() {
		doneChan <- err
	}()

LOOP:
	for {
		select {
		case message, ok = <-readChan:
			if !ok {
				return
			}
			resCodec, err := c.Codec.NewResponse(message)
			if nil != err {
				logging.Logger().Debug(err)
				continue LOOP
			}
			if nil == resCodec.ID() {
				// notification
				notiCodec, err := resCodec.Notification()
				if nil != err {
					logging.Logger().Warnf("%s notification error %v", c.logHeader(), err)
					continue LOOP
				}

				go c.handleNotification(notiCodec)
			} else {
				// response
				go c.handleResponse(resCodec)
			}
		case <-stopChan:
			return
		}
	}

}

func (c *Client) handleResponse(resCodec protocol.ClientResponseCodec) {
	id := reflect.ValueOf(resCodec.ID()).Convert(uint64Type).Uint()
	_rs, ok := c.pendingRequests.Load(id)
	if !ok {
		logging.Logger().Warnf("%s unexpected ID=[%d] obtained from server", c.logHeader(), id)
		return
	}
	rs := _rs.(*requestState)
	rs.setError(resCodec.Error())
	err := resCodec.Result(rs.result)
	if nil != err {
		rs.setError(err)
	}

	rs.done()
}

func (c *Client) handleNotification(notiCodec protocol.ClientNotificationCodec) {
	if nil == c.RPCInvoker {
		logging.Logger().Warnf("%s received notification method[%s] but RPC Invoker is not exist", c.logHeader(), notiCodec.Method())
		return
	}

	_, err := c.RPCInvoker.Invoke(notiCodec)
	if nil != err {
		logging.Logger().Errorf("%s invoking of notification method[%s] has been failed %v", c.logHeader(), notiCodec.Method(), err)
	}
}