NodeSampler

class NodeSampler(val node: ServiceNode, val interval: Duration = DEFAULT_INTERVAL, val maxParked: Int = DEFAULT_MAX_PARKED, readTimeout: Duration = DEFAULT_READ_TIMEOUT, context: CoroutineContext = Dispatchers.Default, nanos: () -> Long = System::nanoTime) : AutoCloseable

One sampler of what a node reports that is read rather than heard, every interval: the head of each partition of each log group the node has positions in, and how many entries each subscription has parked, by its group and name (ParkedSubscription: each context has its own snapshot), counted up to maxParked. Both its health and its gauges read the one sample (latest), so the ledgers are read once per interval, whoever reports them.

Each listener is told every sample, in order, a failing one isolated from the others. A read that fails or takes longer than readTimeout (at most interval) is left out of its sample, and the next one tries again, so a store that hangs delays its readers by that much at most. While the node is not running, a sample reads nothing. start samples at once, then every interval; close stops it.

Constructors

Link copied to clipboard
constructor(node: ServiceNode, interval: Duration = DEFAULT_INTERVAL, maxParked: Int = DEFAULT_MAX_PARKED, readTimeout: Duration = DEFAULT_READ_TIMEOUT, context: CoroutineContext = Dispatchers.Default, nanos: () -> Long = System::nanoTime)

Types

Link copied to clipboard

Properties

Link copied to clipboard
Link copied to clipboard

The last sample taken, NodeSample.EMPTY before the first.

Link copied to clipboard
Link copied to clipboard

Functions

Link copied to clipboard
open override fun close()

Stops sampling.

Link copied to clipboard
fun listen(listener: suspend (NodeSample) -> Unit)

Tells listener of every sample from now on.

Link copied to clipboard
suspend fun sample(): NodeSample

Takes one sample, keeps it as latest and tells every listener of it.

Link copied to clipboard

Samples at once, then every interval; starting twice does nothing.