aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorBas van Kervel <bas@ethdev.com>2015-07-01 14:23:17 +0800
committerBas van Kervel <bas@ethdev.com>2015-07-01 14:23:17 +0800
commit41de1cb723d2eff921da8911e4cf893d8907c178 (patch)
tree8798a60298e44a3805decf35eb2496cd506858de
parent130f3b270a9b2f7bd680188c67013695d5f9412b (diff)
downloaddexon-41de1cb723d2eff921da8911e4cf893d8907c178.tar
dexon-41de1cb723d2eff921da8911e4cf893d8907c178.tar.gz
dexon-41de1cb723d2eff921da8911e4cf893d8907c178.tar.bz2
dexon-41de1cb723d2eff921da8911e4cf893d8907c178.tar.lz
dexon-41de1cb723d2eff921da8911e4cf893d8907c178.tar.xz
dexon-41de1cb723d2eff921da8911e4cf893d8907c178.tar.zst
dexon-41de1cb723d2eff921da8911e4cf893d8907c178.zip
added pipelining support
-rw-r--r--rpc/codec/json.go35
1 files changed, 20 insertions, 15 deletions
diff --git a/rpc/codec/json.go b/rpc/codec/json.go
index 0b1a90562..5d60104f7 100644
--- a/rpc/codec/json.go
+++ b/rpc/codec/json.go
@@ -18,19 +18,19 @@ const (
// Json serialization support
type JsonCodec struct {
c net.Conn
+ d *json.Decoder
}
// Create new JSON coder instance
func NewJsonCoder(conn net.Conn) ApiCoder {
return &JsonCodec{
c: conn,
+ d: json.NewDecoder(conn),
}
}
// Serialize obj to JSON and write it to conn
func (self *JsonCodec) ReadRequest() (requests []*shared.Request, isBatch bool, err error) {
- bytesInBuffer := 0
- buf := make([]byte, MAX_REQUEST_SIZE)
deadline := time.Now().Add(READ_TIMEOUT * time.Second)
if err := self.c.SetDeadline(deadline); err != nil {
@@ -38,31 +38,36 @@ func (self *JsonCodec) ReadRequest() (requests []*shared.Request, isBatch bool,
}
for {
- n, err := self.c.Read(buf[bytesInBuffer:])
- if err != nil {
- self.c.Close()
- return nil, false, err
- }
-
- bytesInBuffer += n
-
+ var err error
singleRequest := shared.Request{}
- err = json.Unmarshal(buf[:bytesInBuffer], &singleRequest)
- if err == nil {
+ if err = self.d.Decode(&singleRequest); err == nil {
requests := make([]*shared.Request, 1)
requests[0] = &singleRequest
return requests, false, nil
}
+ fmt.Printf("err %T %v\n", err)
+
+ if opErr, ok := err.(*net.OpError); ok {
+ if opErr.Timeout() {
+ break
+ }
+ }
+
requests = make([]*shared.Request, 0)
- err = json.Unmarshal(buf[:bytesInBuffer], &requests)
- if err == nil {
+ if err = self.d.Decode(&requests); err == nil {
return requests, true, nil
}
+
+ if opErr, ok := err.(*net.OpError); ok {
+ if opErr.Timeout() {
+ break
+ }
+ }
}
self.c.Close() // timeout
- return nil, false, fmt.Errorf("Unable to read response")
+ return nil, false, fmt.Errorf("Timeout reading request")
}
func (self *JsonCodec) ReadResponse() (interface{}, error) {