Merge pull request #775 from micro/proxy-watcher

Fix the proxy watcher

proxy/mucp/mucp.go

@@ -8,6 +8,7 @@ import (
 	"sort"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/micro/go-micro/client"
 	"github.com/micro/go-micro/codec"
@@ -440,7 +441,16 @@ func NewProxy(opts ...options.Option) proxy.Proxy {
 
 	// watch router service routes
 	p.errChan = make(chan error, 1)
-	go p.watchRoutes()
+
+	go func() {
+		// continuously attempt to watch routes
+		for {
+			// watch the routes
+			p.watchRoutes()
+			// in case of failure just wait a second
+			time.Sleep(time.Second)
+		}
+	}()
 
 	return p
 }

router/handler/router.go

@@ -164,7 +164,7 @@ func (r *Router) Watch(ctx context.Context, req *pb.WatchRequest, stream pb.Rout
 	for {
 		event, err := watcher.Next()
 		if err == router.ErrWatcherStopped {
-			break
+			return errors.InternalServerError("go.micro.router", "watcher stopped")
 		}
 
 		if err != nil {

router/service/service.go

@@ -357,7 +357,9 @@ func (s *svc) Watch(opts ...router.WatchOption) (router.Watcher, error) {
 	if err != nil {
 		return nil, err
 	}
-	var options router.WatchOptions
+	options := router.WatchOptions{
+		Service: "*",
+	}
 	for _, o := range opts {
 		o(&options)
 	}

router/service/watcher.go

@@ -70,13 +70,11 @@ func (w *watcher) watch(stream pb.Router_WatchService) error {
 			Route:     route,
 		}
 
-		for {
 		select {
 		case w.resChan <- event:
 		case <-w.done:
 		}
 	}
-	}
 
 	return watchErr
 }