Finished version + verack.

pkg/network/payload/getblocks.go

@@ -0,0 +1,54 @@
+package payload
+
+import (
+	"encoding/binary"
+	"io"
+
+	. "github.com/anthdm/neo-go/pkg/util"
+)
+
+// HashStartStop contains fields and methods to be shared with the
+// "GetBlocks" and "GetHeaders" payload.
+type HashStartStop struct {
+	// hash of latest block that node requests
+	HashStart []Uint256
+	// hash of last block that node requests
+	HashStop Uint256
+}
+
+// DecodeBinary implements the payload interface.
+func (p *HashStartStop) DecodeBinary(r io.Reader) error {
+	var lenStart uint8
+
+	err := binary.Read(r, binary.LittleEndian, &lenStart)
+	p.HashStart = make([]Uint256, lenStart)
+	err = binary.Read(r, binary.LittleEndian, &p.HashStart)
+	err = binary.Read(r, binary.LittleEndian, &p.HashStop)
+
+	return err
+}
+
+// EncodeBinary implements the payload interface.
+func (p *HashStartStop) EncodeBinary(w io.Writer) error {
+	err := binary.Write(w, binary.LittleEndian, uint8(len(p.HashStart)))
+	err = binary.Write(w, binary.LittleEndian, p.HashStart)
+	err = binary.Write(w, binary.LittleEndian, p.HashStop)
+
+	return err
+}
+
+// Size implements the payload interface.
+func (p *HashStartStop) Size() uint32 { return 0 }
+
+// GetBlocks payload
+type GetBlocks struct {
+	HashStartStop
+}
+
+// NewGetBlocks return a pointer to a GetBlocks object.
+func NewGetBlocks(start []Uint256, stop Uint256) *GetBlocks {
+	p := &GetBlocks{}
+	p.HashStart = start
+	p.HashStop = stop
+	return p
+}

pkg/network/payload/getblocks_test.go

@@ -0,0 +1,37 @@
+package payload
+
+import (
+	"bytes"
+	"crypto/sha256"
+	"reflect"
+	"testing"
+
+	. "github.com/anthdm/neo-go/pkg/util"
+)
+
+func TestGetBlocksEncodeDecode(t *testing.T) {
+	start := []Uint256{
+		sha256.Sum256([]byte("a")),
+		sha256.Sum256([]byte("b")),
+	}
+	stop := sha256.Sum256([]byte("c"))
+
+	p := NewGetBlocks(start, stop)
+	buf := new(bytes.Buffer)
+	if err := p.EncodeBinary(buf); err != nil {
+		t.Fatal(err)
+	}
+
+	if have, want := buf.Len(), 1+64+32; have != want {
+		t.Fatalf("expecting a length of %d got %d", want, have)
+	}
+
+	pDecode := &GetBlocks{}
+	if err := pDecode.DecodeBinary(buf); err != nil {
+		t.Fatal(err)
+	}
+
+	if !reflect.DeepEqual(p, pDecode) {
+		t.Fatalf("expecting both getblocks payloads to be equal %v and %v", p, pDecode)
+	}
+}

pkg/network/payload/getheaders.go

@@ -0,0 +1,17 @@
+package payload
+
+import "github.com/anthdm/neo-go/pkg/util"
+
+// GetHeaders payload is the same as the "GetBlocks" payload.
+type GetHeaders struct {
+	HashStartStop
+}
+
+// NewGetHeaders return a pointer to a GetHeaders object.
+func NewGetHeaders(start []util.Uint256, stop util.Uint256) *GetHeaders {
+	p := &GetHeaders{}
+	p.HashStart = start
+	p.HashStop = stop
+
+	return p
+}

pkg/network/payload/getheaders_test.go

@@ -0,0 +1,37 @@
+package payload
+
+import (
+	"bytes"
+	"crypto/sha256"
+	"reflect"
+	"testing"
+
+	"github.com/anthdm/neo-go/pkg/util"
+)
+
+func TestGetHeadersEncodeDecode(t *testing.T) {
+	start := []util.Uint256{
+		sha256.Sum256([]byte("a")),
+		sha256.Sum256([]byte("b")),
+	}
+	stop := sha256.Sum256([]byte("c"))
+
+	p := NewGetHeaders(start, stop)
+	buf := new(bytes.Buffer)
+	if err := p.EncodeBinary(buf); err != nil {
+		t.Fatal(err)
+	}
+
+	if have, want := buf.Len(), 1+64+32; have != want {
+		t.Fatalf("expecting a length of %d got %d", want, have)
+	}
+
+	pDecode := &GetHeaders{}
+	if err := pDecode.DecodeBinary(buf); err != nil {
+		t.Fatal(err)
+	}
+
+	if !reflect.DeepEqual(p, pDecode) {
+		t.Fatalf("expecting both getheaders payloads to be equal %v and %v", p, pDecode)
+	}
+}

pkg/network/peer.go

