Time Channel Adapter

The time adapter generates periodic ticks and injects them as messages into the flow. It is useful for scheduled pipelines, external resource polling, and timer-based workflows.

Availability

The time-channel-adapter module, a transitive dependency of pipelite-core.

Usage

Pipelite.defineFlow("scheduled-flow")
    .fromSource("time://ticker?period=5000")   (1)
    .process("check", (exchange, contribution) -> {
        // logic executed every 5 seconds
    })
    .toSink("slf4j://log")
    .build();

1	Generates a tick every 5000 ms.

The same settings can be applied through the typed TimeSourceConfigurer instead of a query string (see Custom Channel Adapters for the general mechanism):

.fromSource("time://ticker", (TimeSourceConfigurer c) -> c
    .period(5000)
    .timeUnit(TimeUnit.MILLISECONDS))

URL format

time://<name>[?period=<ms>&initialDelay=<ms>&timeUnit=<unit>&batchSize=<n>&durableInbox=<bool>]

Parameter Default Description

Parameter	Default	Description
`name`	(required)	Logical name of the timer (e.g. `ticker`, `heartbeat`)
`period`	`1000`	Tick period in milliseconds
`initialDelay`	`0`	Delay before the first tick
`timeUnit`	`MILLISECONDS`	Unit for `period`/`initialDelay`
`batchSize`	`1`	Ticks drained per poll
`durableInbox`	`false`	Unlike every other source adapter, `time://` defaults the durable inbox off: a tick’s payload is `LocalDateTime.now()` at the instant it fires, regenerated on the adapter’s own schedule regardless of what a crash destroys - there is never anything external or irreplaceable at risk the way there is for a source whose messages arrive from outside (HTTP, Kafka, a tailed file). Opt back in with `durableInbox=true` (or `c.durableInbox(true)` on the configurer) if something downstream actually needs every tick tracked durably.

name

(required)

Logical name of the timer (e.g. ticker, heartbeat)

period

1000

Tick period in milliseconds

initialDelay

0

Delay before the first tick

timeUnit

MILLISECONDS

Unit for period/initialDelay

batchSize

1

Ticks drained per poll

durableInbox

false

Unlike every other source adapter, time:// defaults the durable inbox off: a tick’s payload is LocalDateTime.now() at the instant it fires, regenerated on the adapter’s own schedule regardless of what a crash destroys - there is never anything external or irreplaceable at risk the way there is for a source whose messages arrive from outside (HTTP, Kafka, a tailed file). Opt back in with durableInbox=true (or c.durableInbox(true) on the configurer) if something downstream actually needs every tick tracked durably.

Internal mechanism

The time source (TimeSource) is a polling source, polled by the framework on the schedule of the endpoint. On each tick, a new Exchange is created with a payload containing the current timestamp. It is started and stopped together with the PipeliteContext.