Recursively run Process() in a processor

[RuneRoven]

I am writing a custom processor where i am using Ticker to have a timer running in the background.
When/if the timer elapse it should trigger the Process again. So far the timer works fine and it trigger when it should. It calls the a.Process() where a is the processor struct. The call for the process works fine without errors but the message does not proceed to the output stage. In the main process i save each new message to a.SaveMsg so i can resend it when the timer elaps. The saved message is correct when debugging. I have also tried to save the context to use the same but it does not work. I believe that the problem has to do with accessing the correct context when calling the process from the ticker using a.Context.. I cannot find a solution for this so i am asking if anyone have an idea on how to solve this.

[mihaitodor]

Hey @RuneRoven 👋 Without seeing your code it's a bit hard to figure out what you're trying to achieve. In general, processors aren't meant to emit new messages while blocking the current message. You can use a processor to expand a message into multiple messages, just like unarchive for example, but all those messages would have to be released immediately. What you're describing sounds a lot more like the generate input or maybe a custom buffer. Have a look a the system_window implementation for example: https://github.com/redpanda-data/benthos/blob/791efc94787ca6d0986923afbe1b3d07481a7dd7/internal/impl/pure/buffer_system_window.go.

[RuneRoven]

I have a custom processor which listens to messages and extract info based on the settings. In the processor i have a timer running in the background which is triggered based on conditions in the input. Each message coming to the processor is saved. If the timer is started and elapsed it should "resend" the message so it is sent with the triggered value to the Process for further processing and sent to the output.. I need to have this to retrigger the entire process after the timer has elapsed because i don't want to have multiple processors or inputs..

[mihaitodor]

I could be wrong, but that sounds like something which may fit better into a custom buffer like I suggested above. Or, dunno, maybe a custom input which also has this timer logic?

[RuneRoven]

Let me try to explain a bit further. This is custom alarming processor that can handle different inputs and outputs and the idea is to make all the heavy logic in the processor instead of using for example lots of conditions in bloblang.
When the alarm is triggered there is a option to delay the output in case the value is for example flickering.
When data is not coming constantly this can cause a problem. Say for example data is usually coming about every 5 min so you set the delay to 6min, this would trigger the alarm if the first data is above the threshold and the second data also is above. However if the next data is not coming after 5min but instead after 15min the alarm will not be triggered after 5min but instead 15min.
I need to handle this exceptions in the processor and trigger the alarm after and exact time.
What i have tried is to make a copy of the last sent message and if the timer elapses i am trying to resend that message internally so the Process() can reapply the logic and send the message further to the output.

In my Process() i have

func (a *alarms) Process(ctx context.Context, msg *service.Message) (service.MessageBatch, error) {
	a.savedMsg = msg
        ....
}

i have a ticker

func (a *alarms) startTicker() {
	ticker := time.NewTicker(a.filterTime)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			a.checkAndTriggerAlarm()
		case <-a.stopTickerChan:
			return
		}
	}
}

and a check and trigger

func (a *alarms) checkAndTriggerAlarm() {
	a.mu.Lock()
	defer a.mu.Unlock()

	if !a.startTime.IsZero() {
		if time.Since(a.startTime) >= a.filterTime && !a.trigger {
			a.trigger = true
			a.send = true
			a.Process(a.mainCtx, a.savedMsg)

		}
	}
}