func (r *registryRouter) watch() {
	var attempts int
	for {
		if r.isClosed() {
			return
		}
		w, err := r.opts.Registry.Watch(registry.WatchDomain(registry.WildcardDomain))
		if err != nil {
			attempts++
			if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
				logger.Errorf("error watching endpoints: %v", err)
			}
			time.Sleep(time.Duration(attempts) * time.Second)
			continue
		}
		ch := make(chan bool)
		go func() {
			select {
			case <-ch:
				w.Stop()
			case <-r.exit:
				w.Stop()
			}
		}()
		attempts = 0
		for {
			res, err := w.Next()
			if err != nil {
				if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
					logger.Errorf("error getting next endpoint: %v", err)
				}
				close(ch)
				break
			}
			r.process(res)
		}
	}
}
watch for endpoint changes
package registry
import (
	"errors"
	"fmt"
	"net/http"
	"regexp"
	"strings"
	"sync"
	"time"
	"github.com/micro/micro/v3/service/api"
	"github.com/micro/micro/v3/service/api/router"
	"github.com/micro/micro/v3/service/context/metadata"
	"github.com/micro/micro/v3/service/logger"
	"github.com/micro/micro/v3/service/registry"
	"github.com/micro/micro/v3/service/registry/cache"
	"github.com/micro/micro/v3/util/namespace"
	util "github.com/micro/micro/v3/util/router"
)
var (
	errEmptyNamespace = errors.New("namespace is empty")
	errNotFound       = errors.New("not found")
)
type endpoint struct {
	hostregs []*regexp.Regexp
	pathregs []util.Pattern
	pcreregs []*regexp.Regexp
}
type namespaceEntry struct {
	sync.RWMutex
	eps map[string]*api.Service
	ceps map[string]*endpoint
}
type registryRouter struct {
	exit chan bool
	opts router.Options
	rc cache.Cache
	refreshChan chan string
	sync.RWMutex
	namespaces map[string]*namespaceEntry
}
func getDomain(srv *registry.Service) string {
	if srv.Metadata != nil && len(srv.Metadata["domain"]) > 0 {
		return srv.Metadata["domain"]
	} else if nodes := srv.Nodes; len(nodes) > 0 && nodes[0].Metadata != nil {
		if len(nodes[0].Metadata["domain"]) > 0 {
			return nodes[0].Metadata["domain"]
		}
	}
	return registry.DefaultDomain
}
func (r *registryRouter) isClosed() bool {
	select {
	case <-r.exit:
		return true
	default:
		return false
	}
}
func (r *registryRouter) refreshNamespace(ns string) error {
	services, err := r.opts.Registry.ListServices(registry.ListDomain(ns))
	if err != nil {
		if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
			logger.Errorf("unable to list services: %v", err)
		}
		return err
	}
	if len(services) == 0 {
		return errEmptyNamespace
	}
	for _, s := range services {
		dns := getDomain(s)
		if len(s.Nodes) > 0 && len(dns) > 0 {
			r.store(dns, []*registry.Service{s})
			continue
		}
		service, err := r.rc.GetService(s.Name, registry.GetDomain(ns))
		if err != nil {
			if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
				logger.Errorf("unable to get service: %v", err)
			}
			continue
		}
		for _, srv := range service {
			ns = getDomain(srv)
			r.store(ns, []*registry.Service{srv})
		}
	}
	return nil
}
func (r *registryRouter) refresh() {
	refreshed := make(map[string]time.Time)
	r.refreshNamespace(registry.WildcardDomain)
	for {
		r.RLock()
		namespaces := r.namespaces
		r.RUnlock()
		for ns, _ := range namespaces {
			err := r.refreshNamespace(ns)
			if err == errEmptyNamespace {
				r.Lock()
				delete(namespaces, ns)
				r.Unlock()
			}
		}
		select {
		case domain := <-r.refreshChan:
			v, ok := refreshed[domain]
			if ok && time.Since(v) < time.Minute {
				break
			}
			r.refreshNamespace(domain)
		case <-time.After(time.Minute):
		case <-r.exit:
			return
		}
	}
}
func (r *registryRouter) process(res *registry.Result) {
	if res == nil || res.Service == nil {
		return
	}
	service, err := r.rc.GetService(res.Service.Name)
	if err != nil {
		if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
			logger.Errorf("unable to get %v service: %v", res.Service.Name, err)
		}
		return
	}
	if len(service) == 0 {
		return
	}
	namespace := getDomain(service[0])
	r.store(namespace, service)
}
func (r *registryRouter) store(namespace string, services []*registry.Service) {
	eps := map[string]*api.Service{}
	names := map[string]bool{}
	for _, service := range services {
		names[service.Name] = true
		for _, sep := range service.Endpoints {
			key := fmt.Sprintf("%s.%s", service.Name, sep.Name)
			end := api.Decode(sep.Metadata)
			if end == nil || len(end.Name) == 0 {
				continue
			}
			if err := api.Validate(end); err != nil {
				if logger.V(logger.TraceLevel, logger.DefaultLogger) {
					logger.Tracef("endpoint validation failed: %v", err)
				}
				continue
			}
			ep, ok := eps[key]
			if !ok {
				ep = &api.Service{Name: service.Name}
			}
			ep.Endpoint = end
			ep.Services = append(ep.Services, service)
			eps[key] = ep
		}
	}
	r.Lock()
	nse, ok := r.namespaces[namespace]
	if !ok {
		nse = &namespaceEntry{
			eps:  map[string]*api.Service{},
			ceps: map[string]*endpoint{},
		}
		r.namespaces[namespace] = nse
	}
	r.Unlock()
	nse.Lock()
	defer nse.Unlock()
	for key, service := range nse.eps {
		if !names[service.Name] {
			continue
		}
		delete(nse.eps, key)
	}
	for name, ep := range eps {
		nse.eps[name] = ep
		cep := &endpoint{}
		for _, h := range ep.Endpoint.Host {
			if h == "" || h == "*" {
				continue
			}
			hostreg, err := regexp.CompilePOSIX(h)
			if err != nil {
				if logger.V(logger.TraceLevel, logger.DefaultLogger) {
					logger.Tracef("endpoint have invalid host regexp: %v", err)
				}
				continue
			}
			cep.hostregs = append(cep.hostregs, hostreg)
		}
		for _, p := range ep.Endpoint.Path {
			var pcreok bool
			if p[0] == '^' && p[len(p)-1] == '$' {
				pcrereg, err := regexp.CompilePOSIX(p)
				if err == nil {
					cep.pcreregs = append(cep.pcreregs, pcrereg)
					pcreok = true
				}
			}
			rule, err := util.Parse(p)
			if err != nil && !pcreok {
				if logger.V(logger.TraceLevel, logger.DefaultLogger) {
					logger.Tracef("endpoint have invalid path pattern: %v", err)
				}
				continue
			} else if err != nil && pcreok {
				continue
			}
			tpl := rule.Compile()
			pathreg, err := util.NewPattern(tpl.Version, tpl.OpCodes, tpl.Pool, "")
			if err != nil {
				if logger.V(logger.TraceLevel, logger.DefaultLogger) {
					logger.Tracef("endpoint have invalid path pattern: %v", err)
				}
				continue
			}
			cep.pathregs = append(cep.pathregs, pathreg)
		}
		nse.ceps[name] = cep
	}
}