// Subscribe watches etcd changes and generates structured events telling vulcand to add or delete frontends, hosts etc.
// It is a blocking function.
func (n *ng) Subscribe(changes chan interface{}, cancelC chan bool) error {
	// This index helps us to get changes in sequence, as they were performed by clients.
	waitIndex := uint64(0)
	for {
		response, err := n.client.Watch(n.etcdKey, waitIndex, true, nil, cancelC)
		if err != nil {
			switch err {
			case etcd.ErrWatchStoppedByUser:
				log.Infof("Stop watching: graceful shutdown")
				return nil
			default:
				log.Errorf("unexpected error: %s, stop watching", err)
				return err
			}
		}
		waitIndex = response.EtcdIndex + 1
		log.Infof("%s", responseToString(response))
		change, err := n.parseChange(response)
		if err != nil {
			log.Warningf("Ignore '%s', error: %s", responseToString(response), err)
			continue
		}
		if change != nil {
			log.Infof("%v", change)
			select {
			case changes <- change:
			case <-cancelC:
				return nil
			}
		}
	}
}

func (m *mux) TakeFiles(files []*FileDescriptor) error {
	log.Infof("%s TakeFiles %s", m, files)

	fMap := make(map[engine.Address]*FileDescriptor, len(files))
	for _, f := range files {
		fMap[f.Address] = f
	}

	m.mtx.Lock()
	defer m.mtx.Unlock()

	for _, srv := range m.servers {

		file, exists := fMap[srv.listener.Address]
		if !exists {
			log.Infof("%s skipping take of files from address %s, has no passed files", m, srv.listener.Address)
			continue
		}
		if err := srv.takeFile(file); err != nil {
			return err
		}
	}

	return nil
}

// supervise() listens for error notifications and triggers graceful restart
func (s *Supervisor) supervise() {
	for {
		select {
		case err := <-s.restartC:
			// This means graceful shutdown, do nothing and return
			if err == nil {
				log.Infof("watchErrors - graceful shutdown")
				s.stop()
				return
			}
			for {
				s.options.Clock.Sleep(retryPeriod)
				log.Infof("supervise() restarting %s on error: %s", s.proxy, err)
				// We failed to initialize server, this error can not be recovered, so send an error and exit
				if err := s.init(); err != nil {
					log.Infof("Failed to initialize %s, will retry", err)
				} else {
					break
				}
			}

		case <-s.broadcastCloseC:
			s.Stop(false)
		}
	}
}

// Start the app on the configured host/port.
//
// Supports graceful shutdown on 'kill' and 'int' signals.
func (app *App) Run() error {
	// toggle heartbeat on SIGUSR1
	go func() {
		app.heartbeater.Start()
		heartbeatChan := make(chan os.Signal, 1)
		signal.Notify(heartbeatChan, syscall.SIGUSR1)

		for s := range heartbeatChan {
			log.Infof("Received signal: %v, toggling heartbeat", s)
			app.heartbeater.Toggle()
		}
	}()

	// listen for a shutdown signal
	go func() {
		exitChan := make(chan os.Signal, 1)
		signal.Notify(exitChan, syscall.SIGINT, syscall.SIGQUIT, syscall.SIGTERM)
		s := <-exitChan
		log.Infof("Got shutdown signal: %v", s)
		manners.Close()
	}()

	addr := fmt.Sprintf("%v:%v", app.Config.ListenIP, app.Config.ListenPort)
	return manners.ListenAndServe(addr, app.router)
}

func (hs *hostStapler) update() {
	re, err := hs.s.getStaple(&hs.host.Settings)
	log.Infof("%v got %v %v", hs, re, err)
	select {
	case hs.s.eventsC <- &stapleFetched{id: hs.id, hostName: hs.host.Name, re: re, err: err}:
	case <-hs.stopC:
		log.Infof("%v stopped", hs)
	}
}

func (s *srv) serve(srv *manners.GracefulServer) {
	log.Infof("%s serve", s)

	s.mux.wg.Add(1)
	defer s.mux.wg.Done()

	srv.ListenAndServe()

	log.Infof("%v stop", s)
}

