aboutsummaryrefslogtreecommitdiffstats
path: root/rpc
diff options
context:
space:
mode:
authorBas van Kervel <bas@ethdev.com>2015-06-25 18:01:28 +0800
committerBas van Kervel <bas@ethdev.com>2015-06-25 18:01:28 +0800
commit6d92fdc0df12337c2ffb8e6d19c83653f1a00ff2 (patch)
treeb0717ba3da3f9db94a062536dc4382cc16477ec1 /rpc
parent22c7ce0162f2d14a7340e00e93697780c91d2087 (diff)
downloaddexon-6d92fdc0df12337c2ffb8e6d19c83653f1a00ff2.tar
dexon-6d92fdc0df12337c2ffb8e6d19c83653f1a00ff2.tar.gz
dexon-6d92fdc0df12337c2ffb8e6d19c83653f1a00ff2.tar.bz2
dexon-6d92fdc0df12337c2ffb8e6d19c83653f1a00ff2.tar.lz
dexon-6d92fdc0df12337c2ffb8e6d19c83653f1a00ff2.tar.xz
dexon-6d92fdc0df12337c2ffb8e6d19c83653f1a00ff2.tar.zst
dexon-6d92fdc0df12337c2ffb8e6d19c83653f1a00ff2.zip
added support for batch requests
Diffstat (limited to 'rpc')
-rw-r--r--rpc/codec/codec.go2
-rw-r--r--rpc/codec/json.go59
-rw-r--r--rpc/comms/comms.go38
3 files changed, 78 insertions, 21 deletions
diff --git a/rpc/codec/codec.go b/rpc/codec/codec.go
index 5e8f38438..3177f77e4 100644
--- a/rpc/codec/codec.go
+++ b/rpc/codec/codec.go
@@ -12,7 +12,7 @@ type Codec int
// (de)serialization support for rpc interface
type ApiCoder interface {
// Parse message to request from underlying stream
- ReadRequest() (*shared.Request, error)
+ ReadRequest() ([]*shared.Request, bool, error)
// Parse response message from underlying stream
ReadResponse() (interface{}, error)
// Encode response to encoded form in underlying stream
diff --git a/rpc/codec/json.go b/rpc/codec/json.go
index 31024ee74..380b4cba7 100644
--- a/rpc/codec/json.go
+++ b/rpc/codec/json.go
@@ -8,33 +8,53 @@ import (
)
const (
- MAX_RESPONSE_SIZE = 64 * 1024
+ MAX_REQUEST_SIZE = 1024 * 1024
+ MAX_RESPONSE_SIZE = 1024 * 1024
)
// Json serialization support
type JsonCodec struct {
c net.Conn
- d *json.Decoder
- e *json.Encoder
+ buffer []byte
+ bytesInBuffer int
}
// Create new JSON coder instance
func NewJsonCoder(conn net.Conn) ApiCoder {
return &JsonCodec{
c: conn,
- d: json.NewDecoder(conn),
- e: json.NewEncoder(conn),
+ buffer: make([]byte, MAX_REQUEST_SIZE),
+ bytesInBuffer: 0,
}
}
// Serialize obj to JSON and write it to conn
-func (self *JsonCodec) ReadRequest() (*shared.Request, error) {
- req := shared.Request{}
- err := self.d.Decode(&req)
+func (self *JsonCodec) ReadRequest() (requests []*shared.Request, isBatch bool, err error) {
+ n, err := self.c.Read(self.buffer[self.bytesInBuffer:])
+ if err != nil {
+ self.bytesInBuffer = 0
+ return nil, false, err
+ }
+
+ self.bytesInBuffer += n
+
+ singleRequest := shared.Request{}
+ err = json.Unmarshal(self.buffer[:self.bytesInBuffer], &singleRequest)
if err == nil {
- return &req, nil
+ self.bytesInBuffer = 0
+ requests := make([]*shared.Request, 1)
+ requests[0] = &singleRequest
+ return requests, false, nil
}
- return nil, err
+
+ requests = make([]*shared.Request, 0)
+ err = json.Unmarshal(self.buffer[:self.bytesInBuffer], &requests)
+ if err == nil {
+ self.bytesInBuffer = 0
+ return requests, true, nil
+ }
+
+ return nil, false, err
}
func (self *JsonCodec) ReadResponse() (interface{}, error) {
@@ -66,7 +86,24 @@ func (self *JsonCodec) Encode(msg interface{}) ([]byte, error) {
// Parse JSON data from conn to obj
func (self *JsonCodec) WriteResponse(res interface{}) error {
- return self.e.Encode(&res)
+ data, err := json.Marshal(res)
+ if err != nil {
+ self.c.Close()
+ return err
+ }
+
+ bytesWritten := 0
+
+ for bytesWritten < len(data) {
+ n, err := self.c.Write(data[bytesWritten:])
+ if err != nil {
+ self.c.Close()
+ return err
+ }
+ bytesWritten += n
+ }
+
+ return nil
}
// Close decoder and encoder
diff --git a/rpc/comms/comms.go b/rpc/comms/comms.go
index bfe625758..1374bde3f 100644
--- a/rpc/comms/comms.go
+++ b/rpc/comms/comms.go
@@ -47,7 +47,7 @@ func handle(conn net.Conn, api shared.EthereumApi, c codec.Codec) {
codec := c.New(conn)
for {
- req, err := codec.ReadRequest()
+ requests, isBatch, err := codec.ReadRequest()
if err == io.EOF {
codec.Close()
return
@@ -57,15 +57,35 @@ func handle(conn net.Conn, api shared.EthereumApi, c codec.Codec) {
return
}
- var rpcResponse interface{}
- res, err := api.Execute(req)
+ if isBatch {
+ responses := make([]*interface{}, len(requests))
+ responseCount := 0
+ for _, req := range requests {
+ res, err := api.Execute(req)
+ if req.Id != nil {
+ rpcResponse := shared.NewRpcResponse(req.Id, req.Jsonrpc, res, err)
+ responses[responseCount] = rpcResponse
+ responseCount += 1
+ }
+ }
- rpcResponse = shared.NewRpcResponse(req.Id, req.Jsonrpc, res, err)
- err = codec.WriteResponse(rpcResponse)
- if err != nil {
- glog.V(logger.Error).Infof("comms send err - %v\n", err)
- codec.Close()
- return
+ err = codec.WriteResponse(responses[:responseCount])
+ if err != nil {
+ glog.V(logger.Error).Infof("comms send err - %v\n", err)
+ codec.Close()
+ return
+ }
+ } else {
+ var rpcResponse interface{}
+ res, err := api.Execute(requests[0])
+
+ rpcResponse = shared.NewRpcResponse(requests[0].Id, requests[0].Jsonrpc, res, err)
+ err = codec.WriteResponse(rpcResponse)
+ if err != nil {
+ glog.V(logger.Error).Infof("comms send err - %v\n", err)
+ codec.Close()
+ return
+ }
}
}
}