func (cons *SimpleConsumer) GetID() string {
	return cons.id
}
GetID returns the ID of this consumer
package core
import (
	"fmt"
	"sync"
	"time"
	"github.com/sirupsen/logrus"
	"github.com/trivago/tgo"
	"github.com/trivago/tgo/tcontainer"
	"github.com/trivago/tgo/thealthcheck"
)
type SimpleConsumer struct {
	id              string
	control         chan PluginControl
	runState        *PluginRunState
	routers         []Router       `config:"Streams"`
	modulators      ModulatorArray `config:"Modulators"`
	onRoll          func()
	onPrepareStop   func()
	onStop          func()
	enqueueMessage  func(*Message)
	modulatorQueue  MessageQueue
	Logger          logrus.FieldLogger
	shutdownTimeout time.Duration `config:"ShutdownTimeoutMs" default:"1000" metric:"ms"`
}
func (cons *SimpleConsumer) Configure(conf PluginConfigReader) {
	cons.id = conf.GetID()
	cons.Logger = conf.GetLogger()
	cons.runState = NewPluginRunState()
	cons.control = make(chan PluginControl, 1)
	numRoutines := conf.GetInt("ModulatorRoutines", 0)
	queueSize := conf.GetInt("ModulatorQueueSize", 1024)
	if numRoutines > 0 {
		cons.Logger.Debugf("Using %d modulator routines", numRoutines)
		cons.modulatorQueue = NewMessageQueue(int(queueSize))
		for i := 0; i < int(numRoutines); i++ {
			go cons.processQueue()
		}
		cons.enqueueMessage = cons.parallelEnqueue
	} else {
		cons.enqueueMessage = cons.directEnqueue
	}
	cons.AddHealthCheckAt("/pluginState", func() (code int, body string) {
		if cons.IsActive() {
			return thealthcheck.StatusOK, fmt.Sprintf("ACTIVE: %s", cons.runState.GetStateString())
		}
		return thealthcheck.StatusServiceUnavailable,
			fmt.Sprintf("NOT_ACTIVE: %s", cons.runState.GetStateString())
	})
}
func (cons *SimpleConsumer) GetLogger() logrus.FieldLogger {
	return cons.Logger
}
func (cons *SimpleConsumer) AddHealthCheckAt(path string, callback thealthcheck.CallbackFunc) {
	thealthcheck.AddEndpoint("/"+cons.GetID()+path, callback)
}