// Watch the store until either there's a store error or we receive a stop request.
// Returns false if we shouldn't attempt watching the store anymore (stop request received).
func (s *Discovery) watchOnce(stopCh <-chan struct{}, watchCh <-chan []*store.KVPair, discoveryCh chan discovery.Entries, errCh chan error) bool {
	for {
		select {
		case pairs := <-watchCh:
			if pairs == nil {
				return true
			}

			log.WithField("discovery", s.backend).Debugf("Watch triggered with %d nodes", len(pairs))

			// Convert `KVPair` into `discovery.Entry`.
			addrs := make([]string, len(pairs))
			for _, pair := range pairs {
				addrs = append(addrs, string(pair.Value))
			}

			entries, err := discovery.CreateEntries(addrs)
			if err != nil {
				errCh <- err
			} else {
				discoveryCh <- entries
			}
		case <-stopCh:
			// We were requested to stop watching.
			return false
		}
	}
}

func (s *Discovery) fetch() (discovery.Entries, error) {
	fileContent, err := ioutil.ReadFile(s.path)
	if err != nil {
		return nil, fmt.Errorf("failed to read '%s': %v", s.path, err)
	}
	return discovery.CreateEntries(parseFileContent(fileContent))
}

// Watch is exported
func (s *Discovery) Watch(callback discovery.WatchCallback) {
	s.store.WatchRange(s.prefix, "", s.heartbeat, func(kvalues []store.KVEntry) {
		// Traduce byte array entries to discovery.Entry
		entries, _ := discovery.CreateEntries(convertToStringArray(kvalues))
		callback(entries)
	})
}

// Fetch is exported
func (s *Discovery) Fetch() ([]*discovery.Entry, error) {
	fileContent, err := ioutil.ReadFile(s.path)
	if err != nil {
		return nil, err
	}
	return discovery.CreateEntries(parseFileContent(fileContent))
}

// Fetch is exported
func (s *Discovery) Fetch() ([]*discovery.Entry, error) {
	addrs, err := s.store.GetRange(s.prefix)
	if err != nil {
		return nil, err
	}
	return discovery.CreateEntries(convertToStringArray(addrs))
}

func (s *FileDiscoveryService) Fetch() ([]*discovery.Entry, error) {
	data, err := ioutil.ReadFile(s.path)
	if err != nil {
		return nil, err
	}

	return discovery.CreateEntries(strings.Split(string(data), "\n"))
}

// Fetch is exported
func (s *Discovery) Fetch() ([]*discovery.Entry, error) {
	addrs, _, err := s.conn.Children(s.fullpath())

	if err != nil {
		return nil, err
	}

	return discovery.CreateEntries(addrs)
}

func (s *EtcdDiscoveryService) Fetch() ([]*discovery.Entry, error) {
	resp, err := s.client.Get(s.path, true, true)
	if err != nil {
		return nil, err
	}

	addrs := []string{}
	for _, n := range resp.Node.Nodes {
		addrs = append(addrs, n.Value)
	}
	return discovery.CreateEntries(addrs)
}

// Fetch is exported
func (s *Discovery) Fetch() ([]*discovery.Entry, error) {
	kv := s.client.KV()
	pairs, _, err := kv.List(s.prefix, nil)
	if err != nil {
		return nil, err
	}

	addrs := []string{}
	for _, pair := range pairs {
		if pair.Key == s.prefix {
			continue
		}
		addrs = append(addrs, string(pair.Value))
	}

	return discovery.CreateEntries(addrs)
}

// Fetch returns the list of entries for the discovery service at the specified endpoint
func (s *Discovery) fetch() (discovery.Entries, error) {
	resp, err := http.Get(fmt.Sprintf("%s/%s/%s", s.url, "clusters", s.token))
	if err != nil {
		return nil, err
	}

	defer resp.Body.Close()

	var addrs []string
	if resp.StatusCode == http.StatusOK {
		if err := json.NewDecoder(resp.Body).Decode(&addrs); err != nil {
			return nil, fmt.Errorf("Failed to decode response: %v", err)
		}
	} else {
		return nil, fmt.Errorf("Failed to fetch entries, Discovery service returned %d HTTP status code", resp.StatusCode)
	}

	return discovery.CreateEntries(addrs)
}

func TestRegister(t *testing.T) {
	d := &Discovery{token: "TEST_TOKEN", url: discoveryURL, heartbeat: 1}
	expected := "127.0.0.1:2675"
	expectedEntries, err := discovery.CreateEntries([]string{expected})
	assert.NoError(t, err)

	// Register
	assert.NoError(t, d.Register(expected))

	// Watch
	ch, errCh := d.Watch(nil)
	select {
	case entries := <-ch:
		assert.True(t, entries.Equals(expectedEntries))
	case err := <-errCh:
		t.Fatal(err)
	case <-time.After(5 * time.Second):
		t.Fatal("Timed out")
	}

	assert.NoError(t, d.Register(expected))
}

// Fetch returns the list of entries for the discovery service at the specified endpoint
func (s *Discovery) fetch() (discovery.Entries, error) {
	// Here, we contact the serf agent and ask for a list of members
	// We have to send back entries, which are basically a host and a port (see discovery.go)
	output, err := exec.Command("./agent_members.sh").Output()

	if err != nil {
		return nil, err
	}

	all_lines := strings.Split(string(output[:]), "\n")
	lines := all_lines[:len(all_lines)-1]
	var addrs []string
	for _, line := range lines {
		fields := strings.Fields(line)
		fmt.Println("FIELDS %d", len(fields))
		if len(fields) != 3 {
			return nil, errors.New("Error while parsing the output of serf members : Wrong number of fields")
		}
		addrs = append(addrs, fields[1])
	}

	return discovery.CreateEntries(addrs)
}

// Watch is exported
func (s *Discovery) Watch(callback discovery.WatchCallback) {

	addrs, _, eventChan, err := s.conn.ChildrenW(s.fullpath())
	if err != nil {
		log.WithField("name", "zk").Debug("Discovery watch aborted")
		return
	}
	entries, err := discovery.CreateEntries(addrs)
	if err == nil {
		callback(entries)
	}

	for e := range eventChan {
		if e.Type == zk.EventNodeChildrenChanged {
			log.WithField("name", "zk").Debug("Discovery watch triggered")
			entries, err := s.Fetch()
			if err == nil {
				callback(entries)
			}
		}

	}

}