mirror of
https://github.com/nspcc-dev/neo-go.git
synced 2025-05-04 09:02:28 +00:00
refactor server RPC.
This commit is contained in:
anthdm
Anthony De Meulemeester
parent
283f3c5a89
commit
0eeb15f62d
4 changed files with 252 additions and 212 deletions
|
|
@ -1,7 +1,6 @@
|
|||
package network
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"net"
|
||||
|
|
@ -9,7 +8,6 @@ import (
|
|||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/anthdm/neo-go/pkg/core"
|
||||
"github.com/anthdm/neo-go/pkg/network/payload"
|
||||
"github.com/anthdm/neo-go/pkg/util"
|
||||
)
|
||||
|
|
@ -20,11 +18,7 @@ const (
|
|||
// official ports according to the protocol.
|
||||
portMainNet = 10333
|
||||
portTestNet = 20333
|
||||
)
|
||||
|
||||
var (
|
||||
// rpcLogger used for debugging RPC messages between nodes.
|
||||
rpcLogger = log.New(os.Stdout, "", 0)
|
||||
maxPeers = 50
|
||||
)
|
||||
|
||||
type messageTuple struct {
|
||||
|
|
@ -35,13 +29,10 @@ type messageTuple struct {
|
|||
// Server is the representation of a full working NEO TCP node.
|
||||
type Server struct {
|
||||
logger *log.Logger
|
||||
|
||||
// id of the server
|
||||
id uint32
|
||||
|
||||
// the port the TCP listener is listening on.
|
||||
port uint16
|
||||
|
||||
// userAgent of the server.
|
||||
userAgent string
|
||||
// The "magic" mode the server is currently running on.
|
||||
|
|
@ -50,26 +41,29 @@ type Server struct {
|
|||
net NetMode
|
||||
// map that holds all connected peers to this server.
|
||||
peers map[Peer]bool
|
||||
|
||||
register chan Peer
|
||||
// channel for handling new registerd peers.
|
||||
register chan Peer
|
||||
// channel for safely removing and disconnecting peers.
|
||||
unregister chan Peer
|
||||
|
||||
// channel for coordinating messages.
|
||||
message chan messageTuple
|
||||
|
||||
// channel used to gracefull shutdown the server.
|
||||
quit chan struct{}
|
||||
|
||||
// Whether this server will receive and forward messages.
|
||||
relay bool
|
||||
|
||||
// TCP listener of the server
|
||||
listener net.Listener
|
||||
|
||||
// RPC channels
|
||||
versionCh chan versionTuple
|
||||
getaddrCh chan getaddrTuple
|
||||
invCh chan invTuple
|
||||
addrCh chan addrTuple
|
||||
}
|
||||
|
||||
// NewServer returns a pointer to a new server.
|
||||
func NewServer(net NetMode) *Server {
|
||||
logger := log.New(os.Stdout, "NEO SERVER :: ", 0)
|
||||
logger := log.New(os.Stdout, "[NEO SERVER] :: ", 0)
|
||||
|
||||
if net != ModeTestNet && net != ModeMainNet && net != ModeDevNet {
|
||||
logger.Fatalf("invalid network mode %d", net)
|
||||
|
|
@ -83,9 +77,13 @@ func NewServer(net NetMode) *Server {
|
|||
register: make(chan Peer),
|
||||
unregister: make(chan Peer),
|
||||
message: make(chan messageTuple),
|
||||
relay: true,
|
||||
relay: true, // currently relay is not handled.
|
||||
net: net,
|
||||
quit: make(chan struct{}),
|
||||
versionCh: make(chan versionTuple),
|
||||
getaddrCh: make(chan getaddrTuple),
|
||||
invCh: make(chan invTuple),
|
||||
addrCh: make(chan addrTuple),
|
||||
}
|
||||
|
||||
return s
|
||||
|
|
@ -131,30 +129,62 @@ func (s *Server) shutdown() {
|
|||
func (s *Server) loop() {
|
||||
for {
|
||||
select {
|
||||
// When a new connection is been established, (by this server or remote node)
|
||||
// its peer will be received on this channel.
|
||||
// Any peer registration must happen via this channel.
|
||||
case peer := <-s.register:
|
||||
// When a new connection is been established, (by this server or remote node)
|
||||
// its peer will be received on this channel.
|
||||
// Any peer registration must happen via this channel.
|
||||
s.logger.Printf("peer registered from address %s", peer.endpoint())
|
||||
s.peers[peer] = true
|
||||
s.handlePeerConnected(peer)
|
||||
if len(s.peers) < maxPeers {
|
||||
s.logger.Printf("peer registered from address %s", peer.addr())
|
||||
s.peers[peer] = true
|
||||
s.handlePeerConnected(peer)
|
||||
}
|
||||
|
||||
// Unregister should take care of all the cleanup that has to be made.
|
||||
case peer := <-s.unregister:
|
||||
// unregister should take care of all the cleanup that has to be made.
|
||||
if _, ok := s.peers[peer]; ok {
|
||||
peer.disconnect()
|
||||
delete(s.peers, peer)
|
||||
s.logger.Printf("peer %s disconnected", peer.endpoint())
|
||||
s.logger.Printf("peer %s disconnected", peer.addr())
|
||||
}
|
||||
|
||||
case tuple := <-s.message:
|
||||
// When a remote node sends data over its connection it will be received
|
||||
// on this channel.
|
||||
// All errors encountered should be return and handled here.
|
||||
if err := s.processMessage(tuple.msg, tuple.peer); err != nil {
|
||||
s.logger.Fatalf("failed to process message: %s", err)
|
||||
s.unregister <- tuple.peer
|
||||
// Process the received version and respond with a verack.
|
||||
case t := <-s.versionCh:
|
||||
if s.id == t.request.Nonce {
|
||||
t.peer.disconnect()
|
||||
}
|
||||
if t.peer.addr().Port != t.request.Port {
|
||||
t.peer.disconnect()
|
||||
}
|
||||
t.response <- newMessage(ModeDevNet, cmdVerack, nil)
|
||||
|
||||
// Process the getaddr cmd.
|
||||
case t := <-s.getaddrCh:
|
||||
t.response <- &Message{} // just for now.
|
||||
|
||||
// Process the addr cmd. Register peer will handle the maxPeers connected.
|
||||
case t := <-s.addrCh:
|
||||
for _, addr := range t.request.Addrs {
|
||||
if !s.peerAlreadyConnected(addr.Addr) {
|
||||
// TODO: this is not transport abstracted.
|
||||
go connectToRemoteNode(s, addr.Addr.String())
|
||||
}
|
||||
}
|
||||
t.response <- true
|
||||
|
||||
// Process inventories cmd.
|
||||
case t := <-s.invCh:
|
||||
if !t.request.Type.Valid() {
|
||||
t.peer.disconnect()
|
||||
break
|
||||
}
|
||||
if len(t.request.Hashes) == 0 {
|
||||
t.peer.disconnect()
|
||||
break
|
||||
}
|
||||
|
||||
payload := payload.NewInventory(t.request.Type, t.request.Hashes)
|
||||
msg := newMessage(s.net, cmdGetData, payload)
|
||||
t.response <- msg
|
||||
|
||||
case <-s.quit:
|
||||
s.shutdown()
|
||||
|
|
@ -162,135 +192,100 @@ func (s *Server) loop() {
|
|||
}
|
||||
}
|
||||
|
||||
// processMessage processes the message received from the peer.
|
||||
func (s *Server) processMessage(msg *Message, peer Peer) error {
|
||||
command := msg.commandType()
|
||||
|
||||
rpcLogger.Printf("[NODE %d] :: IN :: %s :: %+v", peer.id(), command, msg.Payload)
|
||||
|
||||
// Disconnect if the remote is sending messages other then version
|
||||
// if we didn't verack this peer.
|
||||
if !peer.verack() && command != cmdVersion {
|
||||
return errors.New("version noack")
|
||||
}
|
||||
|
||||
switch command {
|
||||
case cmdVersion:
|
||||
return s.handleVersionCmd(msg.Payload.(*payload.Version), peer)
|
||||
case cmdVerack:
|
||||
case cmdGetAddr:
|
||||
// return s.handleGetAddrCmd(msg, peer)
|
||||
case cmdAddr:
|
||||
return s.handleAddrCmd(msg.Payload.(*payload.AddressList), peer)
|
||||
case cmdGetHeaders:
|
||||
case cmdHeaders:
|
||||
case cmdGetBlocks:
|
||||
case cmdInv:
|
||||
return s.handleInvCmd(msg.Payload.(*payload.Inventory), peer)
|
||||
case cmdGetData:
|
||||
case cmdBlock:
|
||||
return s.handleBlockCmd(msg.Payload.(*core.Block), peer)
|
||||
case cmdTX:
|
||||
case cmdConsensus:
|
||||
default:
|
||||
return fmt.Errorf("invalid RPC command received: %s", command)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// When a new peer is connected we send our version.
|
||||
// No further communication should be made before both sides has received
|
||||
// the versions of eachother.
|
||||
func (s *Server) handlePeerConnected(peer Peer) {
|
||||
// TODO get heigth of block when thats implemented.
|
||||
func (s *Server) handlePeerConnected(p Peer) {
|
||||
// TODO: get the blockheight of this server once core implemented this.
|
||||
payload := payload.NewVersion(s.id, s.port, s.userAgent, 0, s.relay)
|
||||
msg := newMessage(s.net, cmdVersion, payload)
|
||||
|
||||
peer.send(msg)
|
||||
p.callVersion(msg)
|
||||
}
|
||||
|
||||
// Version declares the server's version.
|
||||
func (s *Server) handleVersionCmd(v *payload.Version, peer Peer) error {
|
||||
if s.id == v.Nonce {
|
||||
return errors.New("remote nonce equal to server id")
|
||||
}
|
||||
|
||||
if peer.endpoint().Port != v.Port {
|
||||
return errors.New("port mismatch")
|
||||
}
|
||||
|
||||
// we respond with a verack, we successfully received peer's version
|
||||
// at this point.
|
||||
peer.verify(v.Nonce)
|
||||
verackMsg := newMessage(s.net, cmdVerack, nil)
|
||||
peer.send(verackMsg)
|
||||
|
||||
go s.sendLoop(peer)
|
||||
|
||||
return nil
|
||||
type versionTuple struct {
|
||||
peer Peer
|
||||
request *payload.Version
|
||||
response chan *Message
|
||||
}
|
||||
|
||||
// When the remote node reveals its known peers we try to connect to all of them.
|
||||
func (s *Server) handleAddrCmd(addrList *payload.AddressList, peer Peer) error {
|
||||
for _, addr := range addrList.Addrs {
|
||||
if !s.peerAlreadyConnected(addr.Addr) {
|
||||
go connectToRemoteNode(s, addr.Addr.String())
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Server) handleInvCmd(inv *payload.Inventory, peer Peer) error {
|
||||
if !inv.Type.Valid() {
|
||||
return fmt.Errorf("invalid inventory type: %s", inv.Type)
|
||||
}
|
||||
if len(inv.Hashes) == 0 {
|
||||
return nil
|
||||
func (s *Server) handleVersionCmd(msg *Message, p Peer) *Message {
|
||||
t := versionTuple{
|
||||
peer: p,
|
||||
request: msg.Payload.(*payload.Version),
|
||||
response: make(chan *Message),
|
||||
}
|
||||
|
||||
payload := payload.NewInventory(inv.Type, inv.Hashes)
|
||||
msg := newMessage(s.net, cmdGetData, payload)
|
||||
s.versionCh <- t
|
||||
|
||||
peer.send(msg)
|
||||
|
||||
return nil
|
||||
return <-t.response
|
||||
}
|
||||
|
||||
func (s *Server) handleBlockCmd(block *core.Block, peer Peer) error {
|
||||
fmt.Println("Block received")
|
||||
fmt.Printf("%+v\n", block)
|
||||
return nil
|
||||
type getaddrTuple struct {
|
||||
peer Peer
|
||||
request *Message
|
||||
response chan *Message
|
||||
}
|
||||
|
||||
func (s *Server) handleGetaddrCmd(msg *Message, p Peer) *Message {
|
||||
t := getaddrTuple{
|
||||
peer: p,
|
||||
request: msg,
|
||||
response: make(chan *Message),
|
||||
}
|
||||
|
||||
s.getaddrCh <- t
|
||||
|
||||
return <-t.response
|
||||
}
|
||||
|
||||
type invTuple struct {
|
||||
peer Peer
|
||||
request *payload.Inventory
|
||||
response chan *Message
|
||||
}
|
||||
|
||||
func (s *Server) handleInvCmd(msg *Message, p Peer) *Message {
|
||||
t := invTuple{
|
||||
request: msg.Payload.(*payload.Inventory),
|
||||
response: make(chan *Message),
|
||||
}
|
||||
|
||||
s.invCh <- t
|
||||
|
||||
return <-t.response
|
||||
}
|
||||
|
||||
type addrTuple struct {
|
||||
request *payload.AddressList
|
||||
response chan bool
|
||||
}
|
||||
|
||||
func (s *Server) handleAddrCmd(msg *Message, p Peer) bool {
|
||||
t := addrTuple{
|
||||
request: msg.Payload.(*payload.AddressList),
|
||||
response: make(chan bool),
|
||||
}
|
||||
|
||||
s.addrCh <- t
|
||||
|
||||
return <-t.response
|
||||
}
|
||||
|
||||
// check if the addr is already connected to the server.
|
||||
func (s *Server) peerAlreadyConnected(addr net.Addr) bool {
|
||||
// TODO: check for race conditions
|
||||
//s.mtx.RLock()
|
||||
//defer s.mtx.RUnlock()
|
||||
|
||||
// What about ourself ^^
|
||||
|
||||
for peer := range s.peers {
|
||||
if peer.endpoint().String() == addr.String() {
|
||||
if peer.addr().String() == addr.String() {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// After receiving the "getaddr" the server needs to respond with an "addr" message.
|
||||
// providing information about the other nodes in the network.
|
||||
// e.g. this server's connected peers.
|
||||
func (s *Server) handleGetAddrCmd(msg *Message, peer *Peer) error {
|
||||
// TODO
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Server) sendLoop(peer Peer) {
|
||||
// TODO: check if this peer is still connected.
|
||||
for {
|
||||
getaddrMsg := newMessage(s.net, cmdGetAddr, nil)
|
||||
peer.send(getaddrMsg)
|
||||
peer.callGetaddr(getaddrMsg)
|
||||
|
||||
time.Sleep(120 * time.Second)
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue