|
@ -4,6 +4,7 @@ import ( |
|
|
"errors" |
|
|
"errors" |
|
|
"fmt" |
|
|
"fmt" |
|
|
"net" |
|
|
"net" |
|
|
|
|
|
"sync/atomic" |
|
|
"time" |
|
|
"time" |
|
|
|
|
|
|
|
|
. "github.com/tendermint/tendermint/common" |
|
|
. "github.com/tendermint/tendermint/common" |
|
@ -34,6 +35,7 @@ type Switch struct { |
|
|
reactorsByCh map[byte]Reactor |
|
|
reactorsByCh map[byte]Reactor |
|
|
peers *PeerSet |
|
|
peers *PeerSet |
|
|
dialing *CMap |
|
|
dialing *CMap |
|
|
|
|
|
running uint32 |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
var ( |
|
|
var ( |
|
@ -53,6 +55,7 @@ func NewSwitch() *Switch { |
|
|
reactorsByCh: make(map[byte]Reactor), |
|
|
reactorsByCh: make(map[byte]Reactor), |
|
|
peers: NewPeerSet(), |
|
|
peers: NewPeerSet(), |
|
|
dialing: NewCMap(), |
|
|
dialing: NewCMap(), |
|
|
|
|
|
running: 0, |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
return sw |
|
|
return sw |
|
@ -80,62 +83,64 @@ func (sw *Switch) AddReactor(name string, reactor Reactor) Reactor { |
|
|
return reactor |
|
|
return reactor |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
// Not goroutine safe.
|
|
|
|
|
|
func (sw *Switch) Reactors() map[string]Reactor { |
|
|
|
|
|
return sw.reactors |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
// Not goroutine safe.
|
|
|
func (sw *Switch) Reactor(name string) Reactor { |
|
|
func (sw *Switch) Reactor(name string) Reactor { |
|
|
return sw.reactors[name] |
|
|
return sw.reactors[name] |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
// Convenience function
|
|
|
|
|
|
func (sw *Switch) StartReactors() { |
|
|
|
|
|
for _, reactor := range sw.reactors { |
|
|
|
|
|
reactor.Start(sw) |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
// Not goroutine safe.
|
|
|
|
|
|
func (sw *Switch) AddListener(l Listener) { |
|
|
|
|
|
sw.listeners = append(sw.listeners, l) |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
// Convenience function
|
|
|
|
|
|
func (sw *Switch) StopReactors() { |
|
|
|
|
|
// Stop all reactors.
|
|
|
|
|
|
for _, reactor := range sw.reactors { |
|
|
|
|
|
reactor.Stop() |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
func (sw *Switch) Listeners() []Listener { |
|
|
|
|
|
return sw.listeners |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
// Convenience function
|
|
|
|
|
|
func (sw *Switch) StopPeers() { |
|
|
|
|
|
// Stop each peer.
|
|
|
|
|
|
for _, peer := range sw.peers.List() { |
|
|
|
|
|
peer.stop() |
|
|
|
|
|
} |
|
|
|
|
|
sw.peers = NewPeerSet() |
|
|
|
|
|
|
|
|
// Not goroutine safe.
|
|
|
|
|
|
func (sw *Switch) IsListening() bool { |
|
|
|
|
|
return len(sw.listeners) > 0 |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
// Convenience function
|
|
|
|
|
|
func (sw *Switch) StopListeners() { |
|
|
|
|
|
// Stop each listener.
|
|
|
|
|
|
for _, listener := range sw.listeners { |
|
|
|
|
|
listener.Stop() |
|
|
|
|
|
|
|
|
func (sw *Switch) Start() { |
|
|
|
|
|
if atomic.CompareAndSwapUint32(&sw.running, 0, 1) { |
|
|
|
|
|
// Start reactors
|
|
|
|
|
|
for _, reactor := range sw.reactors { |
|
|
|
|
|
reactor.Start(sw) |
|
|
|
|
|
} |
|
|
|
|
|
// Start peers
|
|
|
|
|
|
for _, peer := range sw.peers.List() { |
|
|
|
|
|
sw.startInitPeer(peer) |
|
|
|
|
|
} |
|
|
|
|
|
// Start listeners
|
|
|
|
|
|
for _, listener := range sw.listeners { |
|
|
|
|
|
go sw.listenerRoutine(listener) |
|
|
|
|
|
} |
|
|
} |
|
|
} |
|
|
sw.listeners = nil |
|
|
|
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
// Convenience function
|
|
|
|
|
|
func (sw *Switch) Stop() { |
|
|
func (sw *Switch) Stop() { |
|
|
sw.StopPeers() |
|
|
|
|
|
sw.StopReactors() |
|
|
|
|
|
sw.StopListeners() |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
// Not goroutine safe to modify.
|
|
|
|
|
|
func (sw *Switch) Reactors() map[string]Reactor { |
|
|
|
|
|
return sw.reactors |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
func (sw *Switch) AddListener(l Listener) { |
|
|
|
|
|
sw.listeners = append(sw.listeners, l) |
|
|
|
|
|
go sw.listenerRoutine(l) |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
func (sw *Switch) IsListening() bool { |
|
|
|
|
|
return len(sw.listeners) > 0 |
|
|
|
|
|
|
|
|
if atomic.CompareAndSwapUint32(&sw.running, 1, 0) { |
|
|
|
|
|
// Stop listeners
|
|
|
|
|
|
for _, listener := range sw.listeners { |
|
|
|
|
|
listener.Stop() |
|
|
|
|
|
} |
|
|
|
|
|
sw.listeners = nil |
|
|
|
|
|
// Stop peers
|
|
|
|
|
|
for _, peer := range sw.peers.List() { |
|
|
|
|
|
peer.stop() |
|
|
|
|
|
} |
|
|
|
|
|
sw.peers = NewPeerSet() |
|
|
|
|
|
// Stop reactors
|
|
|
|
|
|
for _, reactor := range sw.reactors { |
|
|
|
|
|
reactor.Stop() |
|
|
|
|
|
} |
|
|
|
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
func (sw *Switch) AddPeerWithConnection(conn net.Conn, outbound bool) (*Peer, error) { |
|
|
func (sw *Switch) AddPeerWithConnection(conn net.Conn, outbound bool) (*Peer, error) { |
|
@ -149,6 +154,13 @@ func (sw *Switch) AddPeerWithConnection(conn net.Conn, outbound bool) (*Peer, er |
|
|
return nil, ErrSwitchDuplicatePeer |
|
|
return nil, ErrSwitchDuplicatePeer |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
if atomic.LoadUint32(&sw.running) == 1 { |
|
|
|
|
|
sw.startInitPeer(peer) |
|
|
|
|
|
} |
|
|
|
|
|
return peer, nil |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
func (sw *Switch) startInitPeer(peer *Peer) { |
|
|
// Start the peer
|
|
|
// Start the peer
|
|
|
peer.start() |
|
|
peer.start() |
|
|
|
|
|
|
|
@ -158,8 +170,6 @@ func (sw *Switch) AddPeerWithConnection(conn net.Conn, outbound bool) (*Peer, er |
|
|
// Send handshake
|
|
|
// Send handshake
|
|
|
msg := &pexHandshakeMessage{Network: sw.network} |
|
|
msg := &pexHandshakeMessage{Network: sw.network} |
|
|
peer.Send(PexChannel, msg) |
|
|
peer.Send(PexChannel, msg) |
|
|
|
|
|
|
|
|
return peer, nil |
|
|
|
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
func (sw *Switch) DialPeerWithAddress(addr *NetAddress) (*Peer, error) { |
|
|
func (sw *Switch) DialPeerWithAddress(addr *NetAddress) (*Peer, error) { |
|
|