func (prod *BufferedProducer) TickerMessageControlLoop(onMessage func(*Message), interval time.Duration, onTimeOut func()) {
	prod.setState(PluginStateActive)
	go prod.ControlLoop()
	go prod.tickerLoop(interval, onTimeOut)
	prod.messageLoop(onMessage)
}
TickerMessageControlLoop is like MessageLoop but executes a given function at
every given interval tick, too. If the onTick function takes longer than
interval, the next tick will be delayed until onTick finishes.
package core
import (
	"time"
	"github.com/trivago/tgo"
)
type BufferedProducer struct {
	DirectProducer `gollumdoc:"embed_type"`
	messages       MessageQueue
	channelTimeout time.Duration `config:"ChannelTimeoutMs" default:"0" metric:"ms"`
}
func (prod *BufferedProducer) Configure(conf PluginConfigReader) {
	prod.onPrepareStop = prod.DefaultDrain
	prod.onStop = prod.DefaultClose
	prod.messages = NewMessageQueue(int(conf.GetInt("Channel", 8192)))
}
func (prod *BufferedProducer) GetQueueTimeout() time.Duration {
	return prod.channelTimeout
}
func (prod *BufferedProducer) Enqueue(msg *Message, timeout time.Duration) {
	defer prod.enqueuePanicHandling(msg)
	if prod.GetState() >= PluginStateStopping {
		prod.TryFallback(msg)
		return  
	}
	if !prod.HasContinueAfterModulate(msg) {
		return
	}
	usedTimeout := prod.channelTimeout
	if timeout != 0 {
		usedTimeout = timeout
	}
	switch prod.messages.Push(msg, usedTimeout) {
	case MessageQueueTimeout:
		prod.TryFallback(msg)
		prod.setState(PluginStateWaiting)
	case MessageQueueDiscard:
		MetricMessagesDiscarded.Inc(1)
		prod.setState(PluginStateWaiting)
	default:
		prod.setState(PluginStateActive)
	}
	MessageTrace(msg, prod.GetID(), "Enqueued by buffered producer")
}
func (prod *BufferedProducer) DefaultDrain() {
	if prod.onMessage != nil {
		prod.DrainMessageChannel(prod.onMessage, prod.shutdownTimeout)
	}
}
func (prod *BufferedProducer) DrainMessageChannel(handleMessage func(*Message), timeout time.Duration) bool {
	for {
		if msg, ok := prod.messages.PopWithTimeout(timeout); ok {
			if !tgo.ReturnAfter(prod.shutdownTimeout, func() { handleMessage(msg) }) {
				return false  
			}
		} else {
			return prod.messages.IsEmpty()  
		}
	}
}
func (prod *BufferedProducer) DefaultClose() {
	if prod.onMessage != nil {
		prod.CloseMessageChannel(prod.onMessage)
	}
}
func (prod *BufferedProducer) CloseMessageChannel(handleMessage func(*Message)) (empty bool) {
	prod.DrainMessageChannel(handleMessage, prod.shutdownTimeout)
	prod.messages.Close()
	defer func() {
		if !prod.messages.IsEmpty() {
			prod.Logger.Errorf("%d messages left after closing.", prod.messages.GetNumQueued())
		}
	}()
	for {
		if msg, ok := prod.messages.Pop(); ok {
			if !tgo.ReturnAfter(prod.shutdownTimeout, func() { handleMessage(msg) }) {
				return false  
			}
		} else {
			return true  
		}
	}
}
func (prod *BufferedProducer) MessageControlLoop(onMessage func(*Message)) {
	prod.setState(PluginStateActive)
	go prod.ControlLoop()
	prod.messageLoop(onMessage)
}