@@ -9,7 +9,6 @@ import (
 type Peer interface {
 	id() uint32
 	addr() util.Endpoint
-	verack() bool
 	disconnect()
 	callVersion(*Message)
 	callGetaddr(*Message)
@@ -20,7 +19,6 @@ type Peer interface {
 type LocalPeer struct {
 	s        *Server
 	nonce    uint32
-	isVerack bool
 	endpoint util.Endpoint
 }
 
@@ -39,6 +37,5 @@ func (p *LocalPeer) callGetaddr(msg *Message) {
 }
 
 func (p *LocalPeer) id() uint32          { return p.nonce }
-func (p *LocalPeer) verack() bool        { return p.isVerack }
 func (p *LocalPeer) addr() util.Endpoint { return p.endpoint }
 func (p *LocalPeer) disconnect()         {}

pkg/network/tcp.go

@@ -22,6 +22,7 @@ func listenTCP(s *Server, port int) error {
 		if err != nil {
 			return err
 		}
+
 		go handleConnection(s, conn)
 	}
 }
@@ -54,8 +55,10 @@ func handleConnection(s *Server, conn net.Conn) {
 		s.unregister <- peer
 	}()
 
-	// Start a goroutine that will handle all writes to the registered peer.
+	// Start a goroutine that will handle all outgoing messages.
 	go peer.writeLoop()
+	// Start a goroutine that will handle all incomming messages.
+	go handleMessage(s, peer)
 
 	// Read from the connection and decode it into a Message ready for processing.
 	buf := make([]byte, 1024)
@@ -71,39 +74,55 @@ func handleConnection(s *Server, conn net.Conn) {
 			s.logger.Printf("decode error %s", err)
 			break
 		}
-		handleMessage(msg, s, peer)
+
+		peer.receive <- msg
 	}
 }
 
 // handleMessage hands the message received from a TCP connection over to the server.
-func handleMessage(msg *Message, s *Server, p *TCPPeer) {
+func handleMessage(s *Server, p *TCPPeer) {
-	command := msg.commandType()
+	// Disconnect the peer when we break out of the loop.
+	defer func() {
+		p.disconnect()
+	}()
 
-	s.logger.Printf("IN :: %d :: %s :: %v", p.id(), command, msg)
+	for {
+		msg := <-p.receive
+		command := msg.commandType()
 
-	switch command {
+		s.logger.Printf("IN :: %d :: %s :: %v", p.id(), command, msg)
-	case cmdVersion:
+
-		resp := s.handleVersionCmd(msg, p)
+		switch command {
-		p.isVerack = true
+		case cmdVersion:
-		p.nonce = msg.Payload.(*payload.Version).Nonce
+			resp := s.handleVersionCmd(msg, p)
-		p.send <- resp
+			p.nonce = msg.Payload.(*payload.Version).Nonce
-	case cmdAddr:
+			p.send <- resp
-		s.handleAddrCmd(msg, p)
+
-	case cmdGetAddr:
+			// after sending our version we want a "verack" and nothing else.
-		s.handleGetaddrCmd(msg, p)
+			msg := <-p.receive
-	case cmdInv:
+			if msg.commandType() != cmdVerack {
-		resp := s.handleInvCmd(msg, p)
+				break
-		p.send <- resp
+			}
-	case cmdBlock:
+			// we can start the protocol now.
-	case cmdConsensus:
+			go s.sendLoop(p)
-	case cmdTX:
+		case cmdAddr:
-	case cmdVerack:
+			s.handleAddrCmd(msg, p)
-		go s.sendLoop(p)
+		case cmdGetAddr:
-	case cmdGetHeaders:
+			s.handleGetaddrCmd(msg, p)
-	case cmdGetBlocks:
+		case cmdInv:
-	case cmdGetData:
+			resp := s.handleInvCmd(msg, p)
-	case cmdHeaders:
+			p.send <- resp
-	default:
+		case cmdBlock:
+		case cmdConsensus:
+		case cmdTX:
+		case cmdVerack:
+			// disconnect the peer, verack should already be handled.
+			break
+		case cmdGetHeaders:
+		case cmdGetBlocks:
+		case cmdGetData:
+		case cmdHeaders:
+		}
 	}
 }
 
@@ -118,8 +137,8 @@ type TCPPeer struct {
 	endpoint util.Endpoint
 	// channel to coordinate messages writen back to the connection.
 	send chan *Message
-	// whether this peers version was acknowledged.
+	// channel to receive from underlying connection.
-	isVerack bool
+	receive chan *Message
 }
 
 // NewTCPPeer returns a pointer to a TCP Peer.
@@ -129,6 +148,7 @@ func NewTCPPeer(conn net.Conn, s *Server) *TCPPeer {
 	return &TCPPeer{
 		conn:     conn,
 		send:     make(chan *Message),
+		receive:  make(chan *Message),
 		endpoint: e,
 		s:        s,
 	}
@@ -148,11 +168,6 @@ func (p *TCPPeer) addr() util.Endpoint {
 	return p.endpoint
 }
 
-// verack implements the peer interface
-func (p *TCPPeer) verack() bool {
-	return p.isVerack
-}
-
 // callGetaddr will send the "getaddr" command to the remote.
 func (p *TCPPeer) callGetaddr(msg *Message) {
 	p.send <- msg