Browse Source

MakeConnectedSwitches function

pull/456/head
Ethan Buchman 8 years ago
parent
commit
711d2541f5
4 changed files with 94 additions and 57 deletions
  1. +40
    -0
      listener_test.go
  2. +5
    -2
      netaddress.go
  3. +42
    -0
      switch.go
  4. +7
    -55
      switch_test.go

+ 40
- 0
listener_test.go View File

@ -0,0 +1,40 @@
package p2p
import (
"bytes"
"testing"
)
func TestListener(t *testing.T) {
// Create a listener
l := NewDefaultListener("tcp", ":8001", true)
// Dial the listener
lAddr := l.ExternalAddress()
connOut, err := lAddr.Dial()
if err != nil {
t.Fatalf("Could not connect to listener address %v", lAddr)
} else {
t.Logf("Created a connection to listener address %v", lAddr)
}
connIn, ok := <-l.Connections()
if !ok {
t.Fatalf("Could not get inbound connection from listener")
}
msg := []byte("hi!")
go connIn.Write(msg)
b := make([]byte, 32)
n, err := connOut.Read(b)
if err != nil {
t.Fatalf("Error reading off connection: %v", err)
}
b = b[:n]
if !bytes.Equal(msg, b) {
t.Fatalf("Got %s, expected %s", b, msg)
}
// Close the server, no longer needed.
l.Stop()
}

+ 5
- 2
netaddress.go View File

@ -5,7 +5,6 @@
package p2p
import (
"fmt"
"net"
"strconv"
"time"
@ -23,7 +22,11 @@ type NetAddress struct {
func NewNetAddress(addr net.Addr) *NetAddress {
tcpAddr, ok := addr.(*net.TCPAddr)
if !ok {
PanicSanity(fmt.Sprintf("Only TCPAddrs are supported. Got: %v", addr))
log.Warn(`Only TCPAddrs are supported. If used for anything but testing,
may result in undefined behaviour!`, "addr", addr)
return NewNetAddressIPPort(net.IP("0.0.0.0"), 0)
// NOTE: it would be nice to only not panic if we're in testing ...
// PanicSanity(Fmt("Only TCPAddrs are supported. Got: %v", addr))
}
ip := tcpAddr.IP
port := uint16(tcpAddr.Port)


+ 42
- 0
switch.go View File

@ -415,3 +415,45 @@ type SwitchEventDonePeer struct {
Peer *Peer
Error interface{}
}
//------------------------------------------------------------------
// Switches connected via arbitrary net.Conn; useful for testing
// Returns n fully connected switches.
// initSwitch defines how the ith switch should be initialized (ie. with what reactors).
func MakeConnectedSwitches(n int, initSwitch func(int, *Switch) *Switch, connPipe func() (net.Conn, net.Conn)) []*Switch {
switches := make([]*Switch, n)
for i := 0; i < n; i++ {
switches[i] = makeSwitch(i, "testing", "123.123.123", initSwitch)
}
for i := 0; i < n; i++ {
switchI := switches[i]
for j := i; j < n; j++ {
switchJ := switches[j]
c1, c2 := connPipe()
go switchI.AddPeerWithConnection(c1, false) // AddPeer is blocking, requires handshake.
go switchJ.AddPeerWithConnection(c2, true)
}
}
// Wait for things to happen, peers to get added...
time.Sleep(100 * time.Millisecond * time.Duration(n*n))
return switches
}
func makeSwitch(i int, network, version string, initSwitch func(int, *Switch) *Switch) *Switch {
privKey := crypto.GenPrivKeyEd25519()
// new switch, add reactors
// TODO: let the config be passed in?
s := initSwitch(i, NewSwitch(cfg.NewMapConfig(nil)))
s.SetNodeInfo(&NodeInfo{
PubKey: privKey.PubKey().(crypto.PubKeyEd25519),
Moniker: Fmt("switch%d", i),
Network: network,
Version: version,
})
s.SetNodePrivKey(privKey)
s.Start() // start switch and reactors
return s
}

+ 7
- 55
switch_test.go View File

@ -2,13 +2,13 @@ package p2p
import (
"bytes"
"net"
"sync"
"testing"
"time"
. "github.com/tendermint/go-common"
cfg "github.com/tendermint/go-config"
"github.com/tendermint/go-crypto"
"github.com/tendermint/go-wire"
)
@ -85,63 +85,15 @@ func (tr *TestReactor) getMsgs(chID byte) []PeerMessage {
//-----------------------------------------------------------------------------
// convenience method for creating two switches connected to each other.
func makeSwitchPair(t testing.TB, initSwitch func(*Switch) *Switch) (*Switch, *Switch) {
s1PrivKey := crypto.GenPrivKeyEd25519()
s2PrivKey := crypto.GenPrivKeyEd25519()
// XXX: note this uses net.Pipe and not a proper TCP conn
func makeSwitchPair(t testing.TB, initSwitch func(int, *Switch) *Switch) (*Switch, *Switch) {
// Create two switches that will be interconnected.
s1 := initSwitch(NewSwitch(config))
s1.SetNodeInfo(&NodeInfo{
PubKey: s1PrivKey.PubKey().(crypto.PubKeyEd25519),
Moniker: "switch1",
Network: "testing",
Version: "123.123.123",
})
s1.SetNodePrivKey(s1PrivKey)
s2 := initSwitch(NewSwitch(config))
s2.SetNodeInfo(&NodeInfo{
PubKey: s2PrivKey.PubKey().(crypto.PubKeyEd25519),
Moniker: "switch2",
Network: "testing",
Version: "123.123.123",
})
s2.SetNodePrivKey(s2PrivKey)
// Start switches and reactors
s1.Start()
s2.Start()
// Create a listener for s1
l := NewDefaultListener("tcp", ":8001", true)
// Dial the listener & add the connection to s2.
lAddr := l.ExternalAddress()
connOut, err := lAddr.Dial()
if err != nil {
t.Fatalf("Could not connect to listener address %v", lAddr)
} else {
t.Logf("Created a connection to listener address %v", lAddr)
}
connIn, ok := <-l.Connections()
if !ok {
t.Fatalf("Could not get inbound connection from listener")
}
go s1.AddPeerWithConnection(connIn, false) // AddPeer is blocking, requires handshake.
s2.AddPeerWithConnection(connOut, true)
// Wait for things to happen, peers to get added...
time.Sleep(100 * time.Millisecond)
// Close the server, no longer needed.
l.Stop()
return s1, s2
switches := MakeConnectedSwitches(2, initSwitch, net.Pipe)
return switches[0], switches[1]
}
func TestSwitches(t *testing.T) {
s1, s2 := makeSwitchPair(t, func(sw *Switch) *Switch {
s1, s2 := makeSwitchPair(t, func(i int, sw *Switch) *Switch {
// Make two reactors of two channels each
sw.AddReactor("foo", NewTestReactor([]*ChannelDescriptor{
&ChannelDescriptor{ID: byte(0x00), Priority: 10},
@ -208,7 +160,7 @@ func BenchmarkSwitches(b *testing.B) {
b.StopTimer()
s1, s2 := makeSwitchPair(b, func(sw *Switch) *Switch {
s1, s2 := makeSwitchPair(b, func(i int, sw *Switch) *Switch {
// Make bar reactors of bar channels each
sw.AddReactor("foo", NewTestReactor([]*ChannelDescriptor{
&ChannelDescriptor{ID: byte(0x00), Priority: 10},


Loading…
Cancel
Save