func (f *Firehose) Stop() {
	close(f.stopCh)
	f.sink.Stop()
}
Stop the firehose
package nodes
import (
	"encoding/json"
	"fmt"
	"time"
	nomad "github.com/hashicorp/nomad/api"
	"github.com/seatgeek/nomad-firehose/sink"
	log "github.com/sirupsen/logrus"
)
type Firehose struct {
	lastChangeIndex   uint64
	lastChangeIndexCh chan interface{}
	nomadClient       *nomad.Client
	sink              sink.Sink
	stopCh            chan struct{}
}
func NewFirehose() (*Firehose, error) {
	nomadClient, err := nomad.NewClient(nomad.DefaultConfig())
	if err != nil {
		return nil, err
	}
	sink, err := sink.GetSink("nodes")
	if err != nil {
		return nil, err
	}
	return &Firehose{
		nomadClient:       nomadClient,
		sink:              sink,
		stopCh:            make(chan struct{}, 1),
		lastChangeIndexCh: make(chan interface{}, 1),
	}, nil
}
func (f *Firehose) Name() string {
	return "nodes"
}
func (f *Firehose) UpdateCh() <-chan interface{} {
	return f.lastChangeIndexCh
}
func (f *Firehose) SetRestoreValue(restoreValue interface{}) error {
	switch restoreValue.(type) {
	case int:
		f.lastChangeIndex = uint64(restoreValue.(int))
	case int64:
		f.lastChangeIndex = uint64(restoreValue.(int64))
	default:
		return fmt.Errorf("Unknown restore type '%T' with value '%+v'", restoreValue, restoreValue)
	}
	return nil
}
func (f *Firehose) Start() {
	go f.sink.Start()
	f.stopCh = make(chan struct{})
	go f.watch()
	go f.persistLastChangeTime(5 * time.Second)
	select {
	case <-f.stopCh:
		return
	}
}