Plugins and lifecycle hooks

Extend the StoatFlow runtime with custom metrics, HTTP endpoints, and health indicators via the RuntimePlugin interface and lifecycle listeners.

The :runtime module is a thin, extensible wrapper around the engine. Two extension points let you add your own behaviour without forking it: a RuntimePlugin runs at startup with access to the runtime's shared components (metrics registry, HTTP server, health registry), and a RuntimeLifecycleListener observes the runtime as it starts and stops.

PluginContext exposes the runtime's HTTP server, metrics registry, and health registry — not the stream-processing engine or a state-store handle. Plugins extend the operational surface (endpoints, metrics, probes); they cannot run interactive state queries against your topology.

The RuntimePlugin interface

A plugin is a class implementing RuntimePlugin. It has a stable id, an initialize(context) method called once during startup, and an optional shutdown() for cleanup.

interface RuntimePlugin {
    val id: String
    fun initialize(context: PluginContext)
    fun shutdown() {}
}

id — a short, descriptive name without spaces (e.g. "prometheus", "otel", "audit-log"). Used for logging and diagnostics. Plugins initialize in registration order.
initialize(context) — called during startup, before the HTTP server starts and before stream processing begins. The metrics registry exists, the HTTP server is created but not yet running. This is where you register HTTP handlers, metrics, and health indicators.
shutdown() — called during shutdown, after stream processing has stopped. The metrics registry is still available; the HTTP server may still be running. Release anything you acquired in initialize.

The PluginContext surface

initialize(context) receives a PluginContext. These are the only members:

Member	Returns	Purpose
`meterRegistry()`	`MeterRegistry?`	Micrometer registry for custom metrics. `null` if metrics are disabled.
`httpServer()`	`RuntimeHttpServer?`	The HTTP server, for registering custom endpoints. `null` if HTTP is disabled. Handlers must be registered during `initialize` — once the server starts, registration throws `IllegalArgumentException`.
`metricsPrefix()`	`String`	The configured metrics prefix (e.g. `"stoatflow"`). Prefix your own meters with it for consistency.
`applicationId()`	`String`	The `stoatflow.application-id` from configuration.
`healthIndicatorRegistry()`	`HealthIndicatorRegistry`	Register custom health indicators for the liveness / readiness probes.

meterRegistry() and httpServer() are nullable — they return null when metrics or the HTTP server are disabled in config. Always guard the call (?.let { ... }) so a plugin degrades gracefully instead of throwing on a misconfigured runtime.

A complete plugin

This plugin registers a custom counter, a custom HTTP endpoint, and a custom health indicator — exercising every part of the PluginContext.

package com.example

import io.stoatflow.runtime.health.Health
import io.stoatflow.runtime.health.HealthIndicator
import io.stoatflow.runtime.plugin.PluginContext
import io.stoatflow.runtime.plugin.RuntimePlugin

class AuditPlugin : RuntimePlugin {
    override val id = "audit"

    override fun initialize(context: PluginContext) {
        val prefix = context.metricsPrefix()

        // 1. Custom metric — prefix it to match the built-in meters.
        context.meterRegistry()?.let { registry ->
            registry.counter("$prefix.audit.events").increment()
        }

        // 2. Custom HTTP endpoint (com.sun.net.httpserver.HttpHandler).
        //    Must be registered now — before the server starts.
        context.httpServer()?.registerHandler("/audit") { exchange ->
            val body = """{"application":"${context.applicationId()}"}"""
                .toByteArray()
            exchange.responseHeaders.add("Content-Type", "application/json")
            exchange.sendResponseHeaders(200, body.size.toLong())
            exchange.responseBody.use { it.write(body) }
        }

        // 3. Custom health indicator — surfaces in /health/ready and /health/live.
        context.healthIndicatorRegistry().register("audit-sink", AuditHealthIndicator())
    }

    override fun shutdown() {
        // Release anything acquired in initialize().
    }
}

class AuditHealthIndicator : HealthIndicator {
    override fun health(): Health =
        Health.up().withDetail("audit-sink", "connected").build()
}

package com.example;

import io.stoatflow.runtime.health.Health;
import io.stoatflow.runtime.health.HealthIndicator;
import io.stoatflow.runtime.plugin.PluginContext;
import io.stoatflow.runtime.plugin.RuntimePlugin;

import java.io.OutputStream;
import java.nio.charset.StandardCharsets;

public class AuditPlugin implements RuntimePlugin {

    @Override
    public String getId() {
        return "audit";
    }

    @Override
    public void initialize(PluginContext context) {
        String prefix = context.metricsPrefix();

        // 1. Custom metric — prefix it to match the built-in meters.
        var registry = context.meterRegistry();
        if (registry != null) {
            registry.counter(prefix + ".audit.events").increment();
        }

        // 2. Custom HTTP endpoint (com.sun.net.httpserver.HttpHandler).
        //    Must be registered now — before the server starts.
        var server = context.httpServer();
        if (server != null) {
            server.registerHandler("/audit", exchange -> {
                byte[] body = ("{\"application\":\"" + context.applicationId() + "\"}")
                    .getBytes(StandardCharsets.UTF_8);
                exchange.getResponseHeaders().add("Content-Type", "application/json");
                exchange.sendResponseHeaders(200, body.length);
                try (OutputStream os = exchange.getResponseBody()) {
                    os.write(body);
                }
            });
        }

        // 3. Custom health indicator — surfaces in /health/ready and /health/live.
        context.healthIndicatorRegistry().register("audit-sink", new AuditHealthIndicator());
    }

    @Override
    public void shutdown() {
        // Release anything acquired in initialize().
    }
}

