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
|
package codec
import (
"encoding/json"
"fmt"
"net"
"time"
"github.com/ethereum/go-ethereum/rpc/shared"
)
const (
READ_TIMEOUT = 15 // read timeout in seconds
MAX_REQUEST_SIZE = 1024 * 1024
MAX_RESPONSE_SIZE = 1024 * 1024
)
// 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) {
deadline := time.Now().Add(READ_TIMEOUT * time.Second)
if err := self.c.SetDeadline(deadline); err != nil {
return nil, false, err
}
for {
var err error
singleRequest := shared.Request{}
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)
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("Timeout reading request")
}
func (self *JsonCodec) ReadResponse() (interface{}, error) {
bytesInBuffer := 0
buf := make([]byte, MAX_RESPONSE_SIZE)
deadline := time.Now().Add(READ_TIMEOUT * time.Second)
if err := self.c.SetDeadline(deadline); err != nil {
return nil, err
}
for {
n, err := self.c.Read(buf[bytesInBuffer:])
if err != nil {
return nil, err
}
bytesInBuffer += n
var success shared.SuccessResponse
if err = json.Unmarshal(buf[:bytesInBuffer], &success); err == nil {
return success, nil
}
var failure shared.ErrorResponse
if err = json.Unmarshal(buf[:bytesInBuffer], &failure); err == nil && failure.Error != nil {
return failure, nil
}
}
self.c.Close()
return nil, fmt.Errorf("Unable to read response")
}
// Decode data
func (self *JsonCodec) Decode(data []byte, msg interface{}) error {
return json.Unmarshal(data, msg)
}
// Encode message
func (self *JsonCodec) Encode(msg interface{}) ([]byte, error) {
return json.Marshal(msg)
}
// Parse JSON data from conn to obj
func (self *JsonCodec) WriteResponse(res interface{}) error {
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
func (self *JsonCodec) Close() {
self.c.Close()
}
|
