1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
// Copyright 2025 Umberto Gotti <umberto.gotti@umbertogotti.dev>
// Licensed under the Apache License, Version 2.0
// http://www.apache.org/licenses/LICENSE-2.0
//! Window-by-count operator that batches stream items into fixed-size chunks.
//!
//! This module provides the [`window_by_count`](WindowByCountExt::window_by_count) operator
//! that groups consecutive stream items into vectors of a specified size.
//!
//! # Overview
//!
//! The `window_by_count` operator collects items into windows (batches) of a fixed size.
//! When the window is full, it emits a `Vec` containing all items in that window.
//! On stream completion, any partial window is also emitted.
//!
//! # Basic Usage
//!
//! ```
//! use fluxion_stream::prelude::*;
//! use fluxion_test_utils::Sequenced;
//! use futures::StreamExt;
//!
//! # #[tokio::main]
//! # async fn main() {
//! let (tx, rx) = async_channel::unbounded();
//! let stream = rx.into_fluxion_stream();
//!
//! let mut windowed = stream.window_by_count::<Sequenced<Vec<i32>>>(3);
//!
//! tx.try_send(Sequenced::new(1)).unwrap();
//! tx.try_send(Sequenced::new(2)).unwrap();
//! tx.try_send(Sequenced::new(3)).unwrap(); // Window complete!
//! tx.try_send(Sequenced::new(4)).unwrap();
//! tx.try_send(Sequenced::new(5)).unwrap();
//! drop(tx); // Partial window [4, 5] emitted on completion
//!
//! // First window: [1, 2, 3]
//! let window1 = windowed.next().await.unwrap().unwrap().into_inner();
//! assert_eq!(window1, vec![1, 2, 3]);
//!
//! // Second window (partial): [4, 5]
//! let window2 = windowed.next().await.unwrap().unwrap().into_inner();
//! assert_eq!(window2, vec![4, 5]);
//! # }
//! ```
//!
//! # Use Cases
//!
//! - **Batch processing**: Process items in groups for efficiency
//! - **Micro-batching**: Balance latency and throughput in data pipelines
//! - **Aggregation windows**: Collect data for periodic analysis
//! - **Protocol framing**: Group bytes or messages into frames
//!
//! # Error Handling
//!
//! When an error occurs, the current partial window is discarded and the error
//! is propagated immediately. This ensures clean error boundaries without
//! emitting potentially incomplete data.
pub use WindowByCountExt;
pub use WindowByCountExt;