func (f *FirehoseBase) Start(w WatchJobListFunc) {
	go f.sink.Start()
	go f.watch(w)
	go f.persistLastChangeTime(5 * time.Second)
	select {
	case <-f.stopCh:
		return
	}
}
Start the firehose
package jobs
import (
	"fmt"
	"time"
	nomad "github.com/hashicorp/nomad/api"
	"github.com/seatgeek/nomad-firehose/sink"
	log "github.com/sirupsen/logrus"
)
type WatchJobListFunc func(job *nomad.JobListStub)
type FirehoseBase struct {
	lastChangeIndex  uint64
	lastChangeTimeCh chan interface{}
	nomadClient      *nomad.Client
	sink             sink.Sink
	stopCh           chan struct{}
}
func NewFirehoseBase() (*FirehoseBase, error) {
	nomadClient, err := nomad.NewClient(nomad.DefaultConfig())
	if err != nil {
		return nil, err
	}
	sink, err := sink.GetSink("jobs")
	if err != nil {
		return nil, err
	}
	return &FirehoseBase{
		nomadClient:      nomadClient,
		sink:             sink,
		stopCh:           make(chan struct{}, 1),
		lastChangeTimeCh: make(chan interface{}, 1),
	}, nil
}
func (f *FirehoseBase) UpdateCh() <-chan interface{} {
	return f.lastChangeTimeCh
}
func (f *FirehoseBase) 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
}