FluxFlow.Components.FileSystem 3.0.0

FluxFlow.Components.FileSystem

Standalone file system nodes for FluxFlow, built on FluxFlow.Nodes. Every node is a self-contained TPL Dataflow processor — new it up and LinkTo the next node. No engine, registry, or runtime required. Every message travels as a FlowMessage<T> envelope, so a correlation id flows from a request to its result without any node copying it by hand.

Nodes

The transforms (FileReadNode / FileWriteNode) accept a FlowMessage<TRequest> on Input and broadcast a FlowMessage<TResult> on Output carrying the same correlation id. The sources (DirectoryEnumerateNode / FileWatchNode) begin producing once StartAsync is called and mint a fresh correlation id per emitted item.

Every node also exposes broadcast Errors (FlowError) and Events (FlowEvent) ports. Domain failures surface on Errors (carrying the in-flight correlation id for transforms) and do not stop later messages from being processed; transforms emit a success/failure note on Events, and the sources emit started/entry/changed/completed notes there too.

Construction

Each node takes its options and an optional TimeProvider (used for all result and event timestamps) directly:

await using var read = new FileReadNode(new FileReadOptions { BaseDirectory = "data" });
await using var write = new FileWriteNode(new FileWriteOptions { BaseDirectory = "data" });
await using var enumerate = new DirectoryEnumerateNode(new DirectoryEnumerateOptions
{
    Directory = "inbox",
    BaseDirectory = "data"
});
await using var watch = new FileWatchNode(new FileWatchOptions
{
    Directory = "inbox",
    BaseDirectory = "data"
});

Pass a Microsoft.Extensions.Time.Testing.FakeTimeProvider (or any TimeProvider) as the second argument to make timestamps deterministic in tests.

Read / Write transforms

await read.Input.SendAsync(FlowMessage.Create(new FileReadRequest
{
    Path = "logs/output.txt",
    ReadAs = FileReadMode.Text
}));
var result = await read.Output.ReceiveAsync(); // FlowMessage<FileReadResult>

Use ReadAs = FileReadMode.Bytes for raw bytes. Text reads use the request Encoding when provided, otherwise the option DefaultEncoding.

await write.Input.SendAsync(FlowMessage.Create(new FileWriteRequest
{
    Path = "logs/output.txt",
    Content = "hello",
    Mode = FileWriteMode.Overwrite,
    CreateDirectories = true
}));

Bytes can be used instead of Content. When both are set, Bytes wins. Supported write modes are Overwrite, Append, and CreateNew; supported read modes are Text and Bytes.

Watch source

var watch = new FileWatchNode(new FileWatchOptions
{
    Directory = "inbox",
    Filter = "*.json",
    IncludeSubdirectories = false,
    NotifyFilters = ["FileName", "LastWrite", "Size"],
    BaseDirectory = "data",
    InternalBufferSize = 8192,
    BoundedCapacity = 128
});
await watch.StartAsync();
// watch.Output emits FlowMessage<FileWatchEvent> per change; watch.Complete() stops it.

FileWatchEvent carries the changed path, directory, name, change type, and old path/name for rename events. InternalBufferSize optionally sets the underlying watcher buffer and must be between 4096 and 65536 bytes when set.

Directory enumerate source

var enumerate = new DirectoryEnumerateNode(new DirectoryEnumerateOptions
{
    Directory = "inbox",
    Filter = "*.json",
    IncludeSubdirectories = true,
    IncludeFiles = true,
    IncludeDirectories = false,
    MaxEntries = 1000,
    BaseDirectory = "data",
    BoundedCapacity = 128
});
await enumerate.StartAsync();
// enumerate.Output emits FlowMessage<DirectoryEnumerateEntry>, then completes.

DirectoryEnumerateEntry carries the resolved path, source directory, name, entry type, optional byte length, timestamps, and file attributes.

Path resolution

Relative paths are resolved under BaseDirectory when it is set. Relative paths that escape the base directory are rejected. Absolute paths are rejected unless AllowAbsolutePaths is true. When BaseDirectory is not set and AllowAbsolutePaths is false, the current working directory is the implicit base and relative paths that escape it are rejected.

FileReadOptions.MaxBytes defaults to 16777216 (16 MiB). Set it higher for larger files, or set it explicitly to null to keep unlimited reads.