1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247
|
package jsonrpc
import (
"bytes"
"context"
"encoding/json"
"io/ioutil"
"net/http"
"net/url"
"sync/atomic"
"github.com/go-kit/kit/endpoint"
httptransport "github.com/go-kit/kit/transport/http"
)
// Client wraps a JSON RPC method and provides a method that implements endpoint.Endpoint.
type Client struct {
client httptransport.HTTPClient
// JSON RPC endpoint URL
tgt *url.URL
// JSON RPC method name.
method string
enc EncodeRequestFunc
dec DecodeResponseFunc
before []httptransport.RequestFunc
after []httptransport.ClientResponseFunc
finalizer httptransport.ClientFinalizerFunc
requestID RequestIDGenerator
bufferedStream bool
}
type clientRequest struct {
JSONRPC string `json:"jsonrpc"`
Method string `json:"method"`
Params json.RawMessage `json:"params"`
ID interface{} `json:"id"`
}
// NewClient constructs a usable Client for a single remote method.
func NewClient(
tgt *url.URL,
method string,
options ...ClientOption,
) *Client {
c := &Client{
client: http.DefaultClient,
method: method,
tgt: tgt,
enc: DefaultRequestEncoder,
dec: DefaultResponseDecoder,
before: []httptransport.RequestFunc{},
after: []httptransport.ClientResponseFunc{},
requestID: NewAutoIncrementID(0),
bufferedStream: false,
}
for _, option := range options {
option(c)
}
return c
}
// DefaultRequestEncoder marshals the given request to JSON.
func DefaultRequestEncoder(_ context.Context, req interface{}) (json.RawMessage, error) {
return json.Marshal(req)
}
// DefaultResponseDecoder unmarshals the result to interface{}, or returns an
// error, if found.
func DefaultResponseDecoder(_ context.Context, res Response) (interface{}, error) {
if res.Error != nil {
return nil, *res.Error
}
var result interface{}
err := json.Unmarshal(res.Result, &result)
if err != nil {
return nil, err
}
return result, nil
}
// ClientOption sets an optional parameter for clients.
type ClientOption func(*Client)
// SetClient sets the underlying HTTP client used for requests.
// By default, http.DefaultClient is used.
func SetClient(client httptransport.HTTPClient) ClientOption {
return func(c *Client) { c.client = client }
}
// ClientBefore sets the RequestFuncs that are applied to the outgoing HTTP
// request before it's invoked.
func ClientBefore(before ...httptransport.RequestFunc) ClientOption {
return func(c *Client) { c.before = append(c.before, before...) }
}
// ClientAfter sets the ClientResponseFuncs applied to the server's HTTP
// response prior to it being decoded. This is useful for obtaining anything
// from the response and adding onto the context prior to decoding.
func ClientAfter(after ...httptransport.ClientResponseFunc) ClientOption {
return func(c *Client) { c.after = append(c.after, after...) }
}
// ClientFinalizer is executed at the end of every HTTP request.
// By default, no finalizer is registered.
func ClientFinalizer(f httptransport.ClientFinalizerFunc) ClientOption {
return func(c *Client) { c.finalizer = f }
}
// ClientRequestEncoder sets the func used to encode the request params to JSON.
// If not set, DefaultRequestEncoder is used.
func ClientRequestEncoder(enc EncodeRequestFunc) ClientOption {
return func(c *Client) { c.enc = enc }
}
// ClientResponseDecoder sets the func used to decode the response params from
// JSON. If not set, DefaultResponseDecoder is used.
func ClientResponseDecoder(dec DecodeResponseFunc) ClientOption {
return func(c *Client) { c.dec = dec }
}
// RequestIDGenerator returns an ID for the request.
type RequestIDGenerator interface {
Generate() interface{}
}
// ClientRequestIDGenerator is executed before each request to generate an ID
// for the request.
// By default, AutoIncrementRequestID is used.
func ClientRequestIDGenerator(g RequestIDGenerator) ClientOption {
return func(c *Client) { c.requestID = g }
}
// BufferedStream sets whether the Response.Body is left open, allowing it
// to be read from later. Useful for transporting a file as a buffered stream.
func BufferedStream(buffered bool) ClientOption {
return func(c *Client) { c.bufferedStream = buffered }
}
// Endpoint returns a usable endpoint that invokes the remote endpoint.
func (c Client) Endpoint() endpoint.Endpoint {
return func(ctx context.Context, request interface{}) (interface{}, error) {
ctx, cancel := context.WithCancel(ctx)
defer cancel()
var (
resp *http.Response
err error
)
if c.finalizer != nil {
defer func() {
if resp != nil {
ctx = context.WithValue(ctx, httptransport.ContextKeyResponseHeaders, resp.Header)
ctx = context.WithValue(ctx, httptransport.ContextKeyResponseSize, resp.ContentLength)
}
c.finalizer(ctx, err)
}()
}
ctx = context.WithValue(ctx, ContextKeyRequestMethod, c.method)
var params json.RawMessage
if params, err = c.enc(ctx, request); err != nil {
return nil, err
}
rpcReq := clientRequest{
JSONRPC: Version,
Method: c.method,
Params: params,
ID: c.requestID.Generate(),
}
req, err := http.NewRequest("POST", c.tgt.String(), nil)
if err != nil {
return nil, err
}
req.Header.Set("Content-Type", "application/json; charset=utf-8")
var b bytes.Buffer
req.Body = ioutil.NopCloser(&b)
err = json.NewEncoder(&b).Encode(rpcReq)
if err != nil {
return nil, err
}
for _, f := range c.before {
ctx = f(ctx, req)
}
resp, err = c.client.Do(req.WithContext(ctx))
if err != nil {
return nil, err
}
if !c.bufferedStream {
defer resp.Body.Close()
}
for _, f := range c.after {
ctx = f(ctx, resp)
}
// Decode the body into an object
var rpcRes Response
err = json.NewDecoder(resp.Body).Decode(&rpcRes)
if err != nil {
return nil, err
}
response, err := c.dec(ctx, rpcRes)
if err != nil {
return nil, err
}
return response, nil
}
}
// ClientFinalizerFunc can be used to perform work at the end of a client HTTP
// request, after the response is returned. The principal
// intended use is for error logging. Additional response parameters are
// provided in the context under keys with the ContextKeyResponse prefix.
// Note: err may be nil. There maybe also no additional response parameters
// depending on when an error occurs.
type ClientFinalizerFunc func(ctx context.Context, err error)
// autoIncrementID is a RequestIDGenerator that generates
// auto-incrementing integer IDs.
type autoIncrementID struct {
v *uint64
}
// NewAutoIncrementID returns an auto-incrementing request ID generator,
// initialised with the given value.
func NewAutoIncrementID(init uint64) RequestIDGenerator {
// Offset by one so that the first generated value = init.
v := init - 1
return &autoIncrementID{v: &v}
}
// Generate satisfies RequestIDGenerator
func (i *autoIncrementID) Generate() interface{} {
id := atomic.AddUint64(i.v, 1)
return id
}
|