// StartMonitorFeed starts a goroutine which processes events published to the
// supplied Subscription. The goroutine will continue running until the
// Subscription's Events feed is closed.
func (nsm *NodeStatusMonitor) StartMonitorFeed(feed *util.Feed) {
	sub := feed.Subscribe()
	go func() {
		for event := range sub.Events() {
			if syncEvent, ok := event.(*TestSyncEvent); ok {
				syncEvent.consume()
			}

			ProcessNodeEvent(nsm, event)
			storage.ProcessStoreEvent(nsm, event)
		}
	}()
}

// StartMonitorFeed starts a goroutine which processes events published to the
// supplied Subscription. The goroutine will continue running until the
// Subscription's Events feed is closed.
func (nsm *NodeStatusMonitor) StartMonitorFeed(feed *util.Feed) {
	go storage.ProcessStoreEvents(nsm, feed.Subscribe())
	go ProcessNodeEvents(nsm, feed.Subscribe())
}

func (ner *nodeEventReader) readEvents(feed *util.Feed) {
	ner.perNodeFeeds = make(map[roachpb.NodeID][]string)
	feed.Subscribe(ner.recordEvent)
}

func (ser *storeEventReader) readEvents(feed *util.Feed) {
	ser.perStoreFeeds = make(map[roachpb.StoreID][]string)
	ser.perStoreUpdateCount = make(map[roachpb.StoreID]map[roachpb.Method]int)
	feed.Subscribe(ser.recordEvent)
}

// StartMonitorFeed starts a goroutine which processes events published to the
// supplied Subscription. The goroutine will continue running until the
// Subscription's Events feed is closed.
func (nsm *NodeStatusMonitor) StartMonitorFeed(feed *util.Feed) {
	feed.Subscribe(func(event interface{}) {
		ProcessNodeEvent(nsm, event)
		storage.ProcessStoreEvent(nsm, event)
	})
}

func newSimpleEventConsumer(feed *util.Feed) *simpleEventConsumer {
	return &simpleEventConsumer{
		sub: feed.Subscribe(),
	}
}

func newConsumer(feed *util.Feed) *storeEventConsumer {
	return &storeEventConsumer{
		sub: feed.Subscribe(),
	}
}