aboutsummaryrefslogtreecommitdiffstats
path: root/rpc
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-02 23:25:11 +0800
commiteffe9cc2cf7203634b9b418e1e9309911d0f2b00 (patch)
tree83e9f1bc02efd521c69089b34e60124c60bda538 /rpc
parent9f6016e8773f951c5b81b30c0257350e2a9c8e5f (diff)
downloadgo-tangerine-effe9cc2cf7203634b9b418e1e9309911d0f2b00.tar
go-tangerine-effe9cc2cf7203634b9b418e1e9309911d0f2b00.tar.gz
go-tangerine-effe9cc2cf7203634b9b418e1e9309911d0f2b00.tar.bz2
go-tangerine-effe9cc2cf7203634b9b418e1e9309911d0f2b00.tar.lz
go-tangerine-effe9cc2cf7203634b9b418e1e9309911d0f2b00.tar.xz
go-tangerine-effe9cc2cf7203634b9b418e1e9309911d0f2b00.tar.zst
go-tangerine-effe9cc2cf7203634b9b418e1e9309911d0f2b00.zip
added pipelining support
Diffstat (limited to 'rpc')
-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) {