plugeth/peer.go

149 lines
3.0 KiB
Go
Raw Normal View History

2014-01-09 22:15:51 +00:00
package main
import (
2014-01-11 14:27:08 +00:00
"github.com/ethereum/ethwire-go"
2014-01-12 15:50:33 +00:00
"github.com/ethereum/ethutil-go"
2014-01-11 14:27:08 +00:00
"log"
"net"
2014-01-09 22:15:51 +00:00
)
type Peer struct {
2014-01-11 14:27:08 +00:00
// Server interface
server *Server
// Net connection
conn net.Conn
// Output queue which is used to communicate and handle messages
2014-01-12 15:50:33 +00:00
outputQueue chan *ethwire.InOutMsg
2014-01-11 14:27:08 +00:00
// Quit channel
quit chan bool
2014-01-12 15:50:33 +00:00
inbound bool // Determines whether it's an inbound or outbound peer
2014-01-09 22:15:51 +00:00
}
2014-01-12 15:50:33 +00:00
func NewPeer(conn net.Conn, server *Server, inbound bool) *Peer {
2014-01-11 14:27:08 +00:00
return &Peer{
2014-01-12 15:50:33 +00:00
outputQueue: make(chan *ethwire.InOutMsg, 1), // Buffered chan of 1 is enough
2014-01-11 14:27:08 +00:00
quit: make(chan bool),
server: server,
conn: conn,
2014-01-12 15:50:33 +00:00
inbound: inbound,
2014-01-11 14:27:08 +00:00
}
2014-01-09 22:15:51 +00:00
}
// Outputs any RLP encoded data to the peer
2014-01-12 15:50:33 +00:00
func (p *Peer) QueueMessage(msg *ethwire.InOutMsg) {
p.outputQueue <- msg//ethwire.InOutMsg{MsgType: msgType, Nonce: ethutil.RandomUint64(), Data: data}
2014-01-09 22:15:51 +00:00
}
2014-01-10 09:58:46 +00:00
// Outbound message handler. Outbound messages are handled here
2014-01-09 22:15:51 +00:00
func (p *Peer) HandleOutbound() {
out:
2014-01-11 14:27:08 +00:00
for {
select {
// Main message queue. All outbound messages are processed through here
case msg := <-p.outputQueue:
// TODO Message checking and handle accordingly
err := ethwire.WriteMessage(p.conn, msg)
if err != nil {
log.Println(err)
// Stop the client if there was an error writing to it
p.Stop()
}
// Break out of the for loop if a quit message is posted
case <-p.quit:
break out
}
}
2014-01-09 22:15:51 +00:00
}
2014-01-10 09:58:46 +00:00
// Inbound handler. Inbound messages are received here and passed to the appropriate methods
2014-01-09 22:15:51 +00:00
func (p *Peer) HandleInbound() {
2014-01-11 14:27:08 +00:00
defer p.Stop()
2014-01-09 22:15:51 +00:00
out:
2014-01-11 14:27:08 +00:00
for {
// Wait for a message from the peer
msg, err := ethwire.ReadMessage(p.conn)
if err != nil {
log.Println(err)
break out
}
2014-01-12 15:50:33 +00:00
if Debug {
log.Printf("Received %s\n", msg.MsgType)
}
// TODO Hash data and check if for existence (= ignore)
switch msg.MsgType {
case "verack":
// Version message
p.handleVersionAck(msg)
case "block":
err := p.server.blockManager.ProcessBlock(ethutil.NewBlock(msg.Data))
if err != nil {
log.Println(err)
}
}
2014-01-11 14:27:08 +00:00
}
// Notify the out handler we're quiting
p.quit <- true
2014-01-09 22:15:51 +00:00
}
func (p *Peer) Start() {
2014-01-12 15:50:33 +00:00
if !p.inbound {
err := p.pushVersionAck()
if err != nil {
log.Printf("Peer can't send outbound version ack", err)
p.Stop()
}
}
2014-01-11 14:27:08 +00:00
// Run the outbound handler in a new goroutine
go p.HandleOutbound()
// Run the inbound handler in a new goroutine
go p.HandleInbound()
2014-01-09 22:15:51 +00:00
}
func (p *Peer) Stop() {
2014-01-11 14:27:08 +00:00
p.conn.Close()
2014-01-11 14:27:08 +00:00
p.quit <- true
}
2014-01-12 15:50:33 +00:00
func (p *Peer) pushVersionAck() error {
msg := ethwire.NewMessage("verack", p.server.Nonce, []byte("01"))
p.QueueMessage(msg)
return nil
}
func (p *Peer) handleVersionAck(msg *ethwire.InOutMsg) {
// Detect self connect
if msg.Nonce == p.server.Nonce {
log.Println("Peer connected to self, disconnecting")
p.Stop()
return
}
log.Println("mnonce", msg.Nonce, "snonce", p.server.Nonce)
// If this is an inbound connection send an ack back
if p.inbound {
err := p.pushVersionAck()
if err != nil {
log.Println("Peer can't send ack back")
p.Stop()
}
}
}