/ 
Acme-Parataxis-v0.1.3
/ Acme::Parataxis::Stream

NAME

Acme::Parataxis::Stream - Chainable Async FRP Streams over Bounded Channels

SYNOPSIS

use v5.40;
use Acme::Parataxis qw[async];
use Acme::Parataxis::Channel;
use Acme::Parataxis::Stream;

async {
    my $raw = Acme::Parataxis::Channel->new( capacity => 1024 );

    # Build and consume an asynchronous stream pipeline:
    my $sink = Acme::Parataxis::Stream->from_channel($raw)
        ->map(     sub ($json)  { decode_json($json)      } )
        ->filter(  sub ($item)  { $item->{priority} > 1   } )
        ->throttle(100)                                   # max 100 items/sec
        ->batch_time(1000)                                # group into 1-sec batches
        ->consume( sub (@batch) { db_bulk_insert(@batch) } );

    # Produce data into the source channel:
    $raw->put($_) for @events;
    $raw->shutdown;

    # Await the pipeline sink until all stages drain:
    $sink->await;
};

DESCRIPTION

An Acme::Parataxis::Stream is a chainable pipeline of asynchronous processing stages backed by bounded Acme::Parataxis::Channels.

Each stage in a stream pipeline is a factory: calling map, filter, batch, batch_time, or throttle allocates a private bounded output channel, spawns an isolated background fiber to process items, and returns a new Stream object wrapping that output channel. Calling consume terminates the pipeline and begins draining items.

Key architectural guarantees:

  • Automatic backpressure: Output channels between stages are strictly bounded (governed by stage_capacity). If a stage processes items slower than its upstream feeder, the intermediate channel fills and the upstream fiber parks. Backpressure propagates naturally all the way to the raw data source without unbounded memory consumption.

  • Cascading graceful teardown: Pipelines shut down cleanly from head to tail. When the source channel shuts down, the first stage reads undef from its input, flushes pending work, and shuts down its output channel. This cascade propagates downstream, unwinding all intermediate fibers until consume finishes.

  • Cancellable stages: Because stages park on standard channels, enclosing with_timeout scopes and cancellation tokens interrupt in-flight stages cleanly.

METHODS

from_channel( $channel, [%opts] )

my $stream = Acme::Parataxis::Stream->from_channel( $raw_ch, stage_capacity => 64 );

Class method. Wraps an existing Acme::Parataxis::Channel as the head of a new stream pipeline.

  • stage_capacity => $int: The buffer capacity assigned to all intermediate channels allocated by stages in this pipeline (default 128). Must be >= 1.

map( sub ($item) { ... } )

$stream->map( sub ($item) { transform($item) } );

Transforms each item by passing it to the provided callback, emitting the callback's return value downstream.

filter( sub ($item) { ... } )

$stream->filter( sub ($item) { $item->{active} } );

Evaluates each item against the callback predicate. Emits the item downstream only if the callback returns true.

batch( $count )

$stream->batch(50);

Groups incoming items into arrayrefs of at most $count items before emitting them downstream. If the upstream channel shuts down mid-batch, any buffered items are emitted as a partial final batch so no data is stranded. $count must be >= 1.

batch_time( $ms )

$stream->batch_time(500);

Groups items by time deadline. When the first item of a batch arrives, a timer of $ms milliseconds is armed. When the deadline fires, the accumulated batch is flushed downstream immediately, even if no further items have arrived. When the source shuts down, pending items are flushed as a final batch.

throttle( $per_second )

$stream->throttle(20);    # max 20 items per second

Paces emissions so that no more than $per_second items pass downstream per second. The first item passes immediately; subsequent items sleep for the remainder of their time window. If backpressure downstream stalls the throttle stage, it does not "burst" to catch up when unblocked. $per_second must be >= 1.

consume( sub (@items) { ... } )

my $sink_fiber = $stream->consume( sub (@batch_or_item) { ... } );

Terminal consumer stage. Invokes the callback for each item (or unpacked batch) arriving at the end of the pipeline. Returns the Acme::Parataxis fiber object driving the consumer. Await this fiber to block until the entire pipeline has drained and terminated.

shutdown( )

$stream->shutdown;

Closes the stream stage's source channel and walks upstream, closing preceding channels. Causes the pipeline to drain and terminate downstream.

cap( )

Returns the configured stage_capacity shared across intermediate channels in this pipeline.

chain( )

Returns the immediate upstream Acme::Parataxis::Stream stage, or undef at the head.

Recipes

Live event processing with count and time flushing

async {
    my $events = Acme::Parataxis::Channel->new( capacity => 500 );

    my $sink = Acme::Parataxis::Stream->from_channel( $events, stage_capacity => 64 )
        ->filter( sub ($ev) { $ev->{level} eq 'ERROR' } )
        ->batch_time(2000)    # flush at least every 2 seconds
        ->consume( sub (@errors) {
            ship_alerts(@errors);
        } );

    # Push events...
    $events->shutdown;
    $sink->await;
};

Behavior Notes

  • Arrayrefs from map are unpacked by consume

    consume treats incoming arrayrefs as batches and unpacks them into @_. If a map callback legitimately returns an arrayref meant to represent a single data item, wrap it in a secondary container (e.g. [ $my_arrayref ]) or a blessed object.

  • Stage callback exceptions terminate the run

    Exceptions thrown inside map, filter, or consume callbacks propagate out of that stage's internal fiber and will terminate the enclosing scheduler run unless guarded. Trap expected errors inside your callbacks.

  • Stacking batch stages creates nested arrays

    Chaining batch_time directly into batch does not create a union constraint; it batches the batches, producing nested arrayrefs (e.g. [ [1, 2], [3, 4] ]).

SEE ALSO

Acme::Parataxis, Acme::Parataxis::Channel, Acme::Parataxis::RateLimiter

AUTHOR

Sanko Robinson https://github.com/sanko

LICENSE

Copyright (C) Sanko Robinson.

This library is free software; you can redistribute it and/or modify it under the terms found in the Artistic License 2.