func (ser *storeEventReader) readEvents(sub *util.Subscription) {
	ser.perStoreFeeds = make(map[proto.StoreID][]string)
	ser.perStoreUpdateCount = make(map[proto.StoreID]map[proto.Method]int)
	for e := range sub.Events() {
		ser.recordEvent(e)
	}
}

// ProcessStoreEvents reads store events from the supplied channel and passes
// them to the correct methods of the supplied StoreEventListener. This method
// will run until the Subscription's events channel is closed.
func ProcessStoreEvents(l StoreEventListener, sub *util.Subscription) {
	for event := range sub.Events() {
		// TODO(tamird): https://github.com/barakmich/go-nyet/issues/7
		switch specificEvent := event.(type) {
		case *StartStoreEvent:
			l.OnStartStore(specificEvent)
		case *RegisterRangeEvent:
			l.OnRegisterRange(specificEvent)
		case *UpdateRangeEvent:
			l.OnUpdateRange(specificEvent)
		case *RemoveRangeEvent:
			l.OnRemoveRange(specificEvent)
		case *SplitRangeEvent:
			l.OnSplitRange(specificEvent)
		case *MergeRangeEvent:
			l.OnMergeRange(specificEvent)
		case *BeginScanRangesEvent:
			l.OnBeginScanRanges(specificEvent)
		case *EndScanRangesEvent:
			l.OnEndScanRanges(specificEvent)
		case *StoreStatusEvent:
			l.OnStoreStatus(specificEvent)
		case *ReplicationStatusEvent:
			l.OnReplicationStatus(specificEvent)
		}
	}
}

// ProcessNodeEvents reads node events from the supplied channel and passes them
// to the correct methods of the supplied NodeEventListener. This method will
// run until the Subscription's events channel is closed.
func ProcessNodeEvents(l NodeEventListener, sub *util.Subscription) {
	for event := range sub.Events() {
		// TODO(tamird): https://github.com/barakmich/go-nyet/issues/7
		switch specificEvent := event.(type) {
		case *CallSuccessEvent:
			l.OnCallSuccess(specificEvent)
		case *CallErrorEvent:
			l.OnCallError(specificEvent)
		}
	}
}

func (ner *nodeEventReader) readEvents(sub *util.Subscription) {
	ner.perNodeFeeds = make(map[proto.NodeID][]string)
	for e := range sub.Events() {
		ner.recordEvent(e)
	}
}