package main import ( "container/list" "github.com/ethereum/ethdb-go" "github.com/ethereum/ethutil-go" "github.com/ethereum/ethwire-go" "log" "net" "sync/atomic" "time" ) func eachPeer(peers *list.List, callback func(*Peer, *list.Element)) { // Loop thru the peers and close them (if we had them) for e := peers.Front(); e != nil; e = e.Next() { if peer, ok := e.Value.(*Peer); ok { callback(peer, e) } } } const ( processReapingTimeout = 60 // TODO increase ) type Server struct { // Channel for shutting down the server shutdownChan chan bool // DB interface //db *ethdb.LDBDatabase db *ethdb.MemDatabase // Block manager for processing new blocks and managing the block chain blockManager *BlockManager // Peers (NYI) peers *list.List // Nonce Nonce uint64 } func NewServer() (*Server, error) { //db, err := ethdb.NewLDBDatabase() db, err := ethdb.NewMemDatabase() if err != nil { return nil, err } ethutil.SetConfig(db) nonce, _ := ethutil.RandomUint64() server := &Server{ shutdownChan: make(chan bool), blockManager: NewBlockManager(), db: db, peers: list.New(), Nonce: nonce, } return server, nil } func (s *Server) AddPeer(conn net.Conn) { peer := NewPeer(conn, s, true) if peer != nil { s.peers.PushBack(peer) peer.Start() log.Println("Peer connected ::", conn.RemoteAddr()) } } func (s *Server) ProcessPeerList(addrs []string) { for _, addr := range addrs { // TODO Probably requires some sanity checks s.ConnectToPeer(addr) } } func (s *Server) ConnectToPeer(addr string) error { peer := NewOutboundPeer(addr, s) s.peers.PushBack(peer) return nil } func (s *Server) OutboundPeers() []*Peer { // Create a new peer slice with at least the length of the total peers outboundPeers := make([]*Peer, s.peers.Len()) length := 0 eachPeer(s.peers, func(p *Peer, e *list.Element) { if !p.inbound { outboundPeers[length] = p length++ } }) return outboundPeers[:length] } func (s *Server) InboundPeers() []*Peer { // Create a new peer slice with at least the length of the total peers inboundPeers := make([]*Peer, s.peers.Len()) length := 0 eachPeer(s.peers, func(p *Peer, e *list.Element) { if p.inbound { inboundPeers[length] = p length++ } }) return inboundPeers[:length] } func (s *Server) Broadcast(msgType ethwire.MsgType, data []byte) { eachPeer(s.peers, func(p *Peer, e *list.Element) { p.QueueMessage(ethwire.NewMessage(msgType, data)) }) } func (s *Server) ReapDeadPeers() { for { eachPeer(s.peers, func(p *Peer, e *list.Element) { if atomic.LoadInt32(&p.disconnect) == 1 || (p.inbound && (time.Now().Unix()-p.lastPong) > int64(5*time.Minute)) { log.Println("Dead peer found .. reaping") s.peers.Remove(e) } }) time.Sleep(processReapingTimeout * time.Second) } } // Start the server func (s *Server) Start() { // For now this function just blocks the main thread ln, err := net.Listen("tcp", ":12345") if err != nil { // This is mainly for testing to create a "network" if Debug { log.Println("Connection listening disabled. Acting as client") err = s.ConnectToPeer("localhost:12345") if err != nil { log.Println("Error starting server", err) s.Stop() } return } else { log.Fatal(err) } } // Start the reaping processes go s.ReapDeadPeers() go func() { for { conn, err := ln.Accept() if err != nil { log.Println(err) continue } go s.AddPeer(conn) } }() // TMP /* go func() { for { s.Broadcast("block", s.blockManager.bc.GenesisBlock().RlpEncode()) time.Sleep(1000 * time.Millisecond) } }() */ } func (s *Server) Stop() { // Close the database defer s.db.Close() eachPeer(s.peers, func(p *Peer, e *list.Element) { p.Stop() }) s.shutdownChan <- true } // This function will wait for a shutdown and resumes main thread execution func (s *Server) WaitForShutdown() { <-s.shutdownChan }