File Channel Adapter
The file-channel-adapter module provides a filesystem-based adapter: a tail-polling source that streams newly appended file content, and a sink that writes the flow payload to a file.
Dependency
<dependency>
<groupId>io.pipelite</groupId>
<artifactId>file-channel-adapter</artifactId>
<version>1.0.0-SNAPSHOT</version>
</dependency>Usage as source (tail consumer)
Pipelite.defineFlow("file-tail")
.fromSource("file:///var/log/app.log?startPosition=beginning&period=500") (1)
.process("handle", exchange -> {
String line = exchange.getInputPayloadAs(String.class);
// process one line at a time
})
.toSink("slf4j://log")
.build();| 1 | Polls /var/log/app.log every 500 ms, starting from the beginning of the file. |
The FileTailPollingConsumer polls the file on a dedicated per-endpoint thread via FileTailConsumerService (one ScheduledExecutorService per endpoint, not shared across the adapter). On each poll it reads only the bytes appended since the last known offset, maps them into records via the configured FileRecordMapper, and emits one Exchange per record.
URL format
file://<path>[?<params>]<path> is a filesystem path (absolute or relative, forward or backward slashes accepted) — e.g. file:///var/log/app.log or file://C:\logs\app.log. Dynamic/templated paths (e.g. #{…} expressions) are not supported: the resource must be a static path known at flow-definition time.
Source (consumer) parameters
| Parameter | Default | Description |
|---|---|---|
|
|
|
|
(built-in line mapper) |
Logical name of a |
|
|
Charset used to decode the bytes read from the file. |
|
|
Consumption mode. |
|
|
Polling interval, in the unit given by |
|
|
Delay before the first poll, in the unit given by |
|
|
Time unit for |
|
|
Number of leading lines to discard (e.g. a CSV header), counted from the very beginning of the file. Only supported with the built-in |
With the default startPosition=end, the first poll after the file is discovered establishes the current size as the "already consumed" baseline — it does not emit pre-existing content. Writing to the file immediately after PipeliteContext#start(), before that first poll has run, can race with this baseline check; the safest pattern for deterministic tests or startup scripts is to allow at least one poll cycle to elapse before writing, or to use startPosition=beginning when the existing content must be read deterministically.
|
Usage as sink (producer)
.toSink("file:///var/log/output.log?append=true")The FileProducer writes the Exchange input payload to the target file. byte[] and InputStream payloads are written as-is; CharSequence payloads are encoded with the configured charset; any other payload type is converted via toString() (a warning is logged in that case). Parent directories are created automatically if missing.
Sink (producer) parameters
| Parameter | Default | Description |
|---|---|---|
|
|
|
|
|
Charset used to encode |
Every consumed record is also enriched, on the source side, with two exchange headers useful downstream:
| Header | Description |
|---|---|
|
The file name (last path segment) of the source file the record was read from. |
|
The absolute path of the source file the record was read from. |
Record mapping
The built-in LineRecordMapper splits newly read content into complete lines, stripping the line terminator (\n/\r\n). A trailing, not-yet-terminated line is left unconsumed and completed on a future poll — so partial writes never produce a truncated record. Because line terminators are stripped, a flow that pipes a file:// source straight into a file:// sink needs an explicit processing step to reintroduce them if the sink output must remain line-delimited:
.fromSource("file://" + sourceFile + "?startPosition=beginning")
.process("restore-line-break", (io, c) -> io.setOutputPayload(io.getInputPayloadAs(String.class) + "\n"))
.toSink("file://" + sinkFile + "?append=true")To use a custom mapping strategy (e.g. parsing CSV rows, JSON lines, or fixed-width records instead of plain lines), implement FileRecordMapper<T> and register it by logical name:
public class UpperCaseLineRecordMapper implements FileRecordMapper<String> {
private final LineRecordMapper delegate = new LineRecordMapper();
@Override
public MappingResult<String> map(String newContent) {
MappingResult<String> result = delegate.map(newContent);
List<String> upperCased = result.getRecords().stream()
.map(String::toUpperCase)
.collect(Collectors.toList());
return new MappingResult<>(upperCased, result.getConsumedLength());
}
}MappingResult carries both the mapped records and consumedLength — the number of bytes to advance the persisted offset by. Returning fewer bytes than were passed in (as the built-in mapper does for an incomplete trailing line) leaves the remainder to be re-read on the next poll.
Configuration
File channel behavior is provided via FileChannelConfigurer:
FileChannelConfigurer configurer = configuration -> {
configuration.setStateDirectory(Path.of("/var/lib/myapp/pipelite-file-state"));
configuration.registerMapper("upper", new UpperCaseLineRecordMapper());
};
context.addChannelConfigurer(configurer);With Spring Boot, the configurer can be defined as a @Bean.
Only the first FileChannelConfigurer registered on a context is applied; any additional ones are silently ignored. Fold every setting (state directory, all registerMapper calls) into a single configurer instance rather than calling addChannelConfigurer more than once for this adapter.
|
Available FileChannelConfiguration properties
| Property | Default | Description |
|---|---|---|
|
|
Directory where per-resource tail offsets are persisted. |
|
(empty) |
Registry of named |
State persistence and resume semantics
For every tailed resource, FileTailStateStore persists the last consumed byte offset in one file per resource under stateDirectory, named after the SHA-256 hex digest of the resource’s absolute path (an index.properties file maps digests back to paths, for debugging only — it is never read by the runtime). Reads and writes are synchronized with a FileLock on the state file.
This makes tailing resumable across restarts: a new PipeliteContext pointed at the same stateDirectory picks up exactly where the previous process left off, without re-emitting already-consumed content or re-applying startPosition semantics. If the source file shrinks below the last known offset (e.g. rotation/truncation), the offset is reset to 0 and a warning is logged.
Availability of the mode parameter
FileEndpoint#createConsumer() only accepts mode=tail (the default); any other value throws IllegalArgumentException when the flow is registered. This is a configuration-time failure, not a runtime one — it surfaces as soon as the context starts, before any polling begins.