Strip Advertise/Process from router

registry.go

@@ -2,12 +2,10 @@ package registry
 
 import (
 	"fmt"
-	"sort"
 	"strings"
 	"sync"
 	"time"
 
-	"github.com/google/uuid"
 	"github.com/micro/go-micro/v3/logger"
 	"github.com/micro/go-micro/v3/registry"
 	"github.com/micro/go-micro/v3/router"
@@ -18,26 +16,17 @@ var (
 	RefreshInterval = time.Second * 120
 	// PruneInterval is how often we prune the routing table
 	PruneInterval = time.Second * 10
-	// AdvertiseEventsTick is time interval in which the router advertises route updates
-	AdvertiseEventsTick = 10 * time.Second
-	// DefaultAdvertTTL is default advertisement TTL
-	DefaultAdvertTTL = 2 * time.Minute
 )
 
 // rtr implements router interface
 type rtr struct {
 	sync.RWMutex
 
-	running   bool
-	table     *table
-	options   router.Options
-	exit      chan bool
-	initChan  chan bool
-	eventChan chan *router.Event
-
-	// advert subscribers
-	sub         sync.RWMutex
-	subscribers map[string]chan *router.Advert
+	running  bool
+	table    *table
+	options  router.Options
+	exit     chan bool
+	initChan chan bool
 }
 
 // NewRouter creates new router and returns it
@@ -52,9 +41,8 @@ func NewRouter(opts ...router.Option) router.Router {
 
 	// construct the router
 	r := &rtr{
-		options:     options,
-		initChan:    make(chan bool),
-		subscribers: make(map[string]chan *router.Advert),
+		options:  options,
+		initChan: make(chan bool),
 	}
 
 	// create the new table, passing the fetchRoute method in as a fallback if
@@ -327,214 +315,6 @@ func (r *rtr) watchRegistry(w registry.Watcher) error {
 	return nil
 }
 
-// watchTable watches routing table entries and either adds or deletes locally registered service to/from network registry
-// It returns error if the locally registered services either fails to be added/deleted to/from network registry.
-func (r *rtr) watchTable(w router.Watcher) error {
-	exit := make(chan bool)
-
-	defer func() {
-		close(exit)
-	}()
-
-	// wait in the background for the router to stop
-	// when the router stops, stop the watcher and exit
-	go func() {
-		defer w.Stop()
-
-		select {
-		case <-r.exit:
-			return
-		case <-exit:
-			return
-		}
-	}()
-
-	for {
-		event, err := w.Next()
-		if err != nil {
-			if err != router.ErrWatcherStopped {
-				return err
-			}
-			break
-		}
-
-		select {
-		case <-r.exit:
-			return nil
-		case r.eventChan <- event:
-			// process event
-		}
-	}
-
-	return nil
-}
-
-// publishAdvert publishes router advert to advert channel
-func (r *rtr) publishAdvert(advType router.AdvertType, events []*router.Event) {
-	a := &router.Advert{
-		Id:        r.options.Id,
-		Type:      advType,
-		TTL:       DefaultAdvertTTL,
-		Timestamp: time.Now(),
-		Events:    events,
-	}
-
-	r.sub.RLock()
-	for _, sub := range r.subscribers {
-		// now send the message
-		select {
-		case sub <- a:
-		case <-r.exit:
-			r.sub.RUnlock()
-			return
-		}
-	}
-	r.sub.RUnlock()
-}
-
-// adverts maintains a map of router adverts
-type adverts map[uint64]*router.Event
-
-// advertiseEvents advertises routing table events
-// It suppresses unhealthy flapping events and advertises healthy events upstream.
-func (r *rtr) advertiseEvents() error {
-	// ticker to periodically scan event for advertising
-	ticker := time.NewTicker(AdvertiseEventsTick)
-	defer ticker.Stop()
-
-	// adverts is a map of advert events
-	adverts := make(adverts)
-
-	// routing table watcher
-	w, err := r.Watch()
-	if err != nil {
-		return err
-	}
-	defer w.Stop()
-
-	go func() {
-		var err error
-
-		for {
-			select {
-			case <-r.exit:
-				return
-			default:
-				if w == nil {
-					// routing table watcher
-					w, err = r.Watch()
-					if err != nil {
-						if logger.V(logger.DebugLevel, logger.DefaultLogger) {
-							logger.Debugf("Error creating watcher: %v", err)
-						}
-						time.Sleep(time.Second)
-						continue
-					}
-				}
-
-				if err := r.watchTable(w); err != nil {
-					if logger.V(logger.DebugLevel, logger.DefaultLogger) {
-						logger.Debugf("Error watching table: %v", err)
-					}
-					time.Sleep(time.Second)
-				}
-
-				if w != nil {
-					// reset
-					w.Stop()
-					w = nil
-				}
-			}
-		}
-	}()
-
-	for {
-		select {
-		case <-ticker.C:
-			// If we're not advertising any events then sip processing them entirely
-			if r.options.Advertise == router.AdvertiseNone {
-				continue
-			}
-
-			var events []*router.Event
-
-			// collect all events which are not flapping
-			for key, event := range adverts {
-				// if we only advertise local routes skip processing anything not link local
-				if r.options.Advertise == router.AdvertiseLocal && event.Route.Link != "local" {
-					continue
-				}
-
-				// copy the event and append
-				e := new(router.Event)
-				// this is ok, because router.Event only contains builtin types
-				// and no references so this creates a deep copy of struct Event
-				*e = *event
-				events = append(events, e)
-				// delete the advert from adverts
-				delete(adverts, key)
-			}
-
-			// advertise events to subscribers
-			if len(events) > 0 {
-				if logger.V(logger.DebugLevel, logger.DefaultLogger) {
-					logger.Debugf("Router publishing %d events", len(events))
-				}
-				go r.publishAdvert(router.RouteUpdate, events)
-			}
-		case e := <-r.eventChan:
-			// if event is nil, continue
-			if e == nil {
-				continue
-			}
-
-			// If we're not advertising any events then skip processing them entirely
-			if r.options.Advertise == router.AdvertiseNone {
-				continue
-			}
-
-			// if we only advertise local routes skip processing anything not link local
-			if r.options.Advertise == router.AdvertiseLocal && e.Route.Link != "local" {
-				continue
-			}
-
-			if logger.V(logger.DebugLevel, logger.DefaultLogger) {
-				logger.Debugf("Router processing table event %s for service %s %s", e.Type, e.Route.Service, e.Route.Address)
-			}
-
-			// check if we have already registered the route
-			hash := e.Route.Hash()
-			ev, ok := adverts[hash]
-			if !ok {
-				ev = e
-				adverts[hash] = e
-				continue
-			}
-
-			// override the route event only if the previous event was different
-			if ev.Type != e.Type {
-				ev = e
-			}
-		case <-r.exit:
-			if w != nil {
-				w.Stop()
-			}
-			return nil
-		}
-	}
-}
-
-// drain all the events, only called on Stop
-func (r *rtr) drain() {
-	for {
-		select {
-		case <-r.eventChan:
-		default:
-			return
-		}
-	}
-}
-
 // start the router. Should be called under lock.
 func (r *rtr) start() error {
 	if r.running {
@@ -621,130 +401,6 @@ func (r *rtr) start() error {
 	return nil
 }
 
-// Advertise stars advertising the routes to the network and returns the advertisements channel to consume from.
-// If the router is already advertising it returns the channel to consume from.
-// It returns error if either the router is not running or if the routing table fails to list the routes to advertise.
-func (r *rtr) Advertise() (<-chan *router.Advert, error) {
-	r.Lock()
-	defer r.Unlock()
-
-	// we're mutating the subscribers so they need to be locked also
-	r.sub.Lock()
-	defer r.sub.Unlock()
-
-	// already advertising
-	if r.eventChan != nil {
-		advertChan := make(chan *router.Advert, 128)
-		r.subscribers[uuid.New().String()] = advertChan
-		return advertChan, nil
-	}
-
-	// list all the routes and pack them into even slice to advertise
-	events, err := r.flushRouteEvents(router.Create)
-	if err != nil {
-		return nil, fmt.Errorf("failed to flush routes: %s", err)
-	}
-
-	// create event channels
-	r.eventChan = make(chan *router.Event)
-
-	// create advert channel
-	advertChan := make(chan *router.Advert, 128)
-	r.subscribers[uuid.New().String()] = advertChan
-
-	// advertise your presence
-	go r.publishAdvert(router.Announce, events)
-
-	go func() {
-		select {
-		case <-r.exit:
-			return
-		default:
-			if err := r.advertiseEvents(); err != nil {
-				if logger.V(logger.DebugLevel, logger.DefaultLogger) {
-					logger.Debugf("Error adveritising events: %v", err)
-				}
-			}
-		}
-	}()
-
-	return advertChan, nil
-
-}
-
-// Process updates the routing table using the advertised values
-func (r *rtr) Process(a *router.Advert) error {
-	// NOTE: event sorting might not be necessary
-	// copy update events intp new slices
-	events := make([]*router.Event, len(a.Events))
-	copy(events, a.Events)
-	// sort events by timestamp
-	sort.Slice(events, func(i, j int) bool {
-		return events[i].Timestamp.Before(events[j].Timestamp)
-	})
-
-	if logger.V(logger.TraceLevel, logger.DefaultLogger) {
-		logger.Tracef("Router %s processing advert from: %s", r.options.Id, a.Id)
-	}
-
-	for _, event := range events {
-		// skip if the router is the origin of this route
-		if event.Route.Router == r.options.Id {
-			if logger.V(logger.TraceLevel, logger.DefaultLogger) {
-				logger.Tracef("Router skipping processing its own route: %s", r.options.Id)
-			}
-			continue
-		}
-		// create a copy of the route
-		route := event.Route
-		action := event.Type
-
-		if logger.V(logger.TraceLevel, logger.DefaultLogger) {
-			logger.Tracef("Router %s applying %s from router %s for service %s %s", r.options.Id, action, route.Router, route.Service, route.Address)
-		}
-
-		if err := r.manageRoute(route, action.String()); err != nil {
-			return fmt.Errorf("failed applying action %s to routing table: %s", action, err)
-		}
-	}
-
-	return nil
-}
-
-// flushRouteEvents returns a slice of events, one per each route in the routing table
-func (r *rtr) flushRouteEvents(evType router.EventType) ([]*router.Event, error) {
-	// get a list of routes for each service in our routing table
-	// for the configured advertising strategy
-	q := []router.QueryOption{
-		router.QueryStrategy(r.options.Advertise),
-	}
-
-	routes, err := r.table.Query(q...)
-	if err != nil && err != router.ErrRouteNotFound {
-		return nil, err
-	}
-
-	if logger.V(logger.DebugLevel, logger.DefaultLogger) {
-		logger.Debugf("Router advertising %d routes with strategy %s", len(routes), r.options.Advertise)
-	}
-
-	// build a list of events to advertise
-	events := make([]*router.Event, len(routes))
-	var i int
-
-	for _, route := range routes {
-		event := &router.Event{
-			Type:      evType,
-			Timestamp: time.Now(),
-			Route:     route,
-		}
-		events[i] = event
-		i++
-	}
-
-	return events, nil
-}
-
 // Lookup routes in the routing table
 func (r *rtr) Lookup(q ...router.QueryOption) ([]router.Route, error) {
 	return r.Table().Query(q...)
@@ -769,24 +425,6 @@ func (r *rtr) Close() error {
 		}
 		close(r.exit)
 
-		// extract the events
-		r.drain()
-
-		r.sub.Lock()
-		// close advert subscribers
-		for id, sub := range r.subscribers {
-			// close the channel
-			close(sub)
-			// delete the subscriber
-			delete(r.subscribers, id)
-		}
-		r.sub.Unlock()
-	}
-
-	// close and remove event chan
-	if r.eventChan != nil {
-		close(r.eventChan)
-		r.eventChan = nil
 	}
 
 	r.running = false