v0.13.0

Changed

• lazyflow-http (breaking): SseConfig gains max_event_size
  (default 16 MiB), bounding the bytes one event may hold while it is read:
  data, type and id plus the line in progress. A server exceeding it ends the
  pipe with an error and is not reconnected to. An exhaustive SseConfig
  struct literal needs the new field; ..Default::default() covers it.
• lazyflow-http: a WebSocket connection now lives as long as its incoming
  pipe. Once that pipe ends or is dropped the socket closes and WsSender
  sends fail with PipeError::Closed.
• lazyflow-http: SSE decodes invalid UTF-8 to U+FFFD, as the
  specification requires, instead of failing the stream.
• lazyflow: par_eval_map keeps at most concurrency results in flight,
  running or finished but not yet emitted, and reports an upstream failure
  after the results of the items before it.
• lazyflow: throttle passes on one element per pull, so downstream sees
  each element as soon as its interval allows.
• lazyflow: from_stream hands over the items a stream has ready, up to
  the chunk size, instead of waiting to fill a chunk.

Fixed

• lazyflow: chunks_timeout no longer drops data held by an upstream pull
  that is still running when the timer fires, and treats size 0 as 1 instead
  of emitting empty batches forever.
• lazyflow: debounce no longer drops an element or error held by an
  upstream pull that is still running when the quiet period ends.
• lazyflow: par_eval_map workers are cancelled when the pipe stops or an
  earlier worker fails, and finished results are emitted while upstream is
  pending.
• lazyflow: a broadcast or partition branch dropped before being
  consumed no longer stalls the other branches.
• lazyflow: a concurrently background failure ends a main pull that is
  waiting.
• lazyflow: chunks(n) emits full batches whatever the source chunking;
  only the last batch may be short.
• lazyflow: un_none_terminate stays terminated after the first None.
• lazyflow: interleave alternates element by element whatever the
  source chunking.
• lazyflow: lines() and lines_with_max_len apply the length limit to
  newline-terminated lines too.
• lazyflow-http: SSE handles characters and CRLF split across body
  chunks, bare CR line endings, and keeps the Last-Event-ID resume point
  across id-only frames and reconnects; an empty id clears it.
• lazyflow-macros: a bare-return #[pipe_fn] async fn may .await.

Full Changelog: v0.12.0...v0.13.0

v0.8.0

Full Changelog: v0.7.0...v0.8.0

v0.7.0

Full Changelog: v0.6.2...v0.7.0

v0.6.2

What's new

• GitHub Actions CI -- fmt, clippy, and test on every push/PR (Linux + macOS, Nix-based)
• Criterion benchmark suite -- throughput baselines for core operators (map, filter, flat_map, chunks, fold, merge)
• CHANGELOG.md -- full version history from v0.1.0 through v0.6.2
• pipe-grpc in README -- install instructions, streaming example, crate ecosystem table
• crates.io metadata -- version fields on path dependencies, keywords/categories on all crates
• TRACKER.md sync -- ADR statuses now reflect shipped work (SSE, WebSocket, gRPC all shipped)

v0.6.1

Re-export tonic types from pipe-grpc so users don't need a direct tonic dependency.

• pipe_grpc::{Request, Response, Status, Code, Channel, Streaming} available directly
• pipe_grpc::async_trait gated behind server feature (on by default)
• Client-only consumers can use default-features = false to skip codegen deps

v0.6.0

New pipe-grpc crate -- gRPC streaming for pipe.

pipe-grpc (new, v0.1.0)

• streaming::from_tonic -- wrap tonic Streaming<T> into Pipe<T> (consume server-streaming RPCs)
• serve::to_stream -- convert Pipe<T> into a tonic-compatible response stream (serve streaming RPCs)
• Bidirectional error mapping between tonic::Status and PipeError
• Integration tests against in-process tonic server

Other changes

• ADR-003: gRPC connector decision record
• README rewrite for clarity and narrative flow
• Code quality fixes: docs, examples, duplicate Arc::clone
• Added protobuf to flake.nix devShell

v0.5.1 -- WebSocket source + sink

New: WebSocket source + sink in pipe-http

Bidirectional WebSocket communication via pipe_http::ws.

use pipe_http::ws;

let (incoming, sender) = ws::connect("wss://example.com/ws");

// Send from any task
let s = sender.clone();
tokio::spawn(async move { s.send_text("hello").await });

// Process incoming as a pipe
incoming
    .and_then(|m| m.text())
    .for_each(|text| println!("{text}"))
    .await?;

Features

• ws::connect(url) -- returns (Pipe<WsMessage>, WsSender) synchronously; connection established lazily on first pull
• WsSender -- cloneable, channel-based write handle decoupled from pong responses
• WsMessage -- Text, Binary, Close variants with Option-returning text() and bytes() accessors
• Automatic ping/pong handling (transparent to user)
• 16 MiB incoming message size limit
• Sender detects closed connection (PipeError::Closed)
• 8 integration tests against in-process axum WebSocket server

No auto-reconnection by design

