aboutsummaryrefslogtreecommitdiffstats
path: root/ethwire
diff options
context:
space:
mode:
authorobscuren <geffobscura@gmail.com>2014-09-16 22:06:38 +0800
committerobscuren <geffobscura@gmail.com>2014-09-16 22:06:38 +0800
commit74de0f1f2ab342466556baddbab166a284f86891 (patch)
treeac875b39888928651d3f07033576212cb5cef344 /ethwire
parentd02024929a3e617c16c5b33e1d245f23fa4f6c7b (diff)
downloadgo-tangerine-74de0f1f2ab342466556baddbab166a284f86891.tar.gz
go-tangerine-74de0f1f2ab342466556baddbab166a284f86891.tar.zst
go-tangerine-74de0f1f2ab342466556baddbab166a284f86891.zip
Rewrote reading strategy
Diffstat (limited to 'ethwire')
-rw-r--r--ethwire/messaging.go241
1 files changed, 38 insertions, 203 deletions
diff --git a/ethwire/messaging.go b/ethwire/messaging.go
index bee6dd526..99f6be8db 100644
--- a/ethwire/messaging.go
+++ b/ethwire/messaging.go
@@ -4,7 +4,6 @@ package ethwire
import (
"bytes"
- "errors"
"fmt"
"net"
"time"
@@ -78,106 +77,10 @@ func NewMessage(msgType MsgType, data interface{}) *Msg {
type Messages []*Msg
-// The connection object allows you to set up a connection to the Ethereum network.
-// The Connection object takes care of all encoding and sending objects properly over
-// the network.
-type Connection struct {
- conn net.Conn
- nTimeout time.Duration
- pendingMessages Messages
-}
-
-// Create a new connection to the Ethereum network
-func New(conn net.Conn) *Connection {
- return &Connection{conn: conn, nTimeout: 500}
-}
-
-// Read, reads from the network. It will block until the next message is received.
-func (self *Connection) Read() *Msg {
- if len(self.pendingMessages) == 0 {
- self.readMessages()
- }
-
- ret := self.pendingMessages[0]
- self.pendingMessages = self.pendingMessages[1:]
-
- return ret
-
-}
-
-// Write to the Ethereum network specifying the type of the message and
-// the data. Data can be of type RlpEncodable or []interface{}. Returns
-// nil or if something went wrong an error.
-func (self *Connection) Write(typ MsgType, v ...interface{}) error {
- var pack []byte
-
- slice := [][]interface{}{[]interface{}{byte(typ)}}
- for _, value := range v {
- if encodable, ok := value.(ethutil.RlpEncodeDecode); ok {
- slice = append(slice, encodable.RlpValue())
- } else if raw, ok := value.([]interface{}); ok {
- slice = append(slice, raw)
- } else {
- panic(fmt.Sprintf("Unable to 'write' object of type %T", value))
- }
- }
-
- // Encode the type and the (RLP encoded) data for sending over the wire
- encoded := ethutil.NewValue(slice).Encode()
- payloadLength := ethutil.NumberToBytes(uint32(len(encoded)), 32)
-
- // Write magic token and payload length (first 8 bytes)
- pack = append(MagicToken, payloadLength...)
- pack = append(pack, encoded...)
-
- // Write to the connection
- _, err := self.conn.Write(pack)
- if err != nil {
- return err
- }
-
- return nil
-}
-
-func (self *Connection) readMessage(data []byte) (msg *Msg, remaining []byte, done bool, err error) {
- if len(data) == 0 {
- return nil, nil, true, nil
- }
-
- if len(data) <= 8 {
- return nil, remaining, false, errors.New("Invalid message")
- }
-
- // Check if the received 4 first bytes are the magic token
- if bytes.Compare(MagicToken, data[:4]) != 0 {
- return nil, nil, false, fmt.Errorf("MagicToken mismatch. Received %v", data[:4])
- }
-
- messageLength := ethutil.BytesToNumber(data[4:8])
- remaining = data[8+messageLength:]
- if int(messageLength) > len(data[8:]) {
- return nil, nil, false, fmt.Errorf("message length %d, expected %d", len(data[8:]), messageLength)
- }
-
- message := data[8 : 8+messageLength]
- decoder := ethutil.NewValueFromBytes(message)
- // Type of message
- t := decoder.Get(0).Uint()
- // Actual data
- d := decoder.SliceFrom(1)
-
- msg = &Msg{
- Type: MsgType(t),
- Data: d,
- }
-
- return
-}
-
// The basic message reader waits for data on the given connection, decoding
// and doing a few sanity checks such as if there's a data type and
// unmarhals the given data
-func (self *Connection) readMessages() (err error) {
+func ReadMessages(conn net.Conn) (msgs []*Msg, err error) {
// The recovering function in case anything goes horribly wrong
defer func() {
if r := recover(); r != nil {
@@ -185,137 +88,69 @@ func (self *Connection) readMessages() (err error) {
}
}()
- // Buff for writing network message to
- //buff := make([]byte, 1440)
- var buff []byte
- var totalBytes int
+ var (
+ buff []byte
+ messages [][]byte
+ msgLength int
+ )
+
for {
// Give buffering some time
- self.conn.SetReadDeadline(time.Now().Add(self.nTimeout * time.Millisecond))
+ conn.SetReadDeadline(time.Now().Add(5 * time.Millisecond))
// Create a new temporarily buffer
b := make([]byte, 1440)
- // Wait for a message from this peer
- n, _ := self.conn.Read(b)
+ n, _ := conn.Read(b)
if err != nil && n == 0 {
if err.Error() != "EOF" {
fmt.Println("err now", err)
- return err
+ return nil, err
} else {
break
}
-
- // Messages can't be empty
- } else if n == 0 {
- break
}
- buff = append(buff, b[:n]...)
- totalBytes += n
- }
-
- // Reslice buffer
- buff = buff[:totalBytes]
- msg, remaining, done, err := self.readMessage(buff)
- for ; done != true; msg, remaining, done, err = self.readMessage(remaining) {
- //log.Println("rx", msg)
-
- if msg != nil {
- self.pendingMessages = append(self.pendingMessages, msg)
+ if n == 0 {
+ continue
}
- }
-
- return
-}
-
-func ReadMessage(data []byte) (msg *Msg, remaining []byte, done bool, err error) {
- if len(data) == 0 {
- return nil, nil, true, nil
- }
-
- if len(data) <= 8 {
- return nil, remaining, false, errors.New("Invalid message")
- }
-
- // Check if the received 4 first bytes are the magic token
- if bytes.Compare(MagicToken, data[:4]) != 0 {
- return nil, nil, false, fmt.Errorf("MagicToken mismatch. Received %v", data[:4])
- }
- messageLength := ethutil.BytesToNumber(data[4:8])
- remaining = data[8+messageLength:]
- if int(messageLength) > len(data[8:]) {
- return nil, nil, false, fmt.Errorf("message length %d, expected %d", len(data[8:]), messageLength)
- }
+ if msgLength == 0 {
+ // Check if the received 4 first bytes are the magic token
+ if bytes.Compare(MagicToken, b[:4]) != 0 {
+ return nil, fmt.Errorf("MagicToken mismatch. Received %v", b[:4])
+ }
- message := data[8 : 8+messageLength]
- decoder := ethutil.NewValueFromBytes(message)
- // Type of message
- t := decoder.Get(0).Uint()
- // Actual data
- d := decoder.SliceFrom(1)
+ // Remove the token
+ b = b[4:]
+ // Read the length of the message
+ msgLength = int(ethutil.BytesToNumber(b[:4]))
- msg = &Msg{
- Type: MsgType(t),
- Data: d,
- }
+ // Remove the length
+ b = b[4:]
- return
-}
+ n -= 8
+ }
-func bufferedRead(conn net.Conn) ([]byte, error) {
- return nil, nil
-}
+ buff = append(buff, b[:n]...)
-// The basic message reader waits for data on the given connection, decoding
-// and doing a few sanity checks such as if there's a data type and
-// unmarhals the given data
-func ReadMessages(conn net.Conn) (msgs []*Msg, err error) {
- // The recovering function in case anything goes horribly wrong
- defer func() {
- if r := recover(); r != nil {
- err = fmt.Errorf("ethwire.ReadMessage error: %v", r)
- }
- }()
+ if len(buff) >= msgLength {
+ messages = append(messages, buff[:msgLength])
+ buff = buff[msgLength:]
+ msgLength = 0
- // Buff for writing network message to
- //buff := make([]byte, 1440)
- var buff []byte
- var totalBytes int
- for {
- // This is a bit of a cheat actually to make buffering extremely fast.
- defer recover()
- // Give buffering some time
- conn.SetReadDeadline(time.Now().Add(5 * time.Millisecond))
- // Create a new temporarily buffer
- b := make([]byte, 1440)
- // Wait for a message from this peer
- n, _ := conn.Read(b)
- if err != nil && n == 0 {
- if err.Error() != "EOF" {
- fmt.Println("err now", err)
- return nil, err
- } else {
+ if len(buff) == 0 {
break
}
-
- // Messages can't be empty
- } else if n == 0 {
- break
}
-
- buff = append(buff, b[:n]...)
- totalBytes += n
}
- // Reslice buffer
- buff = buff[:totalBytes]
- msg, remaining, done, err := ReadMessage(buff)
- for ; done != true; msg, remaining, done, err = ReadMessage(remaining) {
- //log.Println("rx", msg)
+ for _, m := range messages {
+ decoder := ethutil.NewValueFromBytes(m)
+ // Type of message
+ t := decoder.Get(0).Uint()
+ // Actual data
+ d := decoder.SliceFrom(1)
- if msg != nil {
- msgs = append(msgs, msg)
- }
+ msgs = append(msgs, &Msg{Type: MsgType(t), Data: d})
}
return