func (s *Service) watchChanges() {
	for {
		change := <-s.changes
		log.Infof("Service got change: %s", change)
		s.processChange(change)
	}
}

func (c *ProxyController) DeleteHost(w http.ResponseWriter, r *http.Request, params map[string]string) (interface{}, error) {
	log.Infof("Delete host: %s", params["hostname"])
	if err := c.backend.DeleteHost(params["hostname"]); err != nil {
		return nil, api.GenericAPIError{Reason: fmt.Sprintf("%s", err)}
	}
	return api.Response{"message": "Host deleted"}, nil
}

func (s *EtcdBackend) AddLocation(id, hostname, path, upstreamId string) error {
	log.Infof("Add Location(id=%s, hosntame=%s, path=%s, upstream=%s)", id, hostname, path, upstreamId)
	// Make sure location path is a valid regular expression
	if _, err := regexp.Compile(path); err != nil {
		return fmt.Errorf("Path should be a valid Golang regular expression")
	}

	// Make sure upstream actually exists
	_, err := s.readUpstream(upstreamId)
	if err != nil {
		return err
	}
	// Create the location
	if id == "" {
		response, err := s.client.AddChildDir(join(s.etcdKey, "hosts", hostname, "locations"), 0)
		if err != nil {
			return formatErr(err)
		}
		id = suffix(response.Node.Key)
	} else {
		_, err := s.client.CreateDir(join(s.etcdKey, "hosts", hostname, "locations", id), 0)
		if err != nil {
			return formatErr(err)
		}
	}
	locationKey := join(s.etcdKey, "hosts", hostname, "locations", id)
	if _, err := s.client.Create(join(locationKey, "path"), path, 0); err != nil {
		return formatErr(err)
	}
	if _, err := s.client.Create(join(locationKey, "upstream"), upstreamId, 0); err != nil {
		return formatErr(err)
	}
	return nil
}

func (c *Configurator) syncLocationEndpoints(location *Location) error {

	rr := c.a.GetHttpLocationLb(location.Hostname, location.Id)
	if rr == nil {
		return fmt.Errorf("%s lb not found", location)
	}

	// First, collect and parse endpoints to add
	newEndpoints := map[string]endpoint.Endpoint{}
	for _, e := range location.Upstream.Endpoints {
		ep, err := EndpointFromUrl(e.Url, e.Url)
		if err != nil {
			return fmt.Errorf("Failed to parse endpoint url: %s", e)
		}
		newEndpoints[e.Url] = ep
	}

	// Memorize what endpoints exist in load balancer at the moment
	existingEndpoints := map[string]endpoint.Endpoint{}
	for _, e := range rr.GetEndpoints() {
		existingEndpoints[e.GetUrl().String()] = e
	}

	// First, add endpoints, that should be added and are not in lb
	for _, e := range newEndpoints {
		if _, exists := existingEndpoints[e.GetUrl().String()]; !exists {
			if err := rr.AddEndpoint(e); err != nil {
				log.Errorf("Failed to add %s, err: %s", e, err)
			} else {
				log.Infof("Added %s to %s", e, location)
			}
		}
	}

	// Second, remove endpoints that should not be there any more
	for _, e := range existingEndpoints {
		if _, exists := newEndpoints[e.GetUrl().String()]; !exists {
			if err := rr.RemoveEndpoint(e); err != nil {
				log.Errorf("Failed to remove %s, err: %s", e, err)
			} else {
				log.Infof("Removed %s from %s", e, location)
			}
		}
	}
	return nil
}

func (s *Service) configureLocation(host *Host, location *Location) error {
	rr, err := s.getHttpLocationLb(host.Name, location.Name)
	if err != nil {
		return err
	}

	// First, collect and parse endpoints to add
	endpointsToAdd := map[string]endpoint.Endpoint{}
	for _, e := range location.Upstream.Endpoints {
		ep, err := EndpointFromUrl(e.Name, e.Url)
		if err != nil {
			return fmt.Errorf("Failed to parse endpoint url: %s", e)
		}
		endpointsToAdd[ep.GetId()] = ep
	}

	// Memorize what endpoints exist in load balancer at the moment
	existing := map[string]endpoint.Endpoint{}
	for _, e := range rr.GetEndpoints() {
		existing[e.GetId()] = e
	}

	// First, add endpoints, that should be added and are not in lb
	for eid, e := range endpointsToAdd {
		if _, exists := existing[eid]; !exists {
			if err := rr.AddEndpoint(e); err != nil {
				log.Errorf("Failed to add %s, err: %s", e, err)
			} else {
				log.Infof("Added %s", e)
			}
		}
	}

	// Second, remove endpoints that should not be there any more
	for eid, e := range existing {
		if _, exists := endpointsToAdd[eid]; !exists {
			if err := rr.RemoveEndpoint(e); err != nil {
				log.Errorf("Failed to remove %s, err: %s", e, err)
			} else {
				log.Infof("Removed %s", e)
			}
		}
	}
	return nil
}

