neoneo-go/pkg/network/server.go

209 lines
4.9 KiB
Go
Raw Normal View History

2018-01-26 18:04:13 +00:00
package network
import (
"errors"
"fmt"
"log"
"net"
"os"
2018-01-26 20:39:34 +00:00
"strconv"
2018-01-26 18:04:13 +00:00
)
const (
2018-01-26 20:39:34 +00:00
// node version
version = "2.6.0"
// official ports according to the protocol.
2018-01-26 18:04:13 +00:00
portMainNet = 10333
portTestNet = 20333
2018-01-26 20:39:34 +00:00
)
var (
// rpcLogger used for debugging RPC messages between nodes.
rpcLogger = log.New(os.Stdout, "RPC :: ", 0)
2018-01-26 18:04:13 +00:00
)
type messageTuple struct {
peer *Peer
msg *Message
}
// Server is the representation of a full working NEO TCP node.
type Server struct {
logger *log.Logger
2018-01-26 20:39:34 +00:00
// the port the TCP listener is listening on.
port uint16
2018-01-26 18:04:13 +00:00
// userAgent of the server.
userAgent string
// The "magic" mode the server is currently running on.
// This can either be 0x00746e41 or 0x74746e41 for main or test net.
2018-01-26 20:39:34 +00:00
// Or 56753 to work with the docker privnet.
net NetMode
2018-01-26 18:04:13 +00:00
// map that holds all connected peers to this server.
peers map[*Peer]bool
register chan *Peer
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
}
// NewServer returns a pointer to a new server.
2018-01-26 20:39:34 +00:00
func NewServer(net NetMode) *Server {
2018-01-26 18:04:13 +00:00
logger := log.New(os.Stdout, "NEO SERVER :: ", 0)
2018-01-26 20:39:34 +00:00
if net != ModeTestNet && net != ModeMainNet && net != ModeDevNet {
logger.Fatalf("invalid network mode %d", net)
2018-01-26 18:04:13 +00:00
}
s := &Server{
userAgent: fmt.Sprintf("/NEO:%s/", version),
logger: logger,
peers: make(map[*Peer]bool),
register: make(chan *Peer),
unregister: make(chan *Peer),
message: make(chan messageTuple),
relay: true,
2018-01-26 20:39:34 +00:00
net: net,
2018-01-26 18:04:13 +00:00
quit: make(chan struct{}),
}
return s
}
// Start run's the server.
func (s *Server) Start(port string, seeds []string) {
2018-01-26 20:39:34 +00:00
p, err := strconv.Atoi(port[1:len(port)])
if err != nil {
s.logger.Fatalf("could not convert port to integer: %s", err)
}
s.port = uint16(p)
2018-01-26 18:04:13 +00:00
fmt.Println(logo())
2018-01-26 20:39:34 +00:00
s.logger.Printf("running %s on %s - TCP %d - relay: %v",
s.userAgent, s.net, int(s.port), s.relay)
2018-01-26 18:04:13 +00:00
go listenTCP(s, port)
if len(seeds) > 0 {
connectToSeeds(s, seeds)
}
s.loop()
}
// Stop the server, attemping a gracefull shutdown.
func (s *Server) Stop() { s.quit <- struct{}{} }
// shutdown the server, disconnecting all peers.
func (s *Server) shutdown() {
s.logger.Println("attemping a quitefull shutdown.")
s.listener.Close()
// disconnect and remove all connected peers.
for peer := range s.peers {
peer.conn.Close()
s.unregister <- peer
}
}
func (s *Server) loop() {
for {
select {
case peer := <-s.register:
s.logger.Printf("peer registered from address %s", peer.conn.RemoteAddr())
2018-01-26 20:39:34 +00:00
// only respond with the version mesage if the peer initiated the connection.
if peer.initiater {
resp, err := s.handlePeerConnected()
if err != nil {
s.logger.Fatalf("handling initial peer connection failed: %s", err)
}
peer.send <- resp
2018-01-26 18:04:13 +00:00
}
case peer := <-s.unregister:
s.logger.Printf("peer %s disconnected", peer.conn.RemoteAddr())
case tuple := <-s.message:
if err := s.processMessage(tuple.msg, tuple.peer); err != nil {
s.logger.Fatalf("failed to process message: %s", err)
}
case <-s.quit:
s.shutdown()
}
}
}
// TODO: unregister peers on error.
// processMessage processes the received message from a remote node.
func (s *Server) processMessage(msg *Message, peer *Peer) error {
2018-01-26 20:39:34 +00:00
rpcLogger.Printf("IN :: %+v", msg)
2018-01-26 18:04:13 +00:00
switch msg.commandType() {
case cmdVersion:
v, _ := msg.decodePayload()
resp, err := s.handleVersionCmd(v.(*Version))
if err != nil {
return err
}
peer.send <- resp
case cmdVerack:
case cmdGetAddr:
case cmdAddr:
case cmdGetHeaders:
case cmdHeaders:
case cmdGetBlocks:
case cmdInv:
case cmdGetData:
case cmdBlock:
case cmdTX:
default:
return errors.New("invalid RPC command received: " + string(msg.commandType()))
}
return nil
}
// When a new peer is connected we respond with the version command.
// No further communication should been made before both sides has received
// the version of eachother.
func (s *Server) handlePeerConnected() (*Message, error) {
2018-01-26 20:39:34 +00:00
payload := newVersionPayload(s.port, s.userAgent, 0, s.relay)
2018-01-26 18:04:13 +00:00
b, err := payload.encode()
if err != nil {
return nil, err
}
2018-01-26 20:39:34 +00:00
msg := newMessage(s.net, cmdVersion, b)
2018-01-26 18:04:13 +00:00
return msg, nil
}
// Version declares the server's version when a new connection is been made.
// We respond with a instant "verack" message.
func (s *Server) handleVersionCmd(v *Version) (*Message, error) {
// TODO: check version and verify to trust that node.
// Empty payload for the verack message.
2018-01-26 20:39:34 +00:00
msg := newMessage(s.net, cmdVerack, nil)
2018-01-26 18:04:13 +00:00
return msg, nil
}
func logo() string {
return `
_ ____________ __________
/ | / / ____/ __ \ / ____/ __ \
/ |/ / __/ / / / /_____/ / __/ / / /
/ /| / /___/ /_/ /_____/ /_/ / /_/ /
/_/ |_/_____/\____/ \____/\____/
`
}