WebSocket has no built-in resume protocol (unlike SSE's Last-Event-ID). For app-level reconnection, use pipe's retry() or handle_error_with().

Install

[dependencies]
pipe-http = { git = "https://github.com/hamidr/pipe" }

263 total workspace tests, zero warnings.

v0.5.0 -- SSE connector and architecture decision records

New crate: pipe-http

Server-Sent Events (SSE) source for consuming HTTP event streams.

use pipe_http::sse;

sse::connect("https://example.com/events")
    .filter(|e| e.event.as_deref() == Some("update"))
    .map(|e| e.data)
    .for_each(|data| println!("{data}"))
    .await?;

Features

• sse::connect(url) -- returns Pipe<SseEvent> with auto-reconnection
• sse::connect_with(SseConfig) -- custom headers, backoff, Last-Event-ID resume
• W3C EventSource spec-compliant parser: event/data/id/retry fields, multiline data, comments, BOM stripping, null-in-id rejection
• 16 MiB line buffer limit to prevent OOM from malicious servers
• Strict UTF-8 validation (rejects invalid bytes)
• Smart reconnection: retries on 5xx and connection errors, fails immediately on 4xx
• Server retry: field honored for backoff tuning
• 22 tests (16 unit + 6 integration against in-process axum server)
• Example: sse_consumer connecting to Wikimedia recent changes stream

Install

[dependencies]
pipe-http = { git = "https://github.com/hamidr/pipe" }

Architecture Decision Records

• ADR-001: Operator fusion strategy (deferred -- optimize after users, not before)
• ADR-002: Connector strategy -- SSE first, then WebSocket, then message brokers
• docs/TRACKER.md for ADR/PRD tracking

v0.4.0 -- Error unification, safety hardening, and edition 2024

Breaking changes

• Unified error types: removed PipeErr type alias. The Operator trait now returns Result<B, PipeError> instead of Result<B, Box<dyn Error>>. PipeResult<T> is now Result<T, PipeError>. The #[operator] macro auto-converts Box<dyn Error> at the boundary, so most user code compiles unchanged.
• broadcast_through() no longer takes an n parameter. The branch count is derived from transforms.len().
• from_pull_once() and generate_once() return PipeError on double materialization instead of panicking.
• Rust edition bumped to 2024. Minimum supported Rust version is now 1.85.

New features

• #[pipe_fn] macro now documented and exported in prelude.
• from_stream_buffered() and into_stream_buffered() for tunable stream interop channel sizes.
• lines_with_max_len() to guard against OOM from long lines (default limit: 1 MiB).
• Pipe::from_reader_sized() for custom AsyncRead buffer sizes.
• Signal::modify() for in-place mutation with subscriber notification.
• Emitter::emit_all() for batch element emission.
• Expanded Sink constructors: new, collect, count, first, last, drain.
• TcpConnection::into_bytes() and into_split() documented.
• file::read_sized() documented.

Safety and security fixes

• PullLines: max line length limit prevents OOM from newline-free input (P0 fix).
• Integer overflow fixes in intersperse and PullChunks capacity calculations (saturating_mul, div_ceil).
• par_eval_map_unordered: worker tasks now tracked via JoinSet and aborted on early exit, preventing leaked tasks.
• chunks(0) and sliding_window(0) clamp to 1 instead of panicking.
• Security documentation added for distinct() memory growth, group_adjacent_by() unbounded groups, path traversal in file I/O, lossy UTF-8 in lines(), and TCP timeout strategy.

Internal improvements

• Extracted drain_to_channel() helper, deduplicating 5 inner-pipe drain loops across concurrency and operator modules.
• Optimized intersperse buffer pre-allocation per chunk.
• Fixed stale "knot-pipe" references in lib.rs doc comments.
• Added pipe_fn to prelude exports.
• Cargo.toml: added keywords, categories, readme, exclude patterns across all three crates. pipe-macros now has description, license, and repository fields.

Testing

• 8 property-based tests via proptest: partition roundtrip, chunks/unchunks identity, distinct idempotency, map length preservation, take+skip partition, fold sum, filter, enumerate.
• Unit tests for debounce, eval_for_each, empty pipe through 13 operators, error propagation through 5 operator types.
• pipe-io error case tests: directory reads, empty files, mixed line endings, custom buffer sizes, raw TCP bytes, mid-stream disconnect, max line length enforcement.
• Total: 220 tests, zero warnings.

v0.3.1 -- Performance improvements

Reduce allocations and overhead across core operators and concurrency primitives.

Changes

• PullFilter: filter in-place with Vec::retain instead of allocating a new Vec per chunk

• PullAndThen: reuse a persistent buffer across next_chunk calls, eliminating repeated heap allocations for filter_map operations

• par_eval_map: replace per-element oneshot channels with JoinHandles, removing one channel allocation per element. Decouple chunk pulling from task spawning via an intermediate VecDeque buffer.

• BroadcastReceiver: use Arc::try_unwrap to take ownership of shared chunks without cloning when the receiver holds the last reference

Cleanup

• Remove unused to_pascal_case function from pipe-macros
• Remove unused PipeError imports from cancel and meter modules
• Remove redundant name field from PullMeter