Iterator Adapters

Level 2 — Control Flow & Data Structures Methods like .map(), .filter(), .enumerate() that transform iterators lazily.

1. Prerequisites

• Iterator — The lazy sequence of items that adapters attach to.
• Closures (|args| body) — The inline anonymous functions (like |x| x + 1) that tell the adapters exactly what to do.

2. Term Category

Rust Abstraction (lazy iterator transformers): Similar concepts exist in many languages, such as array methods in JavaScript (array.map().filter()) or the Streams API in Java. However, Rust's adapters are famous for being strictly lazy and compiling down to hyper-optimized machine code.

3. Explanation

(1) Design Motivation — "Why did we design this?"

Once you have an Iterator producing values, you usually want to do something with them. You might want to transform them (e.g., multiply every number by 2) or filter them (e.g., keep only the even numbers).

In older languages, you'd write a manual for loop, add an if statement inside, and .push() the results to a new array. This is verbose and error-prone.

Rust provides Iterator Adapters: methods that attach directly to an Iterator to modify the sequence. Crucially, they are lazy. Calling .map() doesn't actually execute any math; it just returns a brand new Iterator that promises to do the math later when it is finally consumed. This allows you to chain a dozen adapters together to build complex data pipelines without allocating any intermediate memory.

(2) Reality Metaphor

If an Iterator is a factory conveyor belt carrying raw products, an Iterator Adapter is a specialized robotic arm you bolt onto the side of the belt.

One robotic arm (.filter()) is programmed to inspect the products and knock the defective ones off the belt. The next robotic arm down the line (.map()) is programmed to paint the remaining products red.

You can bolt as many robotic arms onto the belt as you want. But remember: the robotic arms don't do anything until the factory boss turns the main conveyor belt motor on (by looping over it or .collect()ing it).

(3) Rust Code Examples

Short Snippet (Map and Filter)

fn main() {
    let numbers = vec![1, 2, 3, 4, 5];

    // .filter() keeps only the even numbers
    // .map() multiplies them by 10
    // .collect() turns the machine on!
    let transformed: Vec<i32> = numbers.into_iter()
        .filter(|x| x % 2 == 0)
        .map(|x| x * 10)
        .collect();

    println!("{:?}", transformed); // [20, 40]
}

Fuller Example (The enumerate Adapter)

In languages like Python, you often use enumerate to get the index and the item at the same time. Rust has an adapter for this!

fn main() {
    let fruits = vec!["Apple", "Banana", "Cherry"];

    // `.enumerate()` wraps each item in a tuple: (index, item)
    for (index, fruit) in fruits.iter().enumerate() {
        println!("Fruit #{} is {}", index, fruit);
    }
}

4. Common Mistakes & Pitfalls

Mistake 1: Misunderstanding Iterator Adapters Scoping and Lifecycle Rules

The mistake: Assuming Iterator Adapters instances remain valid beyond their declaring scope block or across asynchronous boundaries without explicit lifetime tracking.

Why it's wrong: Rust strictly enforces lexical scope boundaries and non-lexical lifetimes (NLL) at compile time. Accessing dropped values or failing to handle variable drop order results in compiler errors such as E0597 or E0382.

Incorrect:

fn get_ref() -> &str {
    let s = String::from("iterator_adapters_data");
    &s // ❌ Error E0106/E0515: returns a reference to data owned by the current function
}

Fix:

fn get_string() -> String {
    let s = String::from("iterator_adapters_data");
    s // Ownership of the String is transferred directly to the caller
}

Mistake 2: Mutating Iterator Adapters State Without Exclusive Ownership or mut Borrowing

The mistake: Attempting to mutate data associated with Iterator Adapters through an immutable reference &T or without specifying mut in variable declarations.

Why it's wrong: Rust's aliasing XOR mutability rule (&T for shared immutable access, &mut T for exclusive mutable access) prohibits mutating state through shared references unless interior mutability patterns (e.g. RefCell, Mutex) are explicitly used.

Incorrect:

fn update_val(data: &i32) {
    // *data += 1; // ❌ Error E0594: cannot assign to `*data`, which is behind a `&` reference
}

Fix:

fn update_val(data: &mut i32) {
    *data += 1; // Correct: exclusive mutable reference permits mutation
}

Mistake 3: Concurrent Access to Iterator Adapters Across Threads Without Send / Sync Guards

The mistake: Sharing non-thread-safe Iterator Adapters instances across OS threads via std::thread::spawn.

Why it's wrong: Types that do not implement Send or Sync marker traits cannot safely cross thread boundaries. The compiler prevents data races by raising compile errors E0277 (trait Send is not implemented).

Incorrect:

