Periodic Emitter Component
A complete example of a time-based component with start/stop controls, modeled on the ticker component from common-module.
Overview
This component emits messages at configurable intervals. It demonstrates:
- Background goroutine management with
module.BaseandEmit - Start/Stop dashboard controls via
ControlHandler - Leader-only execution
- Restart survival via node metadata and
ReconcileHandler - Graceful shutdown
Complete Implementation
package emitter
import (
"context"
"errors"
"fmt"
"sync"
"time"
"github.com/goccy/go-json"
"github.com/tiny-systems/module/api/v1alpha1"
"github.com/tiny-systems/module/module"
"github.com/tiny-systems/module/pkg/utils"
"github.com/tiny-systems/module/registry"
)
const (
ComponentName = "periodic_emitter"
OutPort = "out"
metadataKeyRunning = "emitter-running"
metadataKeyConfig = "emitter-config"
)
type Context any
// Settings configuration
type Settings struct {
Context Context `json:"context" configurable:"true" title:"Context" description:"Arbitrary message to send each period of time"`
Delay int `json:"delay" required:"true" title:"Delay (ms)" description:"Delay between emissions" minimum:"0" default:"1000"`
}
// ControlStopped is the _control port schema when the emitter is not running
type ControlStopped struct {
Context Context `json:"context" required:"true" title:"Context"`
Status string `json:"status" title:"Status" readonly:"true"`
Start bool `json:"start" format:"button" title:"Start" required:"true"`
}
// ControlRunning is the _control port schema when the emitter is running
type ControlRunning struct {
Context Context `json:"context" required:"true" title:"Context" readonly:"true"`
Status string `json:"status" title:"Status" readonly:"true"`
Stop bool `json:"stop" format:"button" title:"Stop" required:"true"`
}
type Component struct {
module.Base // provides Emit(), State(), Identity(), Client()
settings Settings
cancelFunc context.CancelFunc
cancelFuncLock *sync.Mutex
runLock *sync.Mutex
// settingsFromPort tracks whether OnSettings/OnControl provided fresh
// values since the runner started, so reconcile won't restore stale
// metadata over them.
settingsFromPort bool
}
func (t *Component) Instance() module.Component {
return &Component{
cancelFuncLock: &sync.Mutex{},
runLock: &sync.Mutex{},
settings: Settings{Delay: 1000},
}
}
func (t *Component) GetInfo() module.ComponentInfo {
return module.ComponentInfo{
Name: ComponentName,
Description: "Periodic Emitter",
Info: "Emits the configured context on Out at a fixed interval. Click Start to begin, Stop to pause. Survives pod restarts via metadata persistence.",
Tags: []string{"timer", "trigger"},
}
}
// OnSettings receives settings from the SettingsPort.
func (t *Component) OnSettings(_ context.Context, msg any) error {
in, ok := msg.(Settings)
if !ok {
return fmt.Errorf("invalid settings")
}
t.settings = in
t.settingsFromPort = true
if t.isRunning() {
t.persistMetadata()
}
return nil
}
// OnReconcile restores running state from node metadata after a pod restart.
func (t *Component) OnReconcile(ctx context.Context, node v1alpha1.TinyNode) error {
if node.Status.Metadata == nil {
return nil
}
if _, running := node.Status.Metadata[metadataKeyRunning]; !running {
return nil
}
if t.isRunning() {
return nil
}
// Only the leader resumes background work
if !utils.IsLeader(ctx) {
return nil
}
if !t.settingsFromPort {
if configStr, ok := node.Status.Metadata[metadataKeyConfig]; ok {
var cfg Settings
if err := json.Unmarshal([]byte(configStr), &cfg); err == nil {
t.settings = cfg
t.settingsFromPort = true
}
}
}
go t.run(ctx)
return nil
}
// OnControl handles Start/Stop dashboard clicks (ControlHandler capability).
func (t *Component) OnControl(ctx context.Context, msg any) error {
if msg == nil {
return nil
}
if !utils.IsLeader(ctx) {
return nil
}
switch ctrl := msg.(type) {
case ControlRunning:
if ctrl.Stop {
t.clearMetadata()
return t.stop()
}
case ControlStopped:
t.settings.Context = ctrl.Context
t.settingsFromPort = true
t.persistMetadata()
if t.isRunning() {
return nil
}
go t.run(context.Background())
}
return nil
}
func (t *Component) run(ctx context.Context) error {
t.runLock.Lock()
defer t.runLock.Unlock()
// Long-running loop: use a background context, bridged to the caller's
runCtx, runCancel := context.WithCancel(context.Background())
defer runCancel()
go func() {
select {
case <-ctx.Done():
runCancel()
case <-runCtx.Done():
}
}()
t.setCancelFunc(runCancel)
// Refresh the dashboard widget
t.Emit(context.Background(), v1alpha1.ControlPort, t.getControl())
defer func() {
t.setCancelFunc(nil)
t.Emit(context.Background(), v1alpha1.ControlPort, t.getControl())
}()
timer := time.NewTimer(0) // first tick fires immediately
defer timer.Stop()
for {
select {
case <-timer.C:
// Emit from the background loop via Base.Emit and CHECK the Result
if err := t.Emit(runCtx, OutPort, t.settings.Context).Err(); err != nil {
// downstream error — log/inspect, keep ticking
_ = err
}
timer.Reset(time.Duration(t.settings.Delay) * time.Millisecond)
case <-runCtx.Done():
if errors.Is(runCtx.Err(), context.Canceled) {
return nil
}
return runCtx.Err()
}
}
}
// persistMetadata stores running state on the node so a restarted pod resumes.
// The reconcile-port payload is an updater func(n *v1alpha1.TinyNode) error.
func (t *Component) persistMetadata() {
configBytes, _ := json.Marshal(t.settings)
t.Emit(context.Background(), v1alpha1.ReconcilePort, func(n *v1alpha1.TinyNode) error {
if n.Status.Metadata == nil {
n.Status.Metadata = make(map[string]string)
}
n.Status.Metadata[metadataKeyRunning] = "true"
n.Status.Metadata[metadataKeyConfig] = string(configBytes)
return nil
})
}
func (t *Component) clearMetadata() {
t.Emit(context.Background(), v1alpha1.ReconcilePort, func(n *v1alpha1.TinyNode) error {
if n.Status.Metadata == nil {
return nil
}
delete(n.Status.Metadata, metadataKeyRunning)
delete(n.Status.Metadata, metadataKeyConfig)
return nil
})
}
func (t *Component) setCancelFunc(f context.CancelFunc) {
t.cancelFuncLock.Lock()
defer t.cancelFuncLock.Unlock()
t.cancelFunc = f
}
func (t *Component) isRunning() bool {
t.cancelFuncLock.Lock()
defer t.cancelFuncLock.Unlock()
return t.cancelFunc != nil
}
func (t *Component) stop() error {
t.cancelFuncLock.Lock()
defer t.cancelFuncLock.Unlock()
if t.cancelFunc == nil {
return nil
}
t.cancelFunc()
return nil
}
func (t *Component) Ports() []module.Port {
return []module.Port{
{
Name: v1alpha1.SettingsPort,
Label: "Settings",
Configuration: t.settings,
},
{
Name: v1alpha1.ReconcilePort,
Label: "Reconcile",
},
{
Name: OutPort,
Label: "Out",
Source: true,
Position: module.Right,
Configuration: new(Context),
},
{
Name: v1alpha1.ControlPort,
Label: "Control",
Source: true,
Configuration: t.getControl(),
},
}
}
// getControl varies the control schema by state — the dashboard re-renders
// whichever struct is current.
func (t *Component) getControl() interface{} {
if t.isRunning() {
return ControlRunning{
Status: "Running",
Context: t.settings.Context,
Stop: true,
}
}
return ControlStopped{
Context: t.settings.Context,
Status: "Not running",
Start: true,
}
}
// Handle is unreachable for this component — every declared port is either
// system (dispatched via capabilities) or source (emit-only). The stub
// guards against accidental routing.
func (t *Component) Handle(_ context.Context, _ module.Handler, port string, _ any) module.Result {
return module.Fail(fmt.Errorf("periodic_emitter has no business-port input: got %q", port))
}
// OnDestroy stops the loop when the node is deleted.
func (t *Component) OnDestroy(_ map[string]string) {
t.stop()
}
var (
_ module.Component = (*Component)(nil)
_ module.SettingsHandler = (*Component)(nil)
_ module.ReconcileHandler = (*Component)(nil)
_ module.ControlHandler = (*Component)(nil)
_ module.Destroyer = (*Component)(nil)
)
func init() {
registry.Register((&Component{}).Instance())
}
Usage Example
Settings Configuration
The _settings port configuration (JSON):
{
"delay": 30000,
"context": {
"source": "scheduled_task",
"taskId": "cleanup_job"
}
}
Control Panel UI
The _control port renders ControlStopped or ControlRunning depending on state:
+--------------------------------------+
| Context: {"source": ...} |
| Status: Running |
| |
| [ Stop ] |
+--------------------------------------+
Output
Each tick emits the configured context on out:
{
"source": "scheduled_task",
"taskId": "cleanup_job"
}
Key Patterns Demonstrated
1. ControlHandler Interface
The SDK capability for dashboard controls:
type ControlHandler interface {
OnControl(ctx context.Context, control any) error
}
The runner decodes the control payload into whichever struct the _control port currently declares (ControlRunning vs ControlStopped), so a type switch dispatches Start vs Stop.
2. Background Emission via module.Base
Components that emit outside Handle embed module.Base (which satisfies EmitterAware) and call Emit from their goroutines:
if err := t.Emit(runCtx, OutPort, t.settings.Context).Err(); err != nil {
// downstream failed
}
Emit returns a module.Result — check it. Never fire a handler with go output(...); that drops the Result.
3. Leader-Only Execution
Only the leader replica runs the timer:
if !utils.IsLeader(ctx) {
return nil
}
4. Restart Survival via Metadata
Persist intent through the reconcile port with an updater function, and restore it in OnReconcile:
t.Emit(context.Background(), v1alpha1.ReconcilePort, func(n *v1alpha1.TinyNode) error {
n.Status.Metadata[metadataKeyRunning] = "true"
return nil
})
5. Graceful Shutdown
A stored context.CancelFunc plus a Destroyer implementation stops the loop on node deletion:
func (t *Component) OnDestroy(_ map[string]string) {
t.stop()
}
Common Use Cases
1. Scheduled Data Sync
{ "delay": 3600000, "context": { "task": "sync_users" } }
2. Health Check Pings
{ "delay": 30000, "context": { "type": "health_check" } }
3. Metrics Collection
{ "delay": 60000, "context": { "action": "collect_metrics", "targets": ["cpu", "memory", "disk"] } }
For calendar-style schedules ("9 AM on weekdays"), see the cron component in common-module — same patterns, cron expression instead of a fixed delay.
Extension Ideas
- Cron Expressions: Support cron-style scheduling (see
common-module/components/cron) - Jitter: Add random jitter to prevent thundering herd
- Max Emissions: Stop automatically after N ticks
- Dynamic Interval: Adjust interval based on conditions