From f38052c499c1fee61423efeddb1f52677f1442e9 Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Tue, 4 Nov 2014 13:21:44 +0100 Subject: p2p: rework protocol API --- p2p/server_test.go | 204 ++++++++++++++++++++++++++++++++++++----------------- 1 file changed, 141 insertions(+), 63 deletions(-) (limited to 'p2p/server_test.go') diff --git a/p2p/server_test.go b/p2p/server_test.go index f749cc490..472759231 100644 --- a/p2p/server_test.go +++ b/p2p/server_test.go @@ -1,8 +1,8 @@ package p2p import ( - "bytes" "fmt" + "io" "net" "testing" "time" @@ -32,6 +32,7 @@ func (self *TestNetwork) Listener(addr net.Addr) (net.Listener, error) { connections: self.connections, addr: addr, max: self.maxinbound, + close: make(chan struct{}), }, nil } @@ -76,24 +77,25 @@ type TestListener struct { addr net.Addr max int i int + close chan struct{} } -func (self *TestListener) Accept() (conn net.Conn, err error) { +func (self *TestListener) Accept() (net.Conn, error) { self.i++ if self.i > self.max { - err = fmt.Errorf("no more") - } else { - addr := &TestAddr{fmt.Sprintf("inboundpeer-%d", self.i)} - tconn := NewTestNetworkConnection(addr) - key := tconn.RemoteAddr().String() - self.connections[key] = tconn - conn = net.Conn(tconn) - fmt.Printf("accepted connection from: %v \n", addr) + <-self.close + return nil, io.EOF } - return + addr := &TestAddr{fmt.Sprintf("inboundpeer-%d", self.i)} + tconn := NewTestNetworkConnection(addr) + key := tconn.RemoteAddr().String() + self.connections[key] = tconn + fmt.Printf("accepted connection from: %v \n", addr) + return tconn, nil } func (self *TestListener) Close() error { + close(self.close) return nil } @@ -101,6 +103,86 @@ func (self *TestListener) Addr() net.Addr { return self.addr } +type TestNetworkConnection struct { + in chan []byte + close chan struct{} + current []byte + Out [][]byte + addr net.Addr +} + +func NewTestNetworkConnection(addr net.Addr) *TestNetworkConnection { + return &TestNetworkConnection{ + in: make(chan []byte), + close: make(chan struct{}), + current: []byte{}, + Out: [][]byte{}, + addr: addr, + } +} + +func (self *TestNetworkConnection) In(latency time.Duration, packets ...[]byte) { + time.Sleep(latency) + for _, s := range packets { + self.in <- s + } +} + +func (self *TestNetworkConnection) Read(buff []byte) (n int, err error) { + if len(self.current) == 0 { + var ok bool + select { + case self.current, ok = <-self.in: + if !ok { + return 0, io.EOF + } + case <-self.close: + return 0, io.EOF + } + } + length := len(self.current) + if length > len(buff) { + copy(buff[:], self.current[:len(buff)]) + self.current = self.current[len(buff):] + return len(buff), nil + } else { + copy(buff[:length], self.current[:]) + self.current = []byte{} + return length, io.EOF + } +} + +func (self *TestNetworkConnection) Write(buff []byte) (n int, err error) { + self.Out = append(self.Out, buff) + fmt.Printf("net write(%d): %x\n", len(self.Out), buff) + return len(buff), nil +} + +func (self *TestNetworkConnection) Close() error { + close(self.close) + return nil +} + +func (self *TestNetworkConnection) LocalAddr() (addr net.Addr) { + return +} + +func (self *TestNetworkConnection) RemoteAddr() (addr net.Addr) { + return self.addr +} + +func (self *TestNetworkConnection) SetDeadline(t time.Time) (err error) { + return +} + +func (self *TestNetworkConnection) SetReadDeadline(t time.Time) (err error) { + return +} + +func (self *TestNetworkConnection) SetWriteDeadline(t time.Time) (err error) { + return +} + func SetupTestServer(handlers Handlers) (network *TestNetwork, server *Server) { network = NewTestNetwork(1) addr := &TestAddr{"test:30303"} @@ -124,12 +206,10 @@ func TestServerListener(t *testing.T) { if !ok { t.Error("not found inbound peer 1") } else { - fmt.Printf("out: %v\n", peer1.Out) if len(peer1.Out) != 2 { - t.Errorf("not enough messages sent to peer 1: %v ", len(peer1.Out)) + t.Errorf("wrong number of writes to peer 1: got %d, want %d", len(peer1.Out), 2) } } - } func TestServerDialer(t *testing.T) { @@ -142,65 +222,63 @@ func TestServerDialer(t *testing.T) { if !ok { t.Error("not found outbound peer 1") } else { - fmt.Printf("out: %v\n", peer1.Out) if len(peer1.Out) != 2 { - t.Errorf("not enough messages sent to peer 1: %v ", len(peer1.Out)) + t.Errorf("wrong number of writes to peer 1: got %d, want %d", len(peer1.Out), 2) } } } -func TestServerBroadcast(t *testing.T) { - handlers := make(Handlers) - testProtocol := &TestProtocol{Msgs: []*Msg{}} - handlers["aaa"] = func(p *Peer) Protocol { return testProtocol } - network, server := SetupTestServer(handlers) - server.Start(true, true) - server.peerConnect <- &TestAddr{"outboundpeer-1"} - time.Sleep(10 * time.Millisecond) - msg, _ := NewMsg(0) - server.Broadcast("", msg) - packet := Packet(0, 0) - time.Sleep(10 * time.Millisecond) - server.Stop() - peer1, ok := network.connections["outboundpeer-1"] - if !ok { - t.Error("not found outbound peer 1") - } else { - fmt.Printf("out: %v\n", peer1.Out) - if len(peer1.Out) != 3 { - t.Errorf("not enough messages sent to peer 1: %v ", len(peer1.Out)) - } else { - if bytes.Compare(peer1.Out[1], packet) != 0 { - t.Errorf("incorrect broadcast packet %v != %v", peer1.Out[1], packet) - } - } - } - peer2, ok := network.connections["inboundpeer-1"] - if !ok { - t.Error("not found inbound peer 2") - } else { - fmt.Printf("out: %v\n", peer2.Out) - if len(peer1.Out) != 3 { - t.Errorf("not enough messages sent to peer 2: %v ", len(peer2.Out)) - } else { - if bytes.Compare(peer2.Out[1], packet) != 0 { - t.Errorf("incorrect broadcast packet %v != %v", peer2.Out[1], packet) - } - } - } -} +// func TestServerBroadcast(t *testing.T) { +// handlers := make(Handlers) +// testProtocol := &TestProtocol{Msgs: []*Msg{}} +// handlers["aaa"] = func(p *Peer) Protocol { return testProtocol } +// network, server := SetupTestServer(handlers) +// server.Start(true, true) +// server.peerConnect <- &TestAddr{"outboundpeer-1"} +// time.Sleep(10 * time.Millisecond) +// msg := NewMsg(0) +// server.Broadcast("", msg) +// packet := Packet(0, 0) +// time.Sleep(10 * time.Millisecond) +// server.Stop() +// peer1, ok := network.connections["outboundpeer-1"] +// if !ok { +// t.Error("not found outbound peer 1") +// } else { +// fmt.Printf("out: %v\n", peer1.Out) +// if len(peer1.Out) != 3 { +// t.Errorf("not enough messages sent to peer 1: %v ", len(peer1.Out)) +// } else { +// if bytes.Compare(peer1.Out[1], packet) != 0 { +// t.Errorf("incorrect broadcast packet %v != %v", peer1.Out[1], packet) +// } +// } +// } +// peer2, ok := network.connections["inboundpeer-1"] +// if !ok { +// t.Error("not found inbound peer 2") +// } else { +// fmt.Printf("out: %v\n", peer2.Out) +// if len(peer1.Out) != 3 { +// t.Errorf("not enough messages sent to peer 2: %v ", len(peer2.Out)) +// } else { +// if bytes.Compare(peer2.Out[1], packet) != 0 { +// t.Errorf("incorrect broadcast packet %v != %v", peer2.Out[1], packet) +// } +// } +// } +// } func TestServerPeersMessage(t *testing.T) { - handlers := make(Handlers) - _, server := SetupTestServer(handlers) + _, server := SetupTestServer(nil) server.Start(true, true) defer server.Stop() server.peerConnect <- &TestAddr{"outboundpeer-1"} - time.Sleep(10 * time.Millisecond) - peersMsg, err := server.PeersMessage() - fmt.Println(peersMsg) - if err != nil { - t.Errorf("expect no error, got %v", err) + time.Sleep(2000 * time.Millisecond) + + pl := server.encodedPeerList() + if pl == nil { + t.Errorf("expect non-nil peer list") } if c := server.PeerCount(); c != 2 { t.Errorf("expect 2 peers, got %v", c) -- cgit