class AuditHealthIndicator implements HealthIndicator {
    @Override
    public Health health() {
        return Health.up().withDetail("audit-sink", "connected").build();
    }
}

In Kotlin the id is an abstract val; in Java it surfaces as getId(). The HTTP handler is the JDK's com.sun.net.httpserver.HttpHandler — a functional interface taking an HttpExchange, so it accepts a lambda directly.

Registering a custom metric

Call context.meterRegistry() and use the standard Micrometer API (counter, gauge, timer, …). Prefix meter names with context.metricsPrefix() so they sit alongside the built-in StoatFlow meters and inherit the runtime's common tags. The new meters appear on the /metrics scrape endpoint. See Metrics for the built-in catalogue.

Registering an HTTP handler

Call context.httpServer()?.registerHandler(path, handler) during initialize. The path is a context root (e.g. /audit); the underlying JDK HttpServer routes by longest-prefix match, so /audit also catches /audit/anything. The handler is a com.sun.net.httpserver.HttpHandler; the server runs each request on a virtual thread. Registration must happen during initialize — calling registerHandler after the server has started throws IllegalArgumentException. See REST API for the built-in endpoints.

Registering a health indicator

Call context.healthIndicatorRegistry().register(name, indicator). A HealthIndicator implements health() (readiness) and may override livenessHealth() (liveness — defaults to the same as health()). The registry offers three registration methods:

Method	Effect
`register(name, indicator)`	Registers for both liveness and readiness.
`registerReadiness(name, indicator)`	Readiness probe only.
`registerLiveness(name, indicator)`	Liveness probe only.

Use readiness-only when a dependency being down should take the instance out of the load-balancer pool but not trigger a restart; use liveness-only for conditions where a restart is the right remedy. See Health checks for the built-in indicators and the probe semantics.

Lifecycle hooks

A RuntimeLifecycleListener is a single-method callback (a Kotlin fun interface) invoked as the runtime transitions through its lifecycle. Unlike a plugin, it has no PluginContext — it receives the LifecycleEvent and the StoatFlowRuntime instance (for state queries).

fun interface RuntimeLifecycleListener {
    fun onEvent(event: LifecycleEvent, runtime: StoatFlowRuntime)
}

LifecycleEvent has four values, emitted in order:

Event	When	State at this point
`PRE_START`	Before initialization	Config validated; no resources allocated yet.
`POST_START`	After successful startup	HTTP server running, metrics collecting, stream processing active.
`PRE_STOP`	Before shutdown begins	Shutdown requested; processing still active.
`POST_STOP`	After shutdown completes	Processing stopped, HTTP server stopped, metrics registry closed.

import io.stoatflow.runtime.LifecycleEvent
import io.stoatflow.runtime.StoatFlowRuntime
import io.stoatflow.runtime.plugin.RuntimeLifecycleListener

val listener = RuntimeLifecycleListener { event, runtime ->
    when (event) {
        LifecycleEvent.POST_START -> println("Started: ${runtime.state()}")
        LifecycleEvent.PRE_STOP -> println("Stopping: ${runtime.state()}")
        else -> {}
    }
}

import io.stoatflow.runtime.LifecycleEvent;
import io.stoatflow.runtime.plugin.RuntimeLifecycleListener;

RuntimeLifecycleListener listener = (event, runtime) -> {
    switch (event) {
        case POST_START -> System.out.println("Started: " + runtime.state());
        case PRE_STOP -> System.out.println("Stopping: " + runtime.state());
        default -> { }
    }
};

Use a lifecycle listener for one-off side effects tied to startup/shutdown transitions (notify an external system, flush a buffer). Use a plugin when you need the PluginContext to register endpoints, metrics, or health indicators. A plugin's initialize/shutdown and the POST_START/POST_STOP events overlap but differ in timing — initialize runs before the HTTP server starts; POST_START fires after it is running.

Registering plugins and listeners

Register both on the runtime builder. With the batteries-included StoatFlowRuntime.fromConfig(...) entry point, reach the builder through the configure { ... } block; with the explicit builder, call the methods directly.

import io.stoatflow.runtime.StoatFlowRuntime

// Via fromConfig — configure block exposes the builder
val runtime = StoatFlowRuntime.fromConfig(
    topologyBuilder = { buildTopology(it) },
    configure = {
        addPlugin(AuditPlugin())
        addLifecycleListener(listener)
    },
)
runtime.start()
runtime.awaitTermination()

import io.stoatflow.runtime.StoatFlowRuntime;

// Via fromConfig — configure block exposes the builder
var runtime = StoatFlowRuntime.fromConfig(
    Main::buildTopology,
    builder -> {
        builder.addPlugin(new AuditPlugin());
        builder.addLifecycleListener(listener);
    }
);
runtime.start();
runtime.awaitTermination();

The builder also offers addHealthIndicator(name, indicator) as a shortcut for registering a health indicator without writing a plugin — it registers for both liveness and readiness. For finer control (readiness-only, liveness-only, or dynamic registration), go through a plugin and context.healthIndicatorRegistry().

Next steps

Metrics — the built-in Micrometer catalogue and the /metrics scrape format your custom meters join.
Health checks — built-in indicators and the liveness vs readiness semantics your custom indicators participate in.
REST API — the built-in HTTP endpoints alongside which your custom handlers are served.
Runtime configuration — toggling the HTTP server and metrics that your plugin depends on.

Pause and resume

Pause and resume record processing at runtime with POST /pause and /unpause — for maintenance windows and downstream backpressure, without restarting the instance.

Docker images

Build a container image for your StoatFlow app with the StoatFlow build conventions (Jib) — base image, JVM flags, reproducible builds, container-aware heap sizing, ports and env — on Gradle or Maven.