Merge branch 'torStatus' of dan/libricochet-go into master

connectivity/acn.go

@@ -25,6 +25,8 @@ type ACN interface {
 	GetBootstrapStatus() (int, string)
 	// WaitTillBootstrapped Blocks until underlying network is bootstrapped
 	WaitTillBootstrapped()
+	// Sets the calback function to be called when ACN status changes
+	SetStatusCallback(callback func(int, string))
 
 	// Open takes a hostname and returns a net.conn to the derived endpoint
 	// Open allows a client to resolve various hostnames to connections

connectivity/localProvider.go

@@ -34,6 +34,10 @@ func (lp *localProvider) GetBootstrapStatus() (int, string) {
 	return 100, "Done"
 }
 
+func (lp *localProvider) SetStatusCallback(callback func(int, string)) {
+	// nop
+}
+
 // WaitTillBootstrapped Blocks until underlying network is bootstrapped
 func (lp *localProvider) WaitTillBootstrapped() {
 }

connectivity/torProvider.go

@@ -30,6 +30,11 @@ const (
 	CannotDialRicochetAddressError = utils.Error("CannotDialRicochetAddressError")
 )
 
+const (
+	minStatusIntervalMs = 200
+	maxStatusIntervalMs = 2000
+)
+
 type onionListenService struct {
 	os *tor.OnionService
 	tp *torProvider
@@ -42,6 +47,7 @@ type torProvider struct {
 	lock            sync.Mutex
 	breakChan       chan bool
 	childListeners  map[string]*onionListenService
+	statusCallback  func(int, string)
 }
 
 func (ols *onionListenService) AddressFull() string {
@@ -61,14 +67,14 @@ func (ols *onionListenService) Close() {
 	ols.os.Close()
 }
 
-// GetBootstrapStatus returns an int 0-100 on the percent the bootstrapping of the underlying network is at and an optional string message
+// GetBootstrapStatus returns an int -1 on error or 0-100 on the percent the bootstrapping of the underlying network is at and an optional string message
 func (tp *torProvider) GetBootstrapStatus() (int, string) {
 	if tp.t == nil {
-		return 0, "error: no tor, trying to restart..."
+		return -1, "error: no tor, trying to restart..."
 	}
 	kvs, err := tp.t.Control.GetInfo("status/bootstrap-phase")
 	if err != nil {
-		return 0, "error"
+		return -1, "error"
 	}
 	progress := 0
 	status := ""
@@ -186,11 +192,25 @@ func (tp *torProvider) Close() {
 	}
 }
 
+func (tp *torProvider) SetStatusCallback(callback func(int, string)) {
+	tp.lock.Lock()
+	defer tp.lock.Unlock()
+	tp.statusCallback = callback
+}
+
 // StartTor creates/starts a Tor ACN and returns a usable ACN object
 func StartTor(appDirectory string, bundledTorPath string) (ACN, error) {
+	tp, err := startTor(appDirectory, bundledTorPath)
+	if err == nil {
+		go tp.monitorRestart()
+	}
+	return tp, err
+}
+
+func startTor(appDirectory string, bundledTorPath string) (*torProvider, error) {
 	dataDir := path.Join(appDirectory, "tor")
 	os.MkdirAll(dataDir, 0700)
-	tp := &torProvider{appDirectory: appDirectory, bundeledTorPath: bundledTorPath, childListeners: make(map[string]*onionListenService), breakChan: make(chan bool)}
+	tp := &torProvider{appDirectory: appDirectory, bundeledTorPath: bundledTorPath, childListeners: make(map[string]*onionListenService), breakChan: make(chan bool), statusCallback: nil}
 
 	// attempt connect to system tor
 	log.Debugf("dialing system tor control port\n")
@@ -205,7 +225,6 @@ func StartTor(appDirectory string, bundledTorPath string) (ACN, error) {
 			if err == nil && minTorVersionReqs(pinfo.TorVersion) {
 				log.Debugln("OK version " + pinfo.TorVersion)
 				tp.t = createFromExisting(controlport, dataDir)
-				go tp.monitorRestart()
 				return tp, nil
 			}
 			controlport.Close()
@@ -217,7 +236,6 @@ func StartTor(appDirectory string, bundledTorPath string) (ACN, error) {
 		t, err := tor.Start(nil, &tor.StartConf{EnableNetwork: true, DataDir: dataDir, DebugWriter: nil})
 		if err == nil {
 			tp.t = t
-			go tp.monitorRestart()
 			return tp, nil
 		}
 		log.Debugf("Error connecting to self-run system tor: %v\n", err)
@@ -231,9 +249,6 @@ func StartTor(appDirectory string, bundledTorPath string) (ACN, error) {
 			log.Debugf("Error running bundled tor: %v\n", err)
 		}
 		tp.t = t
-		if err == nil {
-			go tp.monitorRestart()
-		}
 		return tp, err
 	}
 	return nil, errors.New("Could not connect to or start Tor that met requirments (min Tor version 0.3.5.x)")
@@ -246,41 +261,55 @@ func (tp *torProvider) unregisterListener(id string) {
 }
 
 func (tp *torProvider) monitorRestart() {
+	lastBootstrapProgress := 0
+	interval := minStatusIntervalMs
+
 	for {
 		select {
-		case <-time.After(time.Duration(30 * time.Second)):
+		case <-time.After(time.Millisecond * time.Duration(interval)):
-			tp.lock.Lock()
+			prog, status := tp.GetBootstrapStatus()
-			if tp.t != nil {
-				_, err := tp.t.Control.GetInfo("version")
 
-				if err != nil {
+			if prog == -1 && tp.t != nil {
-					tp.lock.Unlock()
+				if tp.statusCallback != nil {
-					for _, child := range tp.childListeners {
+					tp.statusCallback(prog, status)
-						child.Close()
+				}
-					}
+				tp.restart()
-					tp.lock.Lock()
+				interval = minStatusIntervalMs
-					tp.t.Close()
+			} else if prog != lastBootstrapProgress {
-					tp.t = nil
+				if tp.statusCallback != nil {
+					tp.statusCallback(prog, status)
+				}
+				interval = minStatusIntervalMs
+			} else {
+				if interval < maxStatusIntervalMs {
+					interval *= 2
 				}
 			}
-
+			lastBootstrapProgress = prog
-			if tp.t == nil {
-				newACN, err := StartTor(tp.appDirectory, tp.bundeledTorPath)
-				if err == nil {
-					switch newTp := newACN.(type) {
-					case *torProvider:
-						tp.t = newTp.t
-						// startTor will have started a new monitorRestart thread
-						tp.lock.Unlock()
-						return
-					}
-				}
-			}
-			tp.lock.Unlock()
 		case <-tp.breakChan:
 			return
 		}
+	}
+}
 
+func (tp *torProvider) restart() {
+
+	for _, child := range tp.childListeners {
+		child.Close()
+	}
+
+	tp.lock.Lock()
+	defer tp.lock.Unlock()
+
+	tp.t.Close()
+	tp.t = nil
+
+	for {
+		newTp, err := startTor(tp.appDirectory, tp.bundeledTorPath)
+		if err == nil {
+			tp.t = newTp.t
+			return
+		}
 	}
 }
 

connectivity/torProvider_test.go

@@ -3,21 +3,26 @@ package connectivity
 import (
 	"fmt"
 	"testing"
-	"time"
 )
 
+func getStatusCallback(progChan chan int) func(int, string) {
+	return func(prog int, status string) {
+		fmt.Printf("%v %v\n", prog, status)
+		progChan <- prog
+	}
+}
+
 func TestTorProvider(t *testing.T) {
+	progChan := make(chan int)
 	acn, err := StartTor(".", "")
+	acn.SetStatusCallback(getStatusCallback(progChan))
 	if err != nil {
 		t.Error(err)
 	}
 
 	progress := 0
-	status := ""
 	for progress < 100 {
-		progress, status = acn.GetBootstrapStatus()
+		progress = <-progChan
-		fmt.Printf("%v %v\n", progress, status)
-		time.Sleep(100)
 	}
 
 	acn.Close()

testing/tests.sh

@@ -4,7 +4,7 @@
 
 set -e
 pwd
-go test ${1} -coverprofile=model.cover.out -v ./utils
+go test ${1} -coverprofile=utils.cover.out -v ./utils
 go test ${1} -coverprofile=channels.cover.out -v ./channels
 go test ${1} -coverprofile=channels.v3.inbound.cover.out -v ./channels/v3/inbound
 go test ${1} -coverprofile=connection.cover.out -v ./connection