func (c *Configurator) upsertHost(host *Host) error {
	if c.a.GetHostRouter().GetRouter(host.Name) != nil {
		return nil
	}
	router := pathroute.NewPathRouter()
	c.a.GetHostRouter().SetRouter(host.Name, router)
	log.Infof("Added %s", host)
	return nil
}

func (c *ProxyController) DeleteEndpoint(w http.ResponseWriter, r *http.Request, params map[string]string) (interface{}, error) {
	upstreamId := params["upstream"]
	id := params["endpoint"]

	log.Infof("Delete Endpoint(url=%s) from Upstream(id=%s)", id, upstreamId)
	if err := c.backend.DeleteEndpoint(upstreamId, id); err != nil {
		return nil, api.GenericAPIError{Reason: fmt.Sprintf("%s", err)}
	}
	return api.Response{"message": "Endpoint deleted"}, nil
}

func (s *Service) configureHost(host *Host) error {
	for _, loc := range host.Locations {
		if err := s.addLocation(host, loc); err != nil {
			log.Errorf("Failed adding %s to %s, err: %s", loc, host, err)
		} else {
			log.Infof("Added %s to %s", loc, host)
		}
	}
	return nil
}

func (c *ProxyController) AddUpstream(w http.ResponseWriter, r *http.Request, params map[string]string) (interface{}, error) {
	id, err := api.GetStringField(r, "id")
	if err != nil {
		return nil, err
	}
	log.Infof("Add Upstream: %s", id)
	if err := c.backend.AddUpstream(id); err != nil {
		return nil, api.GenericAPIError{Reason: fmt.Sprintf("%s", err)}
	}
	return api.Response{"message": "Upstream added"}, nil
}

func (c *ProxyController) AddHost(w http.ResponseWriter, r *http.Request, params map[string]string) (interface{}, error) {
	name, err := api.GetStringField(r, "name")
	if err != nil {
		return nil, err
	}
	log.Infof("Add host: %s", name)
	if err := c.backend.AddHost(name); err != nil {
		return nil, api.GenericAPIError{Reason: fmt.Sprintf("%s", err)}
	}

	return api.Response{"message": "Host added"}, nil
}

// Watches etcd changes and generates structured events telling vulcand to add or delete locations, hosts etc.
// if initialSetup is true, reads the existing configuration and generates events for inital configuration of the proxy.
func (s *EtcdBackend) WatchChanges(changes chan interface{}, initialSetup bool) error {
	if initialSetup == true {
		log.Infof("Etcd backend reading initial configuration")
		if err := s.generateChanges(changes); err != nil {
			log.Errorf("Failed to generate changes: %s, stopping watch.", err)
			return err
		}
	}
	// This index helps us to get changes in sequence, as they were performed by clients.
	waitIndex := uint64(0)
	for {
		response, err := s.client.Watch(s.etcdKey, waitIndex, true, nil, s.cancelC)
		if err != nil {
			if err == etcd.ErrWatchStoppedByUser {
				log.Infof("Stop watching: graceful shutdown")
				return nil
			}
			log.Errorf("Stop watching: Etcd client error: %v", err)
			return err
		}

		waitIndex = response.Node.ModifiedIndex + 1
		log.Infof("Got response: %s %s %d %v",
			response.Action, response.Node.Key, response.EtcdIndex, err)
		change, err := s.parseChange(response)
		if err != nil {
			log.Errorf("Failed to process change: %s, ignoring", err)
			continue
		}
		if change != nil {
			log.Infof("Sending change: %s", change)
			select {
			case changes <- change:
			case <-s.stopC:
				return nil
			}
		}
	}
	return nil
}

func (s *Service) deleteLocation(host *Host, loc *Location) error {
	router, err := s.getPathRouter(host.Name)
	if err != nil {
		return err
	}
	location := router.GetLocationById(loc.Name)
	if location == nil {
		return fmt.Errorf("%s not found", loc)
	}
	err = router.RemoveLocation(location)
	if err == nil {
		log.Infof("Removed %s", loc)
	}
	return err
}

func (c *ProxyController) UpdateLocation(w http.ResponseWriter, r *http.Request, params map[string]string) (interface{}, error) {
	hostname := params["hostname"]
	locationId := params["id"]

	upstream, err := api.GetStringField(r, "upstream")
	if err != nil {
		return nil, err
	}

	log.Infof("Update Location: %s %s set upstream", hostname, locationId, upstream)
	if err := c.backend.UpdateLocationUpstream(hostname, locationId, upstream); err != nil {
		return nil, api.GenericAPIError{Reason: fmt.Sprintf("%s", err)}
	}

	return api.Response{"message": "Location upstream updated"}, nil
}

func (c *ProxyController) AddEndpoint(w http.ResponseWriter, r *http.Request, params map[string]string) (interface{}, error) {
	url, err := api.GetStringField(r, "url")
	if err != nil {
		return nil, err
	}
	id, err := api.GetStringField(r, "id")
	if err != nil {
		return nil, err
	}

	upstreamId := params["upstream"]
	log.Infof("Add Endpoint %s to %s", url, upstreamId)

	if err := c.backend.AddEndpoint(upstreamId, id, url); err != nil {
		return nil, api.GenericAPIError{Reason: fmt.Sprintf("%s", err)}
	}
	return api.Response{"message": "Endpoint added"}, nil
}