use std::rc::Rc;
use std::thread;

let rc = Rc::new(42);
// thread::spawn(move || { println!("{}", rc); }); // ❌ Error E0277: `Rc` cannot be sent between threads safely

Fix:

use std::sync::Arc;
use std::thread;

let arc = Arc::new(42);
thread::spawn(move || {
    println!("{}", arc); // Correct: `Arc` implements `Send` and `Sync`
});

5. Practice Exercises

Exercise 1: Industrial IoT Telemetry Filter & Moving Metrics Pipeline

Scenario: An industrial IoT gateway collects raw sensor telemetry records from edge hardware nodes. The incoming record stream contains invalid sensor readings, corrupted flags, and out-of-range temperature readings.

Write a function process_telemetry_stream that accepts an iterator of RawTelemetry records and constructs a lazy iterator adapter pipeline to:

1. Filter out invalid items (where is_valid is false or sensor_id == 0).
2. Filter out out-of-bounds temperature readings (temperatures outside [-40.0, 125.0] Celsius).
3. Transform temperature values from Celsius to Fahrenheit ($F = C \times 1.8 + 32.0$).
4. Use .scan() to compute a cumulative running average temperature across all accepted readings.
5. Use .enumerate() to assign sequential zero-indexed batch execution IDs.
6. Map elements into ProcessedTelemetry structs containing sequence number, sensor ID, Fahrenheit temperature, running average, and alert status (AlertLevel::Normal for F < 100.0, AlertLevel::Warning for 100.0 \le F < 180.0, and AlertLevel::Critical for $F \ge 180.0$).

