This version is still in development and is not considered stable yet.

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

Included in default-channel-adapters, transitively available with 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 TimePollingConsumer uses a ScheduledExecutorService via TimeService. On each tick, a new Exchange is created with a payload containing the current timestamp. The consumer is started and stopped together with the PipeliteContext.