Function RunSourcePartitionSeqs

ns_server.go:1034–1135

(options map[string]string, stopCh chan struct{})

}

func RunSourcePartitionSeqs(options map[string]string, stopCh chan struct{}) {
	sourcePartitionSeqsSleep := SourcePartitionSeqsSleepDefault
	v, exists := options["sourcePartitionSeqsSleepMS"]
	if exists {
		sourcePartitionSeqsSleepMS, err := strconv.Atoi(v)
		if err != nil {
			log.Warnf("ns_server: parse sourcePartitionSeqsSleepMS: %q,"+
				" err: %v", v, err)
		} else {
			sourcePartitionSeqsSleep = time.Millisecond *
				time.Duration(sourcePartitionSeqsSleepMS)
		}
	}

	sourcePartitionSeqsCacheTimeout := SourcePartitionSeqsCacheTimeoutDefault
	v, exists = options["sourcePartitionSeqsCacheTimeoutMS"]
	if exists {
		sourcePartitionSeqsCacheTimeoutMS, err := strconv.Atoi(v)
		if err != nil {
			log.Warnf("ns_server: parse sourcePartitionSeqsCacheTimeoutMS: %q,"+
				" err: %v", v, err)
		} else {
			sourcePartitionSeqsCacheTimeout = time.Millisecond *
				time.Duration(sourcePartitionSeqsCacheTimeoutMS)
		}
	}

	m := &mapSourcePartitionSeqsM

	for {
		select {
		case <-stopCh:
			return
		case <-time.After(sourcePartitionSeqsSleep):
			// NO-OP.
		}

		m.Lock()
		var sourceSpecs []SourceSpec // Snapshot the wanted sourceSpecs.
		for sourceSpec := range mapSourcePartitionSeqs {
			sourceSpecs = append(sourceSpecs, sourceSpec)
		}
		m.Unlock()

		for _, sourceSpec := range sourceSpecs {
			select {
			case <-stopCh:
				return
			default:
				// NO-OP.
			}

			m.Lock()
			s := SourcePartitionSeqs{}
			v, exists := mapSourcePartitionSeqs[sourceSpec]
			if exists && v != nil {
				s = *v // Copy fields.
			}