func (prod *SimpleProducer) HasContinueAfterModulate(msg *Message) bool {
	switch result := prod.Modulate(msg); result {
	case ModulateResultDiscard:
		DiscardMessage(msg, prod.GetID(), "Producer discarded")
		return false
	case ModulateResultFallback:
		if err := Route(msg, msg.GetRouter()); err != nil {
			prod.Logger.WithError(err).Error("Failed to route to fallback")
		}
		return false
	case ModulateResultContinue:
		return true
	default:
		prod.Logger.Error("Modulator result not supported:", result)
		return false
	}
}
HasContinueAfterModulate applies all modulators by Modulate, handle the ModulateResult
and return if you have to continue the message process.
This method is a default producer modulate handling.
package core
import (
	"fmt"
	"sync"
	"time"
	"github.com/sirupsen/logrus"
	"github.com/trivago/tgo"
	"github.com/trivago/tgo/thealthcheck"
)
type SimpleProducer struct {
	id              string
	control         chan PluginControl
	runState        *PluginRunState
	streams         []MessageStreamID `config:"Streams"`
	modulators      ModulatorArray    `config:"Modulators"`
	fallbackStream  Router            `config:"FallbackStream" default:""`
	shutdownTimeout time.Duration     `config:"ShutdownTimeoutMs" default:"1000" metric:"ms"`
	onRoll          func()
	onPrepareStop   func()
	onStop          func()
	Logger          logrus.FieldLogger
}
func (prod *SimpleProducer) Configure(conf PluginConfigReader) {
	prod.id = conf.GetID()
	prod.Logger = conf.GetLogger()
	prod.runState = NewPluginRunState()
	prod.control = make(chan PluginControl, 1)
	prod.AddHealthCheckAt("/pluginState", func() (code int, body string) {
		if prod.IsActive() {
			return thealthcheck.StatusOK,
				fmt.Sprintf("ACTIVE: %s", prod.runState.GetStateString())
		}
		return thealthcheck.StatusServiceUnavailable,
			fmt.Sprintf("NOT_ACTIVE: %s", prod.runState.GetStateString())
	})
}
func (prod *SimpleProducer) GetLogger() logrus.FieldLogger {
	return prod.Logger
}
func (prod *SimpleProducer) AddHealthCheck(callback thealthcheck.CallbackFunc) {
	prod.AddHealthCheckAt("", callback)
}
func (prod *SimpleProducer) AddHealthCheckAt(path string, callback thealthcheck.CallbackFunc) {
	thealthcheck.AddEndpoint("/"+prod.GetID()+path, callback)
}
func (prod *SimpleProducer) GetID() string {
	return prod.id
}
func (prod *SimpleProducer) Streams() []MessageStreamID {
	return prod.streams
}
func (prod *SimpleProducer) Control() chan<- PluginControl {
	return prod.control
}
func (prod *SimpleProducer) GetState() PluginState {
	return prod.runState.GetState()
}
func (prod *SimpleProducer) IsBlocked() bool {
	return prod.GetState() == PluginStateWaiting
}
func (prod *SimpleProducer) IsActive() bool {
	return prod.GetState() <= PluginStatePrepareStop
}
func (prod *SimpleProducer) IsStopping() bool {
	return prod.GetState() >= PluginStatePrepareStop
}
func (prod *SimpleProducer) IsActiveOrStopping() bool {
	return prod.IsActive() || prod.IsStopping()
}
func (prod *SimpleProducer) SetRollCallback(onRoll func()) {
	prod.onRoll = onRoll
}
func (prod *SimpleProducer) SetPrepareStopCallback(onPrepareStop func()) {
	prod.onPrepareStop = onPrepareStop
}
func (prod *SimpleProducer) SetStopCallback(onStop func()) {
	prod.onStop = onStop
}
func (prod *SimpleProducer) SetWorkerWaitGroup(workers *sync.WaitGroup) {
	prod.runState.SetWorkerWaitGroup(workers)
}
func (prod *SimpleProducer) AddMainWorker(workers *sync.WaitGroup) {
	prod.runState.SetWorkerWaitGroup(workers)
	prod.AddWorker()
}
func (prod *SimpleProducer) AddWorker() {
	prod.runState.AddWorker()
}
func (prod *SimpleProducer) WorkerDone() {
	prod.runState.WorkerDone()
}
func (prod *SimpleProducer) GetShutdownTimeout() time.Duration {
	return prod.shutdownTimeout
}
func (prod *SimpleProducer) Modulate(msg *Message) ModulateResult {
	if len(prod.modulators) > 0 {
		msg.FreezeOriginal()
		return prod.modulators.Modulate(msg)
	}
	return ModulateResultContinue
}