This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
The following commit(s) were added to refs/heads/develop by this push:
new 2a15f850be feat(plc4go): add StopWarn callbacks to handle lifecycle
2a15f850be is described below
commit 2a15f850be5a01f5486de602f55b615b63e51ea5
Author: Sebastian Rühl <[email protected]>
AuthorDate: Thu May 28 09:04:57 2026 +0200
feat(plc4go): add StopWarn callbacks to handle lifecycle
---
plc4go/spi/utils/StopWarn.go | 58 +++++++++++++++++++++++++++++++++++--
plc4go/spi/utils/StopWarn_test.go | 60 +++++++++++++++++++++++++++++++++++++++
2 files changed, 115 insertions(+), 3 deletions(-)
diff --git a/plc4go/spi/utils/StopWarn.go b/plc4go/spi/utils/StopWarn.go
index cfdeea6581..c045a8ad49 100644
--- a/plc4go/spi/utils/StopWarn.go
+++ b/plc4go/spi/utils/StopWarn.go
@@ -37,6 +37,7 @@ type stopWarnOptions struct {
interval time.Duration
extraSkipOffset int
includeGoroutinesStack bool
+ registrar func(initial StopWarnTick) (onTick
func(StopWarnTick), onEnd func())
}
// StopWarn gives out warning every interval (default 5 seconds) when a
function doesn't terminate. Usage: `defer StopWarn(log)()`
@@ -58,9 +59,19 @@ func StopWarn(localLog zerolog.Logger, opts
...func(*stopWarnOptions)) func() {
ticker := time.NewTicker(o.interval)
wg := new(sync.WaitGroup)
done := make(chan struct{})
+ startTime := time.Now()
+ var onTick func(StopWarnTick)
+ var onEnd func()
+ if o.registrar != nil {
+ onTick, onEnd = o.registrar(StopWarnTick{
+ ProcessID: o.processId,
+ ProcessInfo: o.processInfo,
+ StartTime: startTime,
+ WarnTime: startTime,
+ })
+ }
wg.Go(func() {
localLog.Trace().Msgf("start checking")
- startTime := time.Now()
for {
localLog.Trace().Msgf("check cycle")
select {
@@ -90,14 +101,24 @@ func StopWarn(localLog zerolog.Logger, opts
...func(*stopWarnOptions)) func() {
TimeDiff("inProgressFor", warnTime,
startTime).
Stringer("stackInfo", stackInfo).
Msgf("%sstill in progress", processId)
+ if onTick != nil {
+ onTick(StopWarnTick{
+ ProcessID: o.processId,
+ ProcessInfo: o.processInfo,
+ StartTime: startTime,
+ WarnTime: warnTime,
+ })
+ }
}
}
})
- start := time.Now()
return func() {
- localLog.Trace().TimeDiff("check duration", time.Now(),
start).Msg("done")
+ localLog.Trace().TimeDiff("check duration", time.Now(),
startTime).Msg("done")
close(done)
wg.Wait() // This is to avoid late logs in case when the
shutdown is really fast
+ if onEnd != nil {
+ onEnd()
+ }
}
}
@@ -135,3 +156,34 @@ func WithStopWarnIncludeGoroutinesStack()
func(*stopWarnOptions) {
o.includeGoroutinesStack = true
}
}
+
+// StopWarnTick describes a single "still in progress" event passed to a
registrar.
+type StopWarnTick struct {
+ ProcessID string
+ ProcessInfo string
+ StartTime time.Time
+ WarnTime time.Time
+}
+
+// WithStopWarnRegistrar registers an external observer that is notified of
the lifecycle
+// of a StopWarn invocation. This allows callers to centralise policies such
as rate-limited
+// goroutine stack dumps across many concurrent StopWarn instances without
each instance
+// emitting its own dump on every tick.
+//
+// register is called once when StopWarn arms, with the initial tick info
(StartTime ==
+// WarnTime). It returns two callbacks:
+// - onTick: invoked on every warn tick, immediately after the existing Warn
log line.
+// May be nil; if nil, ticks are silently ignored by the registrar.
+// - onEnd: invoked exactly once when the returned stop func runs, after
the warner
+// goroutine has drained. May be nil.
+//
+// Neither callback should block for long; the warner goroutine waits on
onTick before
+// processing the next tick. Heavy work (stack dumps, I/O) should be
dispatched to a
+// background goroutine inside the callback.
+func WithStopWarnRegistrar(
+ register func(initial StopWarnTick) (onTick func(StopWarnTick), onEnd
func()),
+) func(*stopWarnOptions) {
+ return func(o *stopWarnOptions) {
+ o.registrar = register
+ }
+}
diff --git a/plc4go/spi/utils/StopWarn_test.go
b/plc4go/spi/utils/StopWarn_test.go
index bb03795dfb..2d737734de 100644
--- a/plc4go/spi/utils/StopWarn_test.go
+++ b/plc4go/spi/utils/StopWarn_test.go
@@ -20,6 +20,7 @@
package utils
import (
+ "sync"
"testing"
"time"
@@ -68,6 +69,65 @@ func TestStopWarn(t *testing.T) {
}
assert.Equalf(t, 0, foundMessages, "%s should contain at least
three warning. Found %d times", logHook.messages, foundMessages)
})
+ t.Run("registrar receives ticks and end", func(t *testing.T) {
+ logger := produceTestingLogger(t)
+ var mu sync.Mutex
+ var initial StopWarnTick
+ var ticks []StopWarnTick
+ ended := false
+ register := func(in StopWarnTick) (func(StopWarnTick), func()) {
+ mu.Lock()
+ initial = in
+ mu.Unlock()
+ return func(t StopWarnTick) {
+ mu.Lock()
+ ticks = append(ticks, t)
+ mu.Unlock()
+ }, func() {
+ mu.Lock()
+ ended = true
+ mu.Unlock()
+ }
+ }
+ func() {
+ defer StopWarn(logger,
+ WithStopWarnInterval(10*time.Millisecond),
+ WithStopWarnProcessId("TestStopWarn"),
+ WithStopWarnRegistrar(register),
+ )()
+ time.Sleep(75 * time.Millisecond)
+ }()
+ mu.Lock()
+ defer mu.Unlock()
+ assert.Equal(t, "TestStopWarn", initial.ProcessID, "registrar
should receive initial tick at arm time")
+ assert.False(t, initial.StartTime.IsZero(), "initial StartTime
should be set")
+ assert.Equal(t, initial.StartTime, initial.WarnTime, "initial
WarnTime equals StartTime")
+ assert.GreaterOrEqual(t, len(ticks), 3, "registrar should
observe multiple ticks")
+ assert.True(t, ended, "registrar onEnd should fire when stop
func returns")
+ for _, tk := range ticks {
+ assert.Equal(t, "TestStopWarn", tk.ProcessID)
+ assert.Equal(t, initial.StartTime, tk.StartTime,
"StartTime should be stable across ticks")
+ assert.False(t, tk.WarnTime.Before(initial.StartTime),
"WarnTime should be at or after StartTime")
+ }
+ })
+ t.Run("registrar onEnd only when nil onTick", func(t *testing.T) {
+ logger := produceTestingLogger(t)
+ var mu sync.Mutex
+ ended := false
+ register := func(StopWarnTick) (func(StopWarnTick), func()) {
+ return nil, func() { mu.Lock(); ended = true;
mu.Unlock() }
+ }
+ func() {
+ defer StopWarn(logger,
+ WithStopWarnInterval(10*time.Millisecond),
+ WithStopWarnRegistrar(register),
+ )()
+ time.Sleep(35 * time.Millisecond)
+ }()
+ mu.Lock()
+ defer mu.Unlock()
+ assert.True(t, ended, "nil onTick should not prevent onEnd from
firing")
+ })
}
type logHook struct {