func (s *Service) startChild() error {
	log.Infof("Starting child")
	path, err := execPath()
	if err != nil {
		return err
	}

	wd, err := os.Getwd()
	if nil != err {
		return err
	}

	// Get socket files currently in use by the underlying http server controlled by supervisor
	extraFiles, err := s.supervisor.GetFiles()
	if err != nil {
		return err
	}

	apiFile, err := s.GetAPIFile()
	if err != nil {
		return err
	}

	extraFiles = append(extraFiles, apiFile)

	// These files will be passed to the child process
	files := []*os.File{os.Stdin, os.Stdout, os.Stderr}
	for _, f := range extraFiles {
		files = append(files, f.File)
	}

	// Serialize files to JSON string representation
	vals, err := filesToString(extraFiles)
	if err != nil {
		return err
	}

	log.Infof("Passing %s to child", vals)
	os.Setenv(vulcandFilesKey, vals)

	p, err := os.StartProcess(path, os.Args, &os.ProcAttr{
		Dir:   wd,
		Env:   os.Environ(),
		Files: files,
		Sys:   &syscall.SysProcAttr{},
	})

	if err != nil {
		return err
	}

	log.Infof("Started new child pid=%d binary=%s", p.Pid, path)
	return nil
}

func (m *mux) Stop(wait bool) {
	log.Infof("%s Stop(%t)", m, wait)

	m.stopServers()

	if wait {
		log.Infof("%s waiting for the wait group to finish", m)
		m.wg.Wait()
		log.Infof("%s wait group finished", m)
	}
}

func (hs *hostStapler) sameTo(host *engine.Host) bool {
	if !hs.host.Settings.KeyPair.Equals(host.Settings.KeyPair) {
		log.Infof("%v key pair updated", hs)
		return false
	}
	if !(&hs.host.Settings.OCSP).Equals(&host.Settings.OCSP) {
		log.Infof("%v ocsp settings updated", hs)
		return false
	}
	return true
}

func (s *Supervisor) stop() {
	srv := s.getCurrentProxy()
	if srv != nil {
		srv.Stop(true)
		log.Infof("%s was stopped by supervisor", srv)
	}
	log.Infof("Wait for any outstanding operations to complete")
	s.wg.Wait()
	log.Infof("All outstanding operations have been completed, signalling stop")
	close(s.closeC)
}

func (r *ratioController) allowRequest() bool {
	log.Infof("%v", r)
	t := r.targetRatio()
	// This condition answers the question - would we satisfy the target ratio if we allow this request?
	e := r.computeRatio(r.allowed+1, r.denied)
	if e < t {
		r.allowed++
		log.Infof("%v allowed", r)
		return true
	}
	r.denied++
	log.Infof("%v denied", r)
	return false
}

func (s *srv) takeFile(f *FileDescriptor) error {
	log.Infof("%s takeFile %v", s, f)

	listener, err := f.ToListener()
	if err != nil {
		return err
	}

	if s.isTLS() {
		tcpListener, ok := listener.(*net.TCPListener)
		if !ok {
			return fmt.Errorf(`%s failed to take file descriptor - it is running in TLS mode so I need a TCP listener, 
but the file descriptor that was given corresponded to a listener of type %T. More about file descriptor: %s`, listener, s, f)
		}
		config, err := s.newTLSConfig()
		if err != nil {
			return err
		}
		listener = manners.NewTLSListener(
			manners.TCPKeepAliveListener{tcpListener}, config)
	}

	s.srv = manners.NewWithOptions(
		manners.Options{
			Server:       s.newHTTPServer(),
			Listener:     listener,
			StateHandler: s.mux.connTracker.onStateChange,
		})
	s.state = srvStateHijacked
	return nil
}

func (m *mux) DeleteHost(hk engine.HostKey) error {
	log.Infof("%s DeleteHost %v", m, &hk)

	m.mtx.Lock()
	defer m.mtx.Unlock()

	host, exists := m.hosts[hk]
	if !exists {
		return &engine.NotFoundError{Message: fmt.Sprintf("%v not found", hk)}
	}

	// delete host from the hosts list
	delete(m.hosts, hk)

	// delete staple from the cache
	m.stapler.DeleteHost(hk)

	if host.Settings.KeyPair == nil {
		return nil
	}

	for _, s := range m.servers {
		s.reload()
	}
	return nil
}

func (s *srv) start() error {
	log.Infof("%s start", s)
	switch s.state {
	case srvStateInit:
		listener, err := net.Listen(s.listener.Address.Network, s.listener.Address.Address)
		if err != nil {
			return err
		}

		if s.isTLS() {
			config, err := s.newTLSConfig()
			if err != nil {
				return err
			}
			listener = manners.NewTLSListener(
				manners.TCPKeepAliveListener{listener.(*net.TCPListener)}, config)
		}
		s.srv = manners.NewWithOptions(
			manners.Options{
				Server:       s.newHTTPServer(),
				Listener:     listener,
				StateHandler: s.mux.connTracker.onStateChange,
			})
		s.state = srvStateActive
		go s.serve(s.srv)
		return nil
	case srvStateHijacked:
		s.state = srvStateActive
		go s.serve(s.srv)
		return nil
	}
	return fmt.Errorf("%v Calling start in unsupported state", s)
}

