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
|
package main
import (
"net"
"errors"
"log"
)
type InMsg struct {
msgType string // Specifies how the encoded data should be interpreted
data []byte // RLP encoded data
}
type OutMsg struct {
msgType string
data []byte
}
func ReadMessage(conn net.Conn) (*InMsg, error) {
buff := make([]byte, 4069)
// Wait for a message from this peer
n, err := conn.Read(buff)
if err != nil {
return nil, err
} else if n == 0 {
return nil, errors.New("Empty message received")
}
// Read the header (MAX n)
// XXX The data specification is made up. This will change once more details have been released on the specification of the format
decoder := NewRlpDecoder(buff[:n])
t := decoder.Get(0).AsString()
// If the msgdata contains no data we throw an error and disconnect the peer
if t == "" {
return nil, errors.New("Data contained no data type")
}
return &InMsg{msgType: t, data: decoder.Get(1).AsBytes()}, nil
}
type Peer struct {
// Server interface
server *Server
// Net connection
conn net.Conn
// Output queue which is used to communicate and handle messages
outputQueue chan OutMsg
// Quit channel
quit chan bool
}
func NewPeer(conn net.Conn, server *Server) *Peer {
return &Peer{
outputQueue: make(chan OutMsg, 1), // Buffered chan of 1 is enough
quit: make(chan bool),
server: server,
conn: conn,
}
}
// Outputs any RLP encoded data to the peer
func (p *Peer) QueueMessage(msgType string, data []byte) {
p.outputQueue <- OutMsg{msgType: msgType, data: data}
}
// Outbound message handler. Outbound messages are handled here
func (p *Peer) HandleOutbound() {
out:
for {
select {
// Main message queue. All outbound messages are processed through here
case msg := <-p.outputQueue:
// TODO Message checking and handle accordingly
p.WriteMessage(msg)
// Break out of the for loop if a quit message is posted
case <- p.quit:
break out
}
}
}
// Write a message to the peer.
func (p *Peer) WriteMessage(msg OutMsg) {
// Encode the type and the (RLP encoded) data for sending over the wire
encoded := Encode([]interface{}{ msg.msgType, msg.data })
// Write to the connection
_, err := p.conn.Write(encoded)
if err != nil {
log.Println(err)
p.Stop()
}
}
// Inbound handler. Inbound messages are received here and passed to the appropriate methods
func (p *Peer) HandleInbound() {
defer p.Stop()
out:
for {
// Wait for a message from the peer
msg, err := ReadMessage(p.conn)
if err != nil {
log.Println(err)
break out
}
// TODO
data, _ := Decode(msg.data, 0)
log.Printf("%s, %s\n", msg.msgType, data)
}
// Notify the out handler we're quiting
p.quit <- true
}
func (p *Peer) Start() {
// Run the outbound handler in a new goroutine
go p.HandleOutbound()
// Run the inbound handler in a new goroutine
go p.HandleInbound()
}
func (p *Peer) Stop() {
defer p.conn.Close()
p.quit <- true
}
|