func (s *Service) Start() error {
	// Init logging
	log.Init([]*log.LogConfig{&log.LogConfig{Name: s.options.Log}})

	backend, err := NewEtcdBackend(s.options.EtcdNodes, s.options.EtcdKey, s.options.EtcdConsistency)
	if err != nil {
		return err
	}
	s.backend = backend

	if s.options.PidPath != "" {
		if err := runtime.WritePid(s.options.PidPath); err != nil {
			return fmt.Errorf("Failed to write PID file: %v\n", err)
		}
	}

	if err := s.createProxy(); err != nil {
		return err
	}
	go s.startProxy()

	s.configurator = NewConfigurator(s.proxy)

	// Tell backend to watch configuration changes and pass them to the channel
	// the second parameter tells backend to do the initial read of the configuration
	// and produce the stream of changes so proxy would initialise initial config
	go s.backend.WatchChanges(s.changes, true)
	// Configurator will listen to the changes from the channel and will
	go s.configurator.WatchChanges(s.changes)

	if err := s.initApi(); err != nil {
		return err
	}

	go s.startApi()

	c := make(chan os.Signal, 1)
	signal.Notify(c, os.Interrupt, os.Kill)

	// Block until a signal is received.
	log.Infof("Got signal %s, exiting now", <-c)
	return nil
}

func (s *Service) configureProxy() error {
	hosts, err := s.backend.GetHosts()
	if err != nil {
		return err
	}
	for _, host := range hosts {
		log.Infof("Configuring %s", host)

		if err := s.addHost(host); err != nil {
			log.Errorf("Failed adding %s, err: %s", host, err)
			continue
		}
		if err := s.configureHost(host); err != nil {
			log.Errorf("Failed configuring %s", host)
			continue
		}
	}
	return nil
}

func (s *EtcdBackend) watchChanges() {
	waitIndex := uint64(0)
	for {
		response, err := s.client.Watch(s.etcdKey, waitIndex, true, nil, nil)
		if err != nil {
			log.Errorf("Failed to get response from etcd: %s, quitting watch goroutine", err)
			return
		}
		log.Infof("Got response: %s %s %d %s",
			response.Action, response.Node.Key, response.EtcdIndex, err)
		change, err := s.parseChange(response)
		if err != nil {
			log.Errorf("Failed to process change: %s, ignoring", err)
			continue
		}
		if change != nil {
			s.changes <- change
		}
		waitIndex = response.Node.ModifiedIndex + 1
	}
}

func (s *EtcdBackend) UpdateLocationUpstream(hostname, id, upstreamId string) error {
	log.Infof("Update Location(id=%s, hostname=%s) set upstream %s", id, hostname, upstreamId)

	// Make sure upstream actually exists
	_, err := s.readUpstream(upstreamId)
	if err != nil {
		return err
	}

	// Make sure location actually exists
	location, err := s.readLocation(hostname, id)
	if err != nil {
		return err
	}

	// Update upstream
	if _, err := s.client.Set(join(location.EtcdKey, "upstream"), upstreamId, 0); err != nil {
		return formatErr(err)
	}

	return nil
}

func (s *Service) deleteEndpoint(upstream *Upstream, e *Endpoint) error {
	endpoint, err := EndpointFromUrl(e.Name, "http://delete.me:4000")
	if err != nil {
		return fmt.Errorf("Failed to parse endpoint url: %s", endpoint)
	}
	locations, err := s.getLocations(upstream.Name)
	if err != nil {
		return err
	}
	for _, l := range locations {
		rr, ok := l.GetLoadBalancer().(*roundrobin.RoundRobin)
		if !ok {
			return fmt.Errorf("Unexpected load balancer type: %T", l.GetLoadBalancer())
		}
		if err := rr.RemoveEndpoint(endpoint); err != nil {
			log.Errorf("Failed to remove endpoint: %s", err)
		} else {
			log.Infof("Removed %s", e)
		}
	}
	return nil
}

func (s *Service) Start() error {
	// Init logging
	log.Init([]*log.LogConfig{&log.LogConfig{Name: "console"}})

	backend, err := NewEtcdBackend(s.options.EtcdNodes, s.options.EtcdKey, s.options.EtcdConsistency, s.changes, s)
	if err != nil {
		return err
	}
	s.backend = backend

	if s.options.PidPath != "" {
		if err := runtime.WritePid(s.options.PidPath); err != nil {
			return fmt.Errorf("Failed to write PID file: %v\n", err)
		}
	}

	if err := s.createProxy(); err != nil {
		return err
	}
	if err := s.configureProxy(); err != nil {
		return err
	}
	if err := s.configureApi(); err != nil {
		return err
	}

	go s.startProxy()
	go s.startApi()
	go s.watchChanges()

	c := make(chan os.Signal, 1)
	signal.Notify(c, os.Interrupt, os.Kill)

	// Block until a signal is received.
	log.Infof("Got signal %s, exiting now", <-c)
	return nil
}