func (m *Manager) releaseConsulLock() {
	m.logger.Info("Releasing Consul lock")
	close(m.voluntarilyReleaseLockCh)
}
releaseConsulLock stops consul lock handler")
package helper
import (
	"fmt"
	"os"
	"os/signal"
	"strconv"
	"strings"
	"time"
	consulapi "github.com/hashicorp/consul/api"
	log "github.com/sirupsen/logrus"
)
type Runner interface {
	Name() string
	SetRestoreValue(restoreTime interface{}) error
	Start()
	Stop()
	UpdateCh() <-chan interface{}
}
func NewManager(r Runner) *Manager {
	consulPrefix := "nomad-firehose"
	if v, ok := os.LookupEnv("NOMAD_FIREHOSE_CONSUL_PREFIX"); ok {
		consulPrefix = strings.TrimSuffix(v, "/")
	}
	return &Manager{
		runner:                   r,
		logger:                   log.WithField("type", r.Name()),
		stopCh:                   make(chan interface{}),
		voluntarilyReleaseLockCh: make(chan interface{}),
		prefix:                   consulPrefix,
	}
}
type Manager struct {
	runner                   Runner
	client                   *consulapi.Client
	lock                     *consulapi.Lock
	lockCh                   <-chan struct{}   
	lockErrorCh              <-chan struct{}   
	logger                   *log.Entry        
	prefix                   string            
	stopCh                   chan interface{}  
	voluntarilyReleaseLockCh chan interface{}
}
func (m *Manager) cleanup() {
	m.logger.Debug("Releasing lock")
	m.releaseConsulLock()
	m.logger.Debug("Closing stopCh")
	close(m.stopCh)
	m.logger.Debugf("Cleanup complete")
}
func (m *Manager) continuouslyAcquireConsulLeadership() error {
	m.logger.Info("Starting to continously acquire leadership")
	interval := 250 * time.Millisecond
	timer := time.NewTimer(interval)
	for {
		select {
		case <-m.stopCh:
			return nil
		case <-timer.C:
			if err := m.acquireConsulLeadership(); err != nil {
				return err
			}
			timer.Reset(interval)
		}
	}
}
func (m *Manager) restoreLastChangeTime() interface{} {
	kv, _, err := m.client.KV().Get(fmt.Sprintf("%s/%s.value", m.prefix, m.runner.Name()), nil)
	if err != nil {
		return 0
	}
	if kv != nil && kv.Value != nil {
		sv := string(kv.Value)
		v, err := strconv.ParseInt(sv, 10, 64)
		if err != nil {
			return 0
		}
		log.Infof("Restoring Last Change Time to %s", sv)
		return v
	}
	log.Info("No Last Change Time restore point, starting from scratch")
	return 0
}
func (m *Manager) acquireConsulLeadership() error {
	var err error
	m.lock, err = m.client.LockOpts(&consulapi.LockOptions{
		Key:              fmt.Sprintf("%s/%s.lock", m.prefix, m.runner.Name()),
		SessionName:      fmt.Sprintf("nomad-firehose-%s", m.runner.Name()),
		MonitorRetries:   10,
		MonitorRetryTime: 5 * time.Second,
	})
	if err != nil {
		return fmt.Errorf("Failed create lock options: %+v", err)
	}
	m.logger.Infof("Trying to acquire consul lock")
	m.lockErrorCh, err = m.lock.Lock(m.lockCh)
	if err != nil {
		return err
	}
	m.logger.Info("Lock successfully acquired")
	m.voluntarilyReleaseLockCh = make(chan interface{})
	if err := m.runner.SetRestoreValue(m.restoreLastChangeTime()); err != nil {
		return err
	}
	go m.runner.Start()
	defer func() {
		m.runner.Stop()
		err := m.lock.Unlock()
		m.handleConsulError(err)
		if err != nil {
			m.logger.Errorf("Could not release Consul Lock: %v", err)
		} else {
			m.logger.Info("Consul Lock successfully released")
		}
	}()
	for {
		select {
		case v := <-m.runner.UpdateCh():
			var r string
			switch v.(type) {
			case int:
				r = strconv.Itoa(v.(int))
			case int64, uint64:
				r = fmt.Sprintf("%d", v)
			default:
				return fmt.Errorf("Unknown update type '%T' with value '%+v'", v, v)
			}
			m.logger.Debugf("Writing lastChangedTime to KV: %s", r)
			kv := &consulapi.KVPair{
				Key:   fmt.Sprintf("%s/%s.value", m.prefix, m.runner.Name()),
				Value: []byte(r),
			}
			_, err := m.client.KV().Put(kv, nil)
			if err != nil {
				log.Error(err)
			}
		case <-m.stopCh:
			return nil
		case data, ok := <-m.lockErrorCh:
			if !ok {
				return fmt.Errorf("Consul Lock error channel was closed, we no longer hold the lock")
			}
			m.logger.Warnf("Something wrote to lock error channel %+v", data)
		case <-m.voluntarilyReleaseLockCh:
			m.logger.Warnf("Voluntarily releasing the Consul lock")
			return nil
		}
	}
}