refactor: separate object creation from activation
This commit is contained in:
1 parent
ebe48f5e58
commit
48c8f9ea01
7 files changed
+63
-46
No files matched your search
+11
-9
@@ -49,23 +49,25 @@ func (conn *RemoteConnection) String() string {
|
||||
return fmt.Sprint("Connection ", from, "->", to)
|
||||
}
|
||||
|
||||
// Async. Does not return anything. If the connection is successful,
|
||||
// it will end up in the local peer's connections map.
|
||||
func NewLocalConnection(connRemote *RemoteConnection, acceptNewPeer bool, tcpConn *net.TCPConn, udpAddr *net.UDPAddr, router *Router) {
|
||||
func NewLocalConnection(connRemote *RemoteConnection, tcpConn *net.TCPConn, udpAddr *net.UDPAddr, router *Router) *LocalConnection {
|
||||
if connRemote.local != router.Ourself.Peer {
|
||||
log.Fatal("Attempt to create local connection from a peer which is not ourself")
|
||||
}
|
||||
|
||||
queryChan := make(chan *ConnectionInteraction, ChannelSize)
|
||||
// NB, we're taking a copy of connRemote here.
|
||||
connLocal := &LocalConnection{
|
||||
return &LocalConnection{
|
||||
RemoteConnection: *connRemote,
|
||||
Router: router,
|
||||
TCPConn: tcpConn,
|
||||
remoteUDPAddr: udpAddr,
|
||||
effectivePMTU: DefaultPMTU,
|
||||
queryChan: queryChan}
|
||||
go connLocal.run(queryChan, acceptNewPeer)
|
||||
effectivePMTU: DefaultPMTU}
|
||||
}
|
||||
|
||||
// Async. Does not return anything. If the connection is successful,
|
||||
// it will end up in the local peer's connections map.
|
||||
func (conn *LocalConnection) Start(acceptNewPeer bool) {
|
||||
queryChan := make(chan *ConnectionInteraction, ChannelSize)
|
||||
conn.queryChan = queryChan
|
||||
go conn.run(queryChan, acceptNewPeer)
|
||||
}
|
||||
|
||||
func (conn *LocalConnection) Established() bool {
|
||||
|
||||
@@ -41,16 +41,18 @@ type ConnectionMakerInteraction struct {
|
||||
address string
|
||||
}
|
||||
|
||||
func StartConnectionMaker(ourself *LocalPeer, peers *Peers) *ConnectionMaker {
|
||||
queryChan := make(chan *ConnectionMakerInteraction, ChannelSize)
|
||||
state := &ConnectionMaker{
|
||||
func NewConnectionMaker(ourself *LocalPeer, peers *Peers) *ConnectionMaker {
|
||||
return &ConnectionMaker{
|
||||
ourself: ourself,
|
||||
peers: peers,
|
||||
queryChan: queryChan,
|
||||
cmdLineAddress: make(map[string]bool),
|
||||
targets: make(map[string]*Target)}
|
||||
go state.queryLoop(queryChan)
|
||||
return state
|
||||
}
|
||||
|
||||
func (cm *ConnectionMaker) Start() {
|
||||
queryChan := make(chan *ConnectionMakerInteraction, ChannelSize)
|
||||
cm.queryChan = queryChan
|
||||
go cm.queryLoop(queryChan)
|
||||
}
|
||||
|
||||
func (cm *ConnectionMaker) InitiateConnection(address string) {
|
||||
|
||||
+7
-3
@@ -50,8 +50,8 @@ func (conn *LocalConnection) ensureForwarders() error {
|
||||
conn.effectivePMTU = forwarder.unverifiedPMTU
|
||||
conn.Unlock()
|
||||
|
||||
go forwarder.Run()
|
||||
go forwarderDF.Run()
|
||||
forwarder.Start()
|
||||
forwarderDF.Start()
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -204,7 +204,11 @@ func NewForwarder(conn *LocalConnection, ch <-chan *ForwardedFrame, stop <-chan
|
||||
return fwd
|
||||
}
|
||||
|
||||
func (fwd *Forwarder) Run() {
|
||||
func (fwd *Forwarder) Start() {
|
||||
go fwd.run()
|
||||
}
|
||||
|
||||
func (fwd *Forwarder) run() {
|
||||
defer fwd.udpSender.Shutdown()
|
||||
var flushed, ok bool
|
||||
var frame *ForwardedFrame
|
||||
|
||||
@@ -17,12 +17,14 @@ type PeerInteraction struct {
|
||||
payload interface{}
|
||||
}
|
||||
|
||||
func StartLocalPeer(name PeerName, router *Router) *LocalPeer {
|
||||
peer := &LocalPeer{Peer: NewPeer(name, 0, 0), Router: router}
|
||||
func NewLocalPeer(name PeerName, router *Router) *LocalPeer {
|
||||
return &LocalPeer{Peer: NewPeer(name, 0, 0), Router: router}
|
||||
}
|
||||
|
||||
func (peer *LocalPeer) Start() {
|
||||
queryChan := make(chan *PeerInteraction, ChannelSize)
|
||||
peer.queryChan = queryChan
|
||||
go peer.queryLoop(queryChan)
|
||||
return peer
|
||||
}
|
||||
|
||||
func (peer *LocalPeer) Forward(dstPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error {
|
||||
@@ -106,7 +108,8 @@ func (peer *LocalPeer) CreateConnection(peerAddr string, acceptNewPeer bool) err
|
||||
return err
|
||||
}
|
||||
connRemote := NewRemoteConnection(peer.Peer, nil, tcpConn.RemoteAddr().String())
|
||||
NewLocalConnection(connRemote, acceptNewPeer, tcpConn, udpAddr, peer.Router)
|
||||
connLocal := NewLocalConnection(connRemote, tcpConn, udpAddr, peer.Router)
|
||||
connLocal.Start(acceptNewPeer)
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
+12
-10
@@ -33,31 +33,32 @@ func NewRouter(iface *net.Interface, name PeerName, password []byte, connLimit i
|
||||
if len(password) > 0 {
|
||||
router.Password = &password
|
||||
}
|
||||
router.Ourself = StartLocalPeer(name, router)
|
||||
router.Ourself = NewLocalPeer(name, router)
|
||||
router.Peers = NewPeers(router.Ourself.Peer, router.Macs, onPeerGC)
|
||||
router.Peers.FetchWithDefault(router.Ourself.Peer)
|
||||
log.Println("Our name is", router.Ourself.Name)
|
||||
|
||||
router.Routes = NewRoutes(router.Ourself.Peer, router.Peers)
|
||||
router.ConnectionMaker = NewConnectionMaker(router.Ourself, router.Peers)
|
||||
return router
|
||||
}
|
||||
|
||||
func (router *Router) UsingPassword() bool {
|
||||
return router.Password != nil
|
||||
}
|
||||
|
||||
func (router *Router) Start() {
|
||||
// we need two pcap handles since they aren't thread-safe
|
||||
pio, err := NewPcapIO(router.Iface.Name, router.BufSz)
|
||||
checkFatal(err)
|
||||
po, err := NewPcapO(router.Iface.Name)
|
||||
checkFatal(err)
|
||||
router.Routes = StartRoutes(router.Ourself.Peer, router.Peers)
|
||||
router.ConnectionMaker = StartConnectionMaker(router.Ourself, router.Peers)
|
||||
router.Ourself.Start()
|
||||
router.Routes.Start()
|
||||
router.ConnectionMaker.Start()
|
||||
router.UDPListener = router.listenUDP(Port, po)
|
||||
router.listenTCP(Port)
|
||||
router.sniff(pio)
|
||||
}
|
||||
|
||||
func (router *Router) UsingPassword() bool {
|
||||
return router.Password != nil
|
||||
}
|
||||
|
||||
func (router *Router) Status() string {
|
||||
var buf bytes.Buffer
|
||||
buf.WriteString(fmt.Sprintln("Our name is", router.Ourself.Name))
|
||||
@@ -159,7 +160,8 @@ func (router *Router) acceptTCP(tcpConn *net.TCPConn) {
|
||||
remoteAddrStr := tcpConn.RemoteAddr().String()
|
||||
log.Printf("->[%s] connection accepted\n", remoteAddrStr)
|
||||
connRemote := NewRemoteConnection(router.Ourself.Peer, nil, remoteAddrStr)
|
||||
NewLocalConnection(connRemote, true, tcpConn, nil, router)
|
||||
connLocal := NewLocalConnection(connRemote, tcpConn, nil, router)
|
||||
connLocal.Start(true)
|
||||
}
|
||||
|
||||
func (router *Router) listenUDP(localPort int, po PacketSink) *net.UDPConn {
|
||||
|
||||
+17
-14
@@ -16,6 +16,23 @@ type Routes struct {
|
||||
queryChan chan<- *Interaction
|
||||
}
|
||||
|
||||
func NewRoutes(ourself *Peer, peers *Peers) *Routes {
|
||||
routes := &Routes{
|
||||
ourself: ourself,
|
||||
peers: peers,
|
||||
unicast: make(map[PeerName]PeerName),
|
||||
broadcast: make(map[PeerName][]PeerName)}
|
||||
routes.unicast[ourself.Name] = UnknownPeerName
|
||||
routes.broadcast[ourself.Name] = []PeerName{}
|
||||
return routes
|
||||
}
|
||||
|
||||
func (routes *Routes) Start() {
|
||||
queryChan := make(chan *Interaction, ChannelSize)
|
||||
routes.queryChan = queryChan
|
||||
go routes.queryLoop(queryChan)
|
||||
}
|
||||
|
||||
func (routes *Routes) Unicast(name PeerName) (PeerName, bool) {
|
||||
routes.RLock()
|
||||
defer routes.RUnlock()
|
||||
@@ -48,20 +65,6 @@ func (routes *Routes) String() string {
|
||||
return buf.String()
|
||||
}
|
||||
|
||||
func StartRoutes(ourself *Peer, peers *Peers) *Routes {
|
||||
queryChan := make(chan *Interaction, ChannelSize)
|
||||
state := &Routes{
|
||||
ourself: ourself,
|
||||
peers: peers,
|
||||
unicast: make(map[PeerName]PeerName),
|
||||
broadcast: make(map[PeerName][]PeerName),
|
||||
queryChan: queryChan}
|
||||
state.unicast[ourself.Name] = UnknownPeerName
|
||||
state.broadcast[ourself.Name] = []PeerName{}
|
||||
go state.queryLoop(queryChan)
|
||||
return state
|
||||
}
|
||||
|
||||
// ACTOR client API
|
||||
|
||||
const (
|
||||
|
||||
@@ -122,6 +122,7 @@ func main() {
|
||||
}
|
||||
|
||||
router := weave.NewRouter(iface, ourName, []byte(password), connLimit, bufSz*1024*1024, logFrame)
|
||||
log.Println("Our name is", router.Ourself.Name)
|
||||
router.Start()
|
||||
for _, peer := range peers {
|
||||
if addr, err := net.ResolveTCPAddr("tcp4", weave.NormalisePeerAddr(peer)); err == nil {
|
||||
|
||||
Reference in new issue
Block a user