func (m *mux) upsertListener(l engine.Listener) error {
	lk := engine.ListenerKey{Id: l.Id}
	s, exists := m.servers[lk]
	if exists {
		return s.updateListener(l)
	}

	// Check if there's a listener with the same address
	for _, srv := range m.servers {
		if srv.listener.Address == l.Address {
			return &engine.AlreadyExistsError{Message: fmt.Sprintf("%v conflicts with existing %v", l, srv.listener)}
		}
	}

	var err error
	if s, err = newSrv(m, l); err != nil {
		return err
	}
	m.servers[lk] = s
	// If we are active, start the server immediatelly
	if m.state == stateActive {
		log.Infof("Mux is in active state, starting the HTTP server")
		if err := s.start(); err != nil {
			return err
		}
	}
	return nil
}

func (m *mux) UpsertListener(l engine.Listener) error {
	log.Infof("%v UpsertListener %v", m, &l)
	m.mtx.Lock()
	defer m.mtx.Unlock()

	return m.upsertListener(l)
}

func (c *ProxyController) deleteFrontend(w http.ResponseWriter, r *http.Request, params map[string]string) (interface{}, error) {
	log.Infof("Delete Frontend(id=%s)", params["id"])
	if err := c.ng.DeleteFrontend(engine.FrontendKey{Id: params["id"]}); err != nil {
		return nil, formatError(err)
	}
	return scroll.Response{"message": "Frontend deleted"}, nil
}

// Make a handler out of HandlerWithBodyFunc, just like regular MakeHandler function.
func MakeHandlerWithBody(app *App, fn HandlerWithBodyFunc, spec Spec) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		if err := parseForm(r); err != nil {
			ReplyInternalError(w, fmt.Sprintf("Failed to parse request form: %v", err))
			return
		}

		body, err := ioutil.ReadAll(r.Body)
		if err != nil {
			ReplyInternalError(w, fmt.Sprintf("Failed to read request body: %v", err))
			return
		}

		start := time.Now()
		response, err := fn(w, r, mux.Vars(r), body)
		elapsedTime := time.Since(start)

		var status int
		if err != nil {
			response, status = responseAndStatusFor(err)
		} else {
			status = http.StatusOK
		}

		log.Infof("Request(Status=%v, Method=%v, Path=%v, Form=%v, Time=%v, Error=%v)",
			status, r.Method, r.URL, r.Form, elapsedTime, err)

		app.stats.TrackRequest(spec.MetricName, status, elapsedTime)

		Reply(w, response, status)
	}
}

// RegisterApp adds a new backend and a single server with Vulcand.
func (s *LeaderRegistry) RegisterApp(registration *AppRegistration) error {
	log.Infof("Registering app: %v", registration)

	endpoint, err := vulcan.NewEndpointWithID(s.Group, registration.Name, registration.Host, registration.Port)
	if err != nil {
		return err
	}

	err = s.client.RegisterBackend(endpoint)
	if err != nil {
		log.Errorf("Failed to register backend for endpoint: %v, %s", endpoint, err)
		return err
	}

	if s.IsMaster {
		err = s.maintainLeader(endpoint)
	} else {
		err = s.initLeader(endpoint)
	}

	if err != nil {
		log.Errorf("Failed to register server for endpoint: %v, %s", endpoint, err)
		return err
	}

	return nil
}

func (w *WebhookSideEffect) Exec() error {
	r, err := http.NewRequest(w.w.Method, w.w.URL, w.getBody())
	if err != nil {
		return err
	}
	if len(w.w.Headers) != 0 {
		utils.CopyHeaders(r.Header, w.w.Headers)
	}
	if len(w.w.Form) != 0 {
		r.Header.Set("Content-Type", "application/x-www-form-urlencoded")
	}
	re, err := http.DefaultClient.Do(r)
	if err != nil {
		return err
	}
	if re.Body != nil {
		defer re.Body.Close()
	}
	body, err := ioutil.ReadAll(re.Body)
	if err != nil {
		return err
	}
	log.Infof("%v got response: (%s): %s", w, re.Status, string(body))
	return nil
}