rpc-go/client/client.go
crusader fe3407551c ing
2018-04-26 21:22:39 +09:00

332 lines
6.8 KiB
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
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
}
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 {
rs, err := c.internalSend(false, nil, method, params...)
if nil != err {
return err
}
defer releaseRequestState(rs)
select {
case <-rs.doneChan:
if nil != rs.clientError {
return rs.clientError
}
}
return nil
}
func (c *Client) Call(result interface{}, method string, params ...interface{}) error {
return c.CallTimeout(10, 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
rs.doneChan = make(chan *requestState, 1)
if hasResponse {
rs.id = c.getRequestID()
rs.result = result
}
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)
}
}