FluxFlow.Components.Projections 3.0.0

FluxFlow.Components.Projections

A standalone event-projection node for FluxFlow, built on the FluxFlow.Nodes kit. No engine required: new the node, post events, link the output.

Node

EventProjectionNode is a FlowNode<ProjectionEvent, EventProjectionSnapshot>. Post a FlowMessage<ProjectionEvent> to Input; the node broadcasts a FlowMessage<EventProjectionSnapshot> on Output carrying the triggering event's correlation id. Errors surface on Errors (FlowError) and diagnostics on Events (FlowEvent).

Example

var node = new EventProjectionNode(new EventProjectionOptions
{
    Name = "failed-operations",
    RateWindowSeconds = 60,
    MaxPreviewChars = 256,
    Filter = new EventFilter
    {
        TypePrefix = "operation.",
        Status = "failed",
        SubjectPrefix = "orders/",
        Attributes = new Dictionary<string, string> { ["tenant"] = "north" }
    }
});

var snapshots = new BufferBlock<FlowMessage<EventProjectionSnapshot>>();
node.Output.LinkTo(snapshots, new DataflowLinkOptions { PropagateCompletion = true });

await node.Input.SendAsync(FlowMessage.Create(new ProjectionEvent
{
    Timestamp = DateTimeOffset.UtcNow,
    Type = "operation.completed",
    Source = "orders",
    Subject = "orders/42",
    Status = "failed",
    Attributes = new Dictionary<string, string> { ["tenant"] = "north" }
}));

The snapshot includes:

• observed event count
• matched event count
• rolling current rate
• first and last matched timestamps
• latest matching event summary
• latest payload preview with a configured length limit

Filters

EventFilter supports:

• event type or type prefix
• subject prefix
• channel prefix
• excluded subject prefix
• excluded channel prefix
• status
• source
• source node id
• component id through an event attribute named componentId
• attribute key/value pairs
• event timestamp range

Filters use ordinal string comparison.

Timing

Pass a TimeProvider to the constructor for deterministic snapshot timestamps in tests or hosts:

var node = new EventProjectionNode(options, timeProvider);

Event rate uses matching event timestamps. Snapshot timestamps use the supplied TimeProvider (defaults to TimeProvider.System).

Final snapshot

With EmitFinalSnapshot = true (and typically EmitEveryMatch = false) the node emits a single closing snapshot when the stream ends. Drain and close it via await node.CompleteWithFinalSnapshotAsync() instead of Complete(); the flush rides the ordered input pump so it lands after every event already posted.

Boundaries

This package has no UI dependency and no host-specific resource assumptions. It depends only on FluxFlow.Nodes. Hosts decide how snapshots are displayed, stored, tested, or forwarded.