package config

import (
	"path"
	"sync"
	"time"
)

type Processor interface {
	Process(listener ChangedListener)
}
type watchProcessor struct {
	lastIndex uint64
	prefix    string
	keys      []string
	stopChan  chan bool
	errChan   chan error
	wg        sync.WaitGroup
	store     StoreClient
}

func WatchProcessor(prefix string, keys []string, stopChan chan bool, errChan chan error, store StoreClient) Processor {
	var wg sync.WaitGroup
	s := make([]string, len(keys))
	for i, k := range keys {
		s[i] = path.Join(prefix, k)
	}
	return &watchProcessor{0, prefix, s, stopChan, errChan, wg, store}
}

func (p *watchProcessor) Process(listener ChangedListener) {
	p.wg.Add(1)
	go p.monitorPrefix(listener)
	p.wg.Wait()
}

func (p *watchProcessor) monitorPrefix(listener ChangedListener) {
	defer p.wg.Done()
	for {
		index, err := p.store.WatchPrefix(p.prefix, p.keys, p.lastIndex, p.stopChan)
		if err != nil {
			p.errChan <- err
			time.Sleep(time.Second * 2) // 防止后端错误占用所有资源.
			continue
		}
		p.lastIndex = index
		if vl, err := p.store.GetValues(p.keys); err == nil {
			listener.Changed(vl)
		} else {
			p.errChan <- err
		}
	}
}