rpc/protocol/json/client.go

package json

import (
	"encoding/json"
	"fmt"
	"io"
	"sync"

	"git.loafle.net/commons_go/rpc/encode"
	"git.loafle.net/commons_go/rpc/protocol"
)

// ----------------------------------------------------------------------------
// Request and Response
// ----------------------------------------------------------------------------

// clientRequest represents a JSON-RPC request sent by a client.
type clientRequest struct {
	// JSON-RPC protocol.
	Version string `json:"jsonrpc"`

	// A String containing the name of the method to be invoked.
	Method string `json:"method"`

	// Object to pass as request parameter to the method.
	Params interface{} `json:"params"`

	// The request id. This can be of any type. It is used to match the
	// response with the request that it is replying to.
	ID interface{} `json:"id"`
}

// clientResponse represents a JSON-RPC response returned to a client.
type clientResponse struct {
	Version string           `json:"jsonrpc"`
	Result  *json.RawMessage `json:"result,omitempty"`
	Error   error            `json:"error,omitempty"`
	ID      interface{}      `json:"id,omitempty"`
}

// clientRequest represents a JSON-RPC request sent by a client.
type clientNotify struct {
	// JSON-RPC protocol.
	Version string `json:"jsonrpc"`

	// A String containing the name of the method to be invoked.
	Method string `json:"method"`

	// Object to pass as request parameter to the method.
	Params *json.RawMessage `json:"params,omitempty"`
}

// ----------------------------------------------------------------------------
// Codec
// ----------------------------------------------------------------------------

// NewCustomClientCodec returns a new JSON Codec based on passed encoder selector.
func NewCustomClientCodec(encSel encode.EncoderSelector) *ClientCodec {
	return &ClientCodec{encSel: encSel}
}

// NewClientCodec returns a new JSON Codec.
func NewClientCodec() *ClientCodec {
	return NewCustomClientCodec(encode.DefaultEncoderSelector)
}

// ClientCodec creates a ClientCodecRequest to process each request.
type ClientCodec struct {
	encSel encode.EncoderSelector
}

func (cc *ClientCodec) Write(w io.Writer, method string, args interface{}, id interface{}) error {
	req := retainClientRequest(method, args, id)
	defer func() {
		if nil != req {
			releaseClientRequest(req)
		}
	}()

	encoder := json.NewEncoder(cc.encSel.SelectByWriter(w).Encode(w))
	if err := encoder.Encode(req); nil != err {
		return err
	}

	return nil
}

// NewResponse returns a ClientCodecResponse.
func (cc *ClientCodec) NewResponseOrNotify(r io.Reader) (protocol.ClientCodecResponseOrNotify, error) {
	return newClientCodecResponseOrNotify(r, cc.encSel.SelectByReader(r))
}

// newCodecRequest returns a new ServerCodecRequest.
func newClientCodecResponseOrNotify(r io.Reader, encoder encode.Encoder) (protocol.ClientCodecResponseOrNotify, error) {
	// Decode the request body and check if RPC method is valid.
	var raw json.RawMessage
	dec := json.NewDecoder(r)
	err := dec.Decode(&raw)

	if err == io.ErrUnexpectedEOF || err == io.EOF {
		return nil, err
	}
	if err != nil {
		err = &Error{
			Code:    E_PARSE,
			Message: err.Error(),
			Data:    raw,
		}
	}

	ccrn := retainClientCodecResponseOrNotify()

	if notify, err := newClientCodecNotify(raw); nil != err {
		res, err := newClientCodecResponse(raw)
		if nil != err {
			releaseClientCodecResponseOrNotify(ccrn)
			return nil, fmt.Errorf("Is not response or notification [%v]", raw)
		}
		ccrn.response = res
	} else {
		ccrn.notify = notify
	}

	// if res, err := newClientCodecResponse(raw); nil != err {
	// 	notify, err := newClientCodecNotify(raw)
	// 	if nil != err {
	// 		releaseClientCodecResponseOrNotify(ccrn)
	// 		return nil, fmt.Errorf("Is not response or notification [%v]", raw)
	// 	}
	// 	ccrn.notify = notify
	// } else {
	// 	ccrn.response = res
	// }

	return ccrn, nil
}

type ClientCodecResponseOrNotify struct {
	notify   protocol.ClientCodecNotify
	response protocol.ClientCodecResponse
}

func (ccrn *ClientCodecResponseOrNotify) IsResponse() bool {
	return nil != ccrn.response
}

func (ccrn *ClientCodecResponseOrNotify) IsNotify() bool {
	return nil != ccrn.notify
}

func (ccrn *ClientCodecResponseOrNotify) GetResponse() protocol.ClientCodecResponse {
	return ccrn.response
}

func (ccrn *ClientCodecResponseOrNotify) GetNotify() protocol.ClientCodecNotify {
	return ccrn.notify
}

func (ccrn *ClientCodecResponseOrNotify) Complete() {
	if nil != ccrn.notify {
		ccrn.notify.Complete()
	}
	if nil != ccrn.response {
		ccrn.response.Complete()
	}
	releaseClientCodecResponseOrNotify(ccrn)
}

var clientRequestPool sync.Pool

func retainClientRequest(method string, params interface{}, id interface{}) *clientRequest {
	var cr *clientRequest
	v := clientRequestPool.Get()
	if v == nil {
		cr = &clientRequest{}
	} else {
		cr = v.(*clientRequest)
	}

	cr.Version = Version
	cr.Method = method
	cr.Params = params
	cr.ID = id

	return cr
}

func releaseClientRequest(cr *clientRequest) {
	cr.Version = ""
	cr.Method = ""
	cr.Params = nil
	cr.ID = nil

	clientRequestPool.Put(cr)
}

var clientCodecResponseOrNotifyPool sync.Pool

func retainClientCodecResponseOrNotify() *ClientCodecResponseOrNotify {
	v := clientCodecResponseOrNotifyPool.Get()
	if v == nil {
		return &ClientCodecResponseOrNotify{}
	}
	return v.(*ClientCodecResponseOrNotify)
}

func releaseClientCodecResponseOrNotify(cr *ClientCodecResponseOrNotify) {

	clientCodecResponseOrNotifyPool.Put(cr)
}