Merge branch '23_retry_peers' of https://github.com/bboreham/weave into 23_retry_peers

connection_maker.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"log"
 	"math/rand"
-	"net"
 	"time"
 )
 
@@ -43,10 +42,7 @@ func (cm *ConnectionMaker) InitiateConnection(address string) {
 }
 
 func (cm *ConnectionMaker) EnsureConnection(address string) {
-	// If we've been given a port number, take it off
-	if addrHost, _, err := net.SplitHostPort(address); err == nil {
-		address = addrHost
-	}
+	address = StripPortFromAddr(address)
 	cm.queryChan <- &ConnectionMakerInteraction{
 		Interaction:   Interaction{code: CMEnsure},
 		acceptAnyPeer: false,
@@ -54,10 +50,7 @@ func (cm *ConnectionMaker) EnsureConnection(address string) {
 }
 
 func (cm *ConnectionMaker) ConnectionEstablished(address string) {
-	// If we've been given a port number, take it off
-	if addrHost, _, err := net.SplitHostPort(address); err == nil {
-		address = addrHost
-	}
+	address = StripPortFromAddr(address)
 	cm.queryChan <- &ConnectionMakerInteraction{
 		Interaction: Interaction{code: CMEstablished},
 		address:     address}
@@ -92,6 +85,7 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio
 			}
 			switch {
 			case query.code == CMEnsure:
+				log.Println("Connection ensure", query.address)
 				cm.addToTargets(query.acceptAnyPeer, query.address)
 				cm.resetTarget(query.address)
 				tickNow()
@@ -100,6 +94,7 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio
 			case query.code == CMEstablished:
 				target := cm.targets[query.address]
 				if target != nil {
+					log.Println("Connection established to", query.address, target)
 					target.state = CSEstablished
 				} else {
 					log.Fatal("Connection established to unknown address", query.address)
@@ -115,9 +110,9 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio
 		case now := <-tick:
 			for address, target := range cm.targets {
 				if target.state == CSUnconnected && now.After(target.tryAfter) {
-					if _, found := cm.router.Ourself.ConnectionOn(address); found {
+					/*if _, found := cm.router.Ourself.ConnectionOn(address); found {
 						target.state = CSEstablished
-					} else {
+					} else*/ {
 						target.attemptCount += 1
 						target.state = CSAttempting
 						go cm.attemptConnection(address, target.acceptAnyPeer)

peer.go

@@ -57,19 +57,8 @@ func (peer *Peer) ConnectionTo(name PeerName) (Connection, bool) {
 	return conn, found // yes, you really can't inline that. FFS.
 }
 
-// given an address like '1.2.3.4:567', return the address if it has a port,
-// otherwise return the address with weave's standard port number
-func normalisePeerAddr(peerAddr string) string {
-	_, _, err := net.SplitHostPort(peerAddr)
-	if err == nil {
-		return peerAddr
-	} else {
-		return fmt.Sprintf("%s:%d", peerAddr, Port)
-	}
-}
-
 func (peer *Peer) ConnectionOn(peerAddr string) (Connection, bool) {
-	peerAddr = normalisePeerAddr(peerAddr)
+	peerAddr = NormalisePeerAddr(peerAddr)
 	peer.RLock()
 	defer peer.RUnlock()
 	for _, conn := range peer.connections {
@@ -220,7 +209,7 @@ func (peer *Peer) CreateConnection(peerAddr string, acceptNewPeer bool) error {
 		return err
 	}
 	// We're dialing the remote so that means connections will come from random ports
-	addrStr := normalisePeerAddr(peerAddr)
+	addrStr := NormalisePeerAddr(peerAddr)
 	tcpAddr, tcpErr := net.ResolveTCPAddr("tcp4", addrStr)
 	udpAddr, udpErr := net.ResolveUDPAddr("udp4", addrStr)
 	if tcpErr != nil || udpErr != nil {

utils.go

@@ -97,3 +97,22 @@ func (lop ListOfPeers) Swap(i, j int) {
 func (lop ListOfPeers) Less(i, j int) bool {
 	return lop[i].Name < lop[i].Name
 }
+
+func StripPortFromAddr(address string) string {
+	// If we've been given a port number, take it off
+	if addrHost, _, err := net.SplitHostPort(address); err == nil {
+		return addrHost
+	}
+	return address
+}
+
+// given an address like '1.2.3.4:567', return the address if it has a port,
+// otherwise return the address with weave's standard port number
+func NormalisePeerAddr(peerAddr string) string {
+	_, _, err := net.SplitHostPort(peerAddr)
+	if err == nil {
+		return peerAddr
+	} else {
+		return fmt.Sprintf("%s:%d", peerAddr, Port)
+	}
+}