#[derive(Debug, Clone, PartialEq)]
pub struct RawTelemetry {
    pub sensor_id: u32,
    pub temp_celsius: f64,
    pub is_valid: bool,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AlertLevel {
    Normal,
    Warning,
    Critical,
}

#[derive(Debug, Clone, PartialEq)]
pub struct ProcessedTelemetry {
    pub seq_num: usize,
    pub sensor_id: u32,
    pub temp_fahrenheit: f64,
    pub running_avg_f: f64,
    pub alert: AlertLevel,
}

pub fn process_telemetry_stream<I>(records: I) -> Vec<ProcessedTelemetry>
where
    I: IntoIterator<Item = RawTelemetry>,
{
    records
        .into_iter()
        .filter(|r| r.is_valid && r.sensor_id > 0)
        .filter(|r| r.temp_celsius >= -40.0 && r.temp_celsius <= 125.0)
        .map(|r| {
            let temp_f = r.temp_celsius * 1.8 + 32.0;
            (r.sensor_id, temp_f)
        })
        .scan((0usize, 0.0f64), |(count, total_sum), (sensor_id, temp_f)| {
            *count += 1;
            *total_sum += temp_f;
            let avg = *total_sum / (*count as f64);
            Some((sensor_id, temp_f, avg))
        })
        .enumerate()
        .map(|(idx, (sensor_id, temp_f, running_avg))| {
            let alert = if temp_f >= 180.0 {
                AlertLevel::Critical
            } else if temp_f >= 100.0 {
                AlertLevel::Warning
            } else {
                AlertLevel::Normal
            };
            ProcessedTelemetry {
                seq_num: idx,
                sensor_id,
                temp_fahrenheit: temp_f,
                running_avg_f: running_avg,
                alert,
            }
        })
        .collect()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_process_telemetry_stream() {
        let raw_data = vec![
            RawTelemetry { sensor_id: 101, temp_celsius: 25.0, is_valid: true },
            RawTelemetry { sensor_id: 0, temp_celsius: 30.0, is_valid: true }, // Invalid sensor_id
            RawTelemetry { sensor_id: 102, temp_celsius: 45.0, is_valid: true },
            RawTelemetry { sensor_id: 103, temp_celsius: 150.0, is_valid: true }, // Out of bounds
            RawTelemetry { sensor_id: 104, temp_celsius: 90.0, is_valid: false }, // Invalid flag
            RawTelemetry { sensor_id: 105, temp_celsius: 90.0, is_valid: true },
        ];

        let results = process_telemetry_stream(raw_data);

        // 1. Explicit assertion with assert_eq!
        assert_eq!(results.len(), 3);
        assert_eq!(results[0].seq_num, 0);
        assert_eq!(results[1].sensor_id, 102);

        // 2. Explicit assertion with assert!
        assert!(results[0].temp_fahrenheit < 100.0);
        assert!(results[1].running_avg_f > 80.0 && results[1].running_avg_f < 100.0);

        // 3. Explicit assertion with assert_ne!
        assert_ne!(results[0].sensor_id, 0);
        assert_ne!(results[0].temp_fahrenheit, results[1].temp_fahrenheit);

        // 4. Explicit assertion with matches!
        assert!(matches!(results[0].alert, AlertLevel::Normal));
        assert!(matches!(results[1].alert, AlertLevel::Warning));
        assert!(matches!(results[2].alert, AlertLevel::Critical));
    }
}

Exercise 2: Financial Order Book Log Auditor & Cumulative Volume Pipeline

Scenario: A financial matching engine produces raw textual execution logs formatted as "TIMESTAMP|ORDER_ID|SIDE|PRICE|QUANTITY|STATUS". Before emitting trades to downstream clearing houses, an auditing pipeline must parse string records, filter out invalid or canceled orders, exclude low-volume transactions below a configurable threshold, track running volume totals, and emit structured AuditedTrade objects.

Write a function audit_trade_stream accepting an iterator of string slices and a minimum total volume threshold (min_volume_threshold: f64) that applies iterator adapters (.filter_map(), .filter(), .scan(), .enumerate(), .map()) to produce a Vec<AuditedTrade>.

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Side {
    Buy,
    Sell,
}

#[derive(Debug, Clone, PartialEq)]
pub struct TradeRecord {
    pub order_id: u64,
    pub side: Side,
    pub price: f64,
    pub quantity: f64,
}

#[derive(Debug, Clone, PartialEq)]
pub struct AuditedTrade {
    pub audit_id: usize,
    pub order_id: u64,
    pub side: Side,
    pub trade_value: f64,
    pub cumulative_volume: f64,
}

pub fn audit_trade_stream<'a, I>(logs: I, min_volume_threshold: f64) -> Vec<AuditedTrade>
where
    I: IntoIterator<Item = &'a str>,
{
    logs.into_iter()
        .filter_map(|line| {
            let parts: Vec<&str> = line.trim().split('|').collect();
            if parts.len() != 6 {
                return None;
            }
            let _timestamp = parts[0];
            let order_id: u64 = parts[1].parse().ok()?;
            let side = match parts[2] {
                "BUY" => Side::Buy,
                "SELL" => Side::Sell,
                _ => return None,
            };
            let price: f64 = parts[3].parse().ok()?;
            let quantity: f64 = parts[4].parse().ok()?;
            let status = parts[5];

            if status == "CANCELLED" || price <= 0.0 || quantity <= 0.0 {
                return None;
            }

            Some(TradeRecord {
                order_id,
                side,
                price,
                quantity,
            })
        })
        .filter(|trade| (trade.price * trade.quantity) >= min_volume_threshold)
        .scan(0.0f64, |cum_vol, trade| {
            let trade_value = trade.price * trade.quantity;
            *cum_vol += trade_value;
            Some((trade, trade_value, *cum_vol))
        })
        .enumerate()
        .map(|(idx, (trade, trade_value, cumulative_volume))| AuditedTrade {
            audit_id: idx + 1,
            order_id: trade.order_id,
            side: trade.side,
            trade_value,
            cumulative_volume,
        })
        .collect()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_audit_trade_stream() {
        let logs = vec![
            "1620000000|1001|BUY|150.50|10.0|FILLED",
            "1620000001|1002|SELL|200.00|2.0|CANCELLED",
            "1620000002|INVALID_PIPE_RECORD",
            "1620000003|1003|SELL|50.00|5.0|FILLED", // Volume = 250.0 (Below 1000.0)
            "1620000004|1004|SELL|300.00|5.0|FILLED", // Volume = 1500.0
        ];

        let audited = audit_trade_stream(logs, 1000.0);

        // 1. Explicit assertion with assert_eq!
        assert_eq!(audited.len(), 2);
        assert_eq!(audited[0].audit_id, 1);
        assert_eq!(audited[1].audit_id, 2);
        assert_eq!(audited[0].order_id, 1001);

        // 2. Explicit assertion with assert!
        assert!(audited[0].trade_value >= 1000.0);
        assert!(audited[1].cumulative_volume > audited[0].cumulative_volume);

        // 3. Explicit assertion with assert_ne!
        assert_ne!(audited[0].order_id, audited[1].order_id);
        assert_ne!(audited[0].side, audited[1].side);

        // 4. Explicit assertion with matches!
        assert!(matches!(audited[0].side, Side::Buy));
        assert!(matches!(audited[1].side, Side::Sell));
    }
}

Exercise 3: Network Packet Frame Decoder & Cryptographic XOR Processor

Scenario: An embedded packet inspection module receives raw network packet frames from a buffer interface. Each frame contains a byte flags header, a frame type indicator byte, and a payload byte vector.

Write a function process_network_frames that takes an iterator of RawFrame structs and a 1-byte XOR decryption key, applying iterator adapters to:

1. Filter out frames missing mandatory flag bitmasks (FLAG_VALID = 0x01 and FLAG_ENCRYPTED = 0x04).
2. Filter out empty payload buffers or oversized buffers ($> 1024$ bytes).
3. Decrypt the payload byte slice using an inner iterator .map() that XORs each byte with xor_key.
4. Fold over the decrypted payload bytes to calculate a wrapping 16-bit checksum (u16).
5. Enumerate and transform frames into DecryptedFrame structs containing frame_id, FrameKind variant (Control if payload_type == 0x01, Data otherwise), decrypted_payload, and checksum.

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FrameKind {
    Control,
    Data,
}

#[derive(Debug, Clone, PartialEq)]
pub struct RawFrame<'a> {
    pub flags: u8,
    pub payload_type: u8,
    pub payload: &'a [u8],
}

#[derive(Debug, Clone, PartialEq)]
pub struct DecryptedFrame {
    pub frame_id: usize,
    pub kind: FrameKind,
    pub decrypted_payload: Vec<u8>,
    pub checksum: u16,
}

pub const FLAG_VALID: u8 = 0x01;
pub const FLAG_ENCRYPTED: u8 = 0x04;

pub fn process_network_frames<'a, I>(frames: I, xor_key: u8) -> Vec<DecryptedFrame>
where
    I: IntoIterator<Item = RawFrame<'a>>,
{
    frames
        .into_iter()
        .filter(|frame| (frame.flags & FLAG_VALID != 0) && (frame.flags & FLAG_ENCRYPTED != 0))
        .filter(|frame| !frame.payload.is_empty() && frame.payload.len() <= 1024)
        .map(|frame| {
            let decrypted_payload: Vec<u8> = frame
                .payload
                .iter()
                .map(|&byte| byte ^ xor_key)
                .collect();

            let checksum = decrypted_payload
                .iter()
                .fold(0u16, |acc, &b| acc.wrapping_add(b as u16));

            let kind = if frame.payload_type == 0x01 {
                FrameKind::Control
            } else {
                FrameKind::Data
            };

            (kind, decrypted_payload, checksum)
        })
        .enumerate()
        .map(|(idx, (kind, decrypted_payload, checksum))| DecryptedFrame {
            frame_id: idx + 1,
            kind,
            decrypted_payload,
            checksum,
        })
        .collect()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_process_network_frames() {
        let key = 0xAA;
        let payload1 = vec![0xDE ^ key, 0xAD ^ key];
        let payload2 = vec![0xBE ^ key, 0xEF ^ key];

        let frames = vec![
            RawFrame {
                flags: FLAG_VALID | FLAG_ENCRYPTED,
                payload_type: 0x01,
                payload: &payload1,
            },
            RawFrame {
                flags: FLAG_VALID, // Lacks FLAG_ENCRYPTED
                payload_type: 0x02,
                payload: &[0x11, 0x22],
            },
            RawFrame {
                flags: FLAG_VALID | FLAG_ENCRYPTED,
                payload_type: 0x02,
                payload: &payload2,
            },
        ];

        let processed = process_network_frames(frames, key);

        // 1. Explicit assertion with assert_eq!
        assert_eq!(processed.len(), 2);
        assert_eq!(processed[0].frame_id, 1);
        assert_eq!(processed[0].decrypted_payload, vec![0xDE, 0xAD]);
        assert_eq!(processed[1].decrypted_payload, vec![0xBE, 0xEF]);

        // 2. Explicit assertion with assert!
        assert!(processed[0].checksum > 0);
        assert!(processed[1].checksum > processed[0].checksum);

        // 3. Explicit assertion with assert_ne!
        assert_ne!(processed[0].frame_id, processed[1].frame_id);
        assert_ne!(processed[0].kind, processed[1].kind);

        // 4. Explicit assertion with matches!
        assert!(matches!(processed[0].kind, FrameKind::Control));
        assert!(matches!(processed[1].kind, FrameKind::Data));
    }
}

6. Related Terms

• Collecting — The terminal operation that forces the lazy adapters to finally execute and save their work.
• Iterator — The prerequisite lazy sequence that adapters attach to.
• Rayon — Related concept: Rayon.

7. Key Takeaways

• Iterator Adapters (map, filter, enumerate, zip, etc.) are methods that modify the sequence of an Iterator.
• They are lazy and return a new Iterator. They do not execute any work until they are consumed.
• They rely heavily on Closures (inline anonymous functions like |x| x + 1) to dictate their specific behavior.
• You can chain multiple adapters together to build complex data transformation pipelines with zero runtime overhead.