Struct ArcConsumer 

pub struct ArcConsumer<T> { /* private fields */ }

ArcConsumer struct

Non-mutating consumer implementation based on Arc<dyn Fn(&T) + Send + Sync>, for thread-safe shared ownership scenarios. The wrapper does not need Mutex because it only invokes a shared Fn.

Features

• Shared Ownership: Cloneable through Arc, allows multiple owners
• Thread Safe: Implements Send + Sync, can be safely used concurrently
• Lock-free Wrapper: No Mutex protection needed by the wrapper
• Non-consuming API: and_then borrows &self, original object remains usable

Use Cases

Choose ArcConsumer when:

• Need to share non-mutating consumer across multiple threads
• Pure observation operations, such as logging, monitoring, notifications
• Need high-concurrency reads with no lock overhead

Performance Advantages

Compared to ArcStatefulConsumer, ArcConsumer has no Mutex lock overhead, performing better in high-concurrency observation scenarios.

Examples

use qubit_function::{Consumer, ArcConsumer};

let consumer = ArcConsumer::new(|x: &i32| {
    println!("Observed: {}", x);
});
let clone = consumer.clone();

consumer.accept(&5);
clone.accept(&10);

Implementations

impl<T> ArcConsumer<T>

pub fn new<F>(f: F) -> Self

where F: Fn(&T) + Send + Sync + 'static,

Creates a new consumer.

Wraps the provided closure in the appropriate smart pointer type for this consumer implementation.

Examples found in repository

examples/consumers/consumer_fn_usage.rs (lines 47-51)

fn main() {
    println!("=== Consumer into_fn/to_fn Usage Examples ===\n");

    // Example 1: Using BoxConsumer::into_fn to pass to standard library's map
    println!("1. BoxConsumer::into_fn used with Iterator::for_each");
    let log = Arc::new(Mutex::new(Vec::new()));
    let l = log.clone();
    let consumer = BoxConsumer::new(move |x: &i32| {
        l.lock().expect("mutex should not be poisoned").push(*x * 2);
    });

    // Convert consumer to closure and pass to for_each
    [1, 2, 3, 4, 5].iter().for_each(consumer.into_fn());
    println!(
        "   Result: {:?}\n",
        *log.lock().expect("mutex should not be poisoned")
    );

    // Example 2: Using ArcConsumer::to_fn can be used multiple times
    println!("2. ArcConsumer::to_fn can be used multiple times");
    let log2 = Arc::new(Mutex::new(Vec::new()));
    let l2 = log2.clone();
    let consumer2 = ArcConsumer::new(move |x: &i32| {
        l2.lock()
            .expect("mutex should not be poisoned")
            .push(*x + 10);
    });

    // to_fn doesn't consume consumer, can be called multiple times
    [1, 2, 3].iter().for_each(consumer2.to_fn());
    println!(
        "   First time: {:?}",
        *log2.lock().expect("mutex should not be poisoned")
    );

    [4, 5].iter().for_each(consumer2.to_fn());
    println!(
        "   Second time: {:?}\n",
        *log2.lock().expect("mutex should not be poisoned")
    );

    // Example 3: Using RcConsumer::to_fn
    println!("3. RcConsumer::to_fn used for single-threaded scenarios");
    let log3 = Rc::new(RefCell::new(Vec::new()));
    let l3 = log3.clone();
    let consumer3 = RcConsumer::new(move |x: &i32| {
        l3.borrow_mut().push(*x * 3);
    });

    [1, 2, 3, 4].iter().for_each(consumer3.to_fn());
    println!("   Result: {:?}\n", *log3.borrow());

    // Example 4: Using in custom functions
    println!("4. Using in custom functions");
    fn process_items<F>(items: Vec<i32>, consumer: F)
    where
        F: FnMut(&i32),
    {
        items.iter().for_each(consumer);
    }

    let log4 = Arc::new(Mutex::new(Vec::new()));
    let l4 = log4.clone();
    let consumer4 = BoxConsumer::new(move |x: &i32| {
        l4.lock()
            .expect("mutex should not be poisoned")
            .push(*x * 5);
    });

    // Use into_fn to convert Consumer to closure and pass to function
    process_items(vec![1, 2, 3], consumer4.into_fn());
    println!(
        "   Result: {:?}\n",
        *log4.lock().expect("mutex should not be poisoned")
    );

    // Example 5: Using into_fn after chained operations
    println!("5. Using into_fn after chained operations");
    let log5 = Arc::new(Mutex::new(Vec::new()));
    let l5 = log5.clone();
    let l6 = log5.clone();

    let chained = BoxConsumer::new(move |x: &i32| {
        l5.lock()
            .expect("mutex should not be poisoned")
            .push(format!("A: {}", x));
    })
    .and_then(move |x: &i32| {
        l6.lock()
            .expect("mutex should not be poisoned")
            .push(format!("B: {}", x));
    });

    [1, 2].iter().for_each(chained.into_fn());
    println!(
        "   Result: {:?}\n",
        *log5.lock().expect("mutex should not be poisoned")
    );

    println!("=== Demo Complete ===");
}

examples/consumers/consumer_demo.rs (line 175)

fn main() {
    println!("=== Consumer Examples ===\n");
    println!("Note: Consumer only reads values, does not modify the original value");
    println!("If you need to modify values, please refer to mutator_demo.rs\n");

    // ========================================================================
    // Example 1: BoxConsumer basic usage
    // ========================================================================
    println!("Example 1: BoxConsumer basic usage");
    println!("{}", "-".repeat(50));

    let consumer = BoxConsumer::new(|x: &i32| {
        println!("Read and calculate: {} * 2 = {}", x, x * 2);
    });
    let value = 5;
    println!("Value: {}", value);
    consumer.accept(&value);
    println!("Value remains: {} (not modified)\n", value);

    // ========================================================================
    // Example 2: BoxConsumer method chaining
    // ========================================================================
    println!("Example 2: BoxConsumer method chaining");
    println!("{}", "-".repeat(50));

    let results = Arc::new(Mutex::new(Vec::new()));
    let r1 = results.clone();
    let r2 = results.clone();
    let r3 = results.clone();

    let chained = BoxConsumer::new(move |x: &i32| {
        r1.lock()
            .expect("mutex should not be poisoned")
            .push(*x * 2);
    })
    .and_then(move |x: &i32| {
        r2.lock()
            .expect("mutex should not be poisoned")
            .push(*x + 10);
    })
    .and_then(move |x: &i32| {
        r3.lock().expect("mutex should not be poisoned").push(*x);
        println!("Processing value: {}", x);
    });

    let value = 5;
    println!("Initial value: {}", value);
    chained.accept(&value);
    println!(
        "Collected results: {:?}",
        *results.lock().expect("mutex should not be poisoned")
    );
    println!("Original value: {} (not modified)\n", value);

    // ========================================================================
    // Example 3: Closure extension methods
    // ========================================================================
    println!("Example 3: Direct use of extension methods on closures");
    println!("{}", "-".repeat(50));

    let result = Arc::new(Mutex::new(0));
    let r1 = result.clone();
    let r2 = result.clone();

    let closure_chain = (move |x: &i32| {
        *r1.lock().expect("mutex should not be poisoned") = *x * 2;
    })
    .and_then(move |_x: &i32| {
        *r2.lock().expect("mutex should not be poisoned") += 10;
    });

    let value = 5;
    println!("Initial value: {}", value);
    closure_chain.accept(&value);
    println!(
        "Calculation result: {}",
        *result.lock().expect("mutex should not be poisoned")
    );
    println!("Original value: {} (not modified)\n", value);

    // ========================================================================
    // Example 4: BoxConsumer factory methods
    // ========================================================================
    println!("Example 4: BoxConsumer factory methods");
    println!("{}", "-".repeat(50));

    // noop
    println!("noop - does nothing:");
    let noop = BoxConsumer::<i32>::noop();
    let value = 42;
    noop.accept(&value);
    println!("Value: {}\n", value);

    // print
    print!("print - prints value: ");
    let print = BoxConsumer::new(|x: &i32| println!("{}", x));
    let value = 42;
    print.accept(&value);
    println!();

    // print with prefix
    let print_with = BoxConsumer::new(|x: &i32| println!("Value is: {}", x));
    let value = 42;
    print_with.accept(&value);
    println!();

    // ========================================================================
    // Example 5: Conditional Consumer
    // ========================================================================
    println!("Example 5: Conditional Consumer");
    println!("{}", "-".repeat(50));

    // when
    let mut check_positive =
        BoxStatefulConsumer::new(|x: &i32| println!("Positive: {}", x)).when(|x: &i32| *x > 0);

    let positive = 5;
    let negative = -5;
    print!("Check {}: ", positive);
    check_positive.accept(&positive);
    print!("Check {}: ", negative);
    check_positive.accept(&negative);
    println!("(negative numbers not printed)\n");

    // when().or_else()
    let mut categorize = BoxStatefulConsumer::new(|x: &i32| println!("Positive: {}", x))
        .when(|x: &i32| *x > 0)
        .or_else(|x: &i32| println!("Non-positive: {}", x));

    let positive = 10;
    let negative = -10;
    categorize.accept(&positive);
    categorize.accept(&negative);
    println!();

    // ========================================================================
    // Example 6: ArcConsumer - multi-threaded sharing
    // ========================================================================
    println!("Example 6: ArcConsumer - multi-threaded sharing");
    println!("{}", "-".repeat(50));

    let shared = ArcConsumer::new(|x: &i32| println!("Processing value: {}", x * 2));

    // Clone for another thread
    let shared_clone = shared.clone();
    let handle = thread::spawn(move || {
        let value = 5;
        let consumer = shared_clone;
        consumer.accept(&value);
        value
    });

    // Use in main thread
    let value = 3;
    let consumer = shared;
    consumer.accept(&value);

    let thread_result = handle.join().expect("thread should not panic");
    println!("Thread result: {}\n", thread_result);

    // ========================================================================
    // Example 7: ArcConsumer composition (does not consume original consumer)
    // ========================================================================
    println!("Example 7: ArcConsumer composition (borrowing &self)");
    println!("{}", "-".repeat(50));

    let double = ArcConsumer::new(|x: &i32| println!("double: {}", x * 2));
    let add_ten = ArcConsumer::new(|x: &i32| println!("add_ten: {}", x + 10));

    // Composition does not consume original consumer
    let pipeline1 = double.and_then(add_ten.clone());
    let pipeline2 = add_ten.and_then(double.clone());

    let value1 = 5;
    let p1 = pipeline1;
    print!("pipeline1 processing 5: ");
    p1.accept(&value1);

    let value2 = 5;
    let p2 = pipeline2;
    print!("pipeline2 processing 5: ");
    p2.accept(&value2);

    // double and add_ten are still available
    let value3 = 10;
    let d = double;
    print!("Original double still available, processing 10: ");
    d.accept(&value3);
    println!();

    // ========================================================================
    // Example 8: RcConsumer - single-threaded sharing
    // ========================================================================
    println!("Example 8: RcConsumer - single-threaded sharing");
    println!("{}", "-".repeat(50));

    let rc_consumer = RcConsumer::new(|x: &i32| println!("Processing: {}", x * 2));

    // Clone multiple copies
    let clone1 = rc_consumer.clone();
    let clone2 = rc_consumer.clone();

    let value1 = 5;
    let c1 = clone1;
    print!("clone1 processing 5: ");
    c1.accept(&value1);

    let value2 = 3;
    let c2 = clone2;
    print!("clone2 processing 3: ");
    c2.accept(&value2);

    let value3 = 7;
    let c3 = rc_consumer;
    print!("Original processing 7: ");
    c3.accept(&value3);
    println!();

    // ========================================================================
    // Example 9: RcConsumer composition (borrowing &self)
    // ========================================================================
    println!("Example 9: RcConsumer composition (borrowing &self)");
    println!("{}", "-".repeat(50));

    let double = RcConsumer::new(|x: &i32| println!("double: {}", x * 2));
    let add_ten = RcConsumer::new(|x: &i32| println!("add_ten: {}", x + 10));

    let pipeline1 = double.and_then(add_ten.clone());
    let pipeline2 = add_ten.and_then(double.clone());

    let value1 = 5;
    let p1 = pipeline1;
    print!("pipeline1 processing 5: ");
    p1.accept(&value1);

    let value2 = 5;
    let p2 = pipeline2;
    print!("pipeline2 processing 5: ");
    p2.accept(&value2);
    println!();

    // ========================================================================
    // Example 10: Unified Consumer trait
    // ========================================================================
    println!("Example 10: Unified Consumer trait");
    println!("{}", "-".repeat(50));

    fn log_all<C: Consumer<i32>>(consumer: &mut C, values: &[i32]) {
        for value in values.iter() {
            consumer.accept(value);
        }
    }

    let values = vec![1, 2, 3, 4, 5];

    let mut box_con = BoxConsumer::new(|x: &i32| print!("{} ", x * 2));
    print!("BoxConsumer processing {:?}: ", values);
    log_all(&mut box_con, &values);
    println!();

    let mut arc_con = ArcConsumer::new(|x: &i32| print!("{} ", x * 2));
    print!("ArcConsumer processing {:?}: ", values);
    log_all(&mut arc_con, &values);
    println!();

    let mut rc_con = RcConsumer::new(|x: &i32| print!("{} ", x * 2));
    print!("RcConsumer processing {:?}: ", values);
    log_all(&mut rc_con, &values);
    println!();

    let mut closure = |x: &i32| print!("{} ", x * 2);
    print!("Closure processing {:?}: ", values);
    log_all(&mut closure, &values);
    println!("\n");

    // ========================================================================
    // Example 11: Data validation and logging
    // ========================================================================
    println!("Example 11: Data validation and logging");
    println!("{}", "-".repeat(50));

    let validator = BoxConsumer::new(|x: &i32| {
        let status = if *x >= 0 && *x <= 100 {
            "valid"
        } else {
            "out of range"
        };
        println!("Validate {}: {}", x, status);
    });

    let logger = BoxConsumer::new(|x: &i32| {
        println!("Log to file: value={}, square={}", x, x * x);
    });

    let pipeline = validator.and_then(logger);

    let test_values = vec![-50, 30, 200];
    for value in test_values {
        pipeline.accept(&value);
    }
    println!();

    // ========================================================================
    // Example 12: String analysis
    // ========================================================================
    println!("Example 12: String analysis");
    println!("{}", "-".repeat(50));

    let string_analyzer = BoxConsumer::new(|s: &String| {
        println!("Length: {}", s.len());
    })
    .and_then(|s: &String| {
        println!("Lowercase: {}", s.to_lowercase());
    })
    .and_then(|s: &String| {
        println!("Uppercase: {}", s.to_uppercase());
    })
    .and_then(|s: &String| {
        let word_count = s.split_whitespace().count();
        println!("Word count: {}", word_count);
    });

    let text = String::from("Hello World");
    println!("Analyzing text: \"{}\"", text);
    string_analyzer.accept(&text);
    println!("Original text: \"{}\" (not modified)\n", text);

    // ========================================================================
    // Example 13: Type conversion
    // ========================================================================
    println!("Example 13: Type conversion");
    println!("{}", "-".repeat(50));

    // Closure -> BoxConsumer
    let closure = |x: &i32| print!("Processing: {} ", x * 2);
    let box_con = Consumer::into_box(closure);
    let value = 5;
    print!("Closure -> BoxConsumer: ");
    box_con.accept(&value);
    println!();

    // Closure -> RcConsumer
    let closure = |x: &i32| print!("Processing: {} ", x * 2);
    let rc_con = Consumer::into_rc(closure);
    let value = 5;
    print!("Closure -> RcConsumer: ");
    rc_con.accept(&value);
    println!();

    // Closure -> ArcConsumer
    let closure = |x: &i32| print!("Processing: {} ", x * 2);
    let arc_con = Consumer::into_arc(closure);
    let value = 5;
    print!("Closure -> ArcConsumer: ");
    arc_con.accept(&value);
    println!();

    // BoxConsumer -> RcConsumer
    let box_con = BoxConsumer::new(|x: &i32| print!("Processing: {} ", x * 2));
    let rc_con = box_con.into_rc();
    let value = 5;
    print!("BoxConsumer -> RcConsumer: ");
    rc_con.accept(&value);
    println!();

    // RcConsumer -> BoxConsumer
    let rc_con = RcConsumer::new(|x: &i32| print!("Processing: {} ", x * 2));
    let box_con = rc_con.into_box();
    let value = 5;
    print!("RcConsumer -> BoxConsumer: ");
    box_con.accept(&value);
    println!("\n");

    // ========================================================================
    // Example 14: Custom types
    // ========================================================================
    println!("Example 14: Custom types");
    println!("{}", "-".repeat(50));

    #[derive(Debug, Clone)]
    struct Point {
        x: i32,
        y: i32,
    }

    let analyzer = BoxConsumer::new(|p: &Point| {
        println!("Point coordinates: ({}, {})", p.x, p.y);
    })
    .and_then(|p: &Point| {
        let distance = ((p.x * p.x + p.y * p.y) as f64).sqrt();
        println!("Distance from origin: {:.2}", distance);
    })
    .and_then(|p: &Point| {
        let quadrant = match (p.x >= 0, p.y >= 0) {
            (true, true) => "First quadrant",
            (false, true) => "Second quadrant",
            (false, false) => "Third quadrant",
            (true, false) => "Fourth quadrant",
        };
        println!("Quadrant: {}", quadrant);
    });

    let point = Point { x: 3, y: 4 };
    println!("Analyzing point: {:?}", point);
    analyzer.accept(&point);
    println!("Original point: {:?} (not modified)\n", point);

    // ========================================================================
    // Example 15: Data collection and statistics
    // ========================================================================
    println!("Example 15: Data collection and statistics");
    println!("{}", "-".repeat(50));

    let sum = Arc::new(Mutex::new(0));
    let count = Arc::new(Mutex::new(0));
    let sum_clone = sum.clone();
    let count_clone = count.clone();

    let collector = BoxConsumer::new(move |x: &i32| {
        *sum_clone.lock().expect("mutex should not be poisoned") += *x;
        *count_clone.lock().expect("mutex should not be poisoned") += 1;
    });

    let numbers = vec![10, 20, 30, 40, 50];
    println!("Numbers: {:?}", numbers);
    for num in &numbers {
        collector.accept(num);
    }

    let total = *sum.lock().expect("mutex should not be poisoned");
    let cnt = *count.lock().expect("mutex should not be poisoned");
    println!("Sum: {}", total);
    println!("Count: {}", cnt);
    println!("Average: {:.2}\n", total as f64 / cnt as f64);

    println!("=== All examples completed ===");
    println!("\nTip: For value modification functionality, please refer to mutator_demo.rs");
}

pub fn new_with_name<F>(name: &str, f: F) -> Self

where F: Fn(&T) + Send + Sync + 'static,

Creates a new named consumer.

Wraps the provided closure and assigns it a name, which is useful for debugging and logging purposes.

pub fn new_with_optional_name<F>(f: F, name: Option<String>) -> Self

where F: Fn(&T) + Send + Sync + 'static,

Creates a new named consumer with an optional name.

Wraps the provided closure and assigns it an optional name.

pub fn name(&self) -> Option<&str>

Gets the name of this consumer.

Returns

Returns Some(&str) if a name was set, None otherwise.

pub fn set_name(&mut self, name: &str)

Sets the name of this consumer.

Parameters

• name - The name to set for this consumer

pub fn clear_name(&mut self)

Clears the name of this consumer.

pub fn noop() -> Self

Creates a no-operation consumer.

Creates a consumer that does nothing when called. Useful for default values or placeholder implementations.

Returns

Returns a new consumer instance that performs no operation.

pub fn when<P>(&self, predicate: P) -> ArcConditionalConsumer<T>

where T: 'static, P: Predicate<T> + Send + Sync + 'static,

Creates a conditional consumer

Wraps this consumer with a predicate condition, creating a new conditional consumer that will only execute the original consumer when the predicate evaluates to true.

Parameters

• predicate - The condition that must be satisfied for the consumer to execute

Returns

Returns a conditional consumer that executes this consumer only when the predicate is satisfied

Examples

use qubit_function::{ArcConsumer, Consumer};
let consumer = ArcConsumer::new(|x: &i32| println!("{}", x));
let conditional = consumer.when(|x: &i32| *x > 0);

conditional.accept(&5);  // prints: 5
conditional.accept(&-5); // prints nothing

pub fn and_then<C>(&self, after: C) -> ArcConsumer<T>

where T: 'static, C: Consumer<T> + Send + Sync + 'static,

Chains another consumer in sequence

Combines this consumer with another consumer into a new consumer that executes both consumers in sequence. The returned consumer first executes this consumer, then unconditionally executes the after consumer.

Parameters

• after - The consumer to execute after this one (always executed)

Returns

Returns a new consumer that executes both consumers in sequence

Examples

use qubit_function::{ArcConsumer, Consumer};
let consumer1 = ArcConsumer::new(|x: &i32| print!("first: {}", x));
let consumer2 = ArcConsumer::new(|x: &i32| println!(" second: {}", x));

let chained = consumer1.and_then(consumer2);

chained.accept(&5);  // prints: first: 5 second: 5

Examples found in repository

examples/consumers/consumer_demo.rs (line 204)

fn main() {
    println!("=== Consumer Examples ===\n");
    println!("Note: Consumer only reads values, does not modify the original value");
    println!("If you need to modify values, please refer to mutator_demo.rs\n");

    // ========================================================================
    // Example 1: BoxConsumer basic usage
    // ========================================================================
    println!("Example 1: BoxConsumer basic usage");
    println!("{}", "-".repeat(50));

    let consumer = BoxConsumer::new(|x: &i32| {
        println!("Read and calculate: {} * 2 = {}", x, x * 2);
    });
    let value = 5;
    println!("Value: {}", value);
    consumer.accept(&value);
    println!("Value remains: {} (not modified)\n", value);

    // ========================================================================
    // Example 2: BoxConsumer method chaining
    // ========================================================================
    println!("Example 2: BoxConsumer method chaining");
    println!("{}", "-".repeat(50));

    let results = Arc::new(Mutex::new(Vec::new()));
    let r1 = results.clone();
    let r2 = results.clone();
    let r3 = results.clone();

    let chained = BoxConsumer::new(move |x: &i32| {
        r1.lock()
            .expect("mutex should not be poisoned")
            .push(*x * 2);
    })
    .and_then(move |x: &i32| {
        r2.lock()
            .expect("mutex should not be poisoned")
            .push(*x + 10);
    })
    .and_then(move |x: &i32| {
        r3.lock().expect("mutex should not be poisoned").push(*x);
        println!("Processing value: {}", x);
    });

    let value = 5;
    println!("Initial value: {}", value);
    chained.accept(&value);
    println!(
        "Collected results: {:?}",
        *results.lock().expect("mutex should not be poisoned")
    );
    println!("Original value: {} (not modified)\n", value);

    // ========================================================================
    // Example 3: Closure extension methods
    // ========================================================================
    println!("Example 3: Direct use of extension methods on closures");
    println!("{}", "-".repeat(50));

    let result = Arc::new(Mutex::new(0));
    let r1 = result.clone();
    let r2 = result.clone();

    let closure_chain = (move |x: &i32| {
        *r1.lock().expect("mutex should not be poisoned") = *x * 2;
    })
    .and_then(move |_x: &i32| {
        *r2.lock().expect("mutex should not be poisoned") += 10;
    });

    let value = 5;
    println!("Initial value: {}", value);
    closure_chain.accept(&value);
    println!(
        "Calculation result: {}",
        *result.lock().expect("mutex should not be poisoned")
    );
    println!("Original value: {} (not modified)\n", value);

    // ========================================================================
    // Example 4: BoxConsumer factory methods
    // ========================================================================
    println!("Example 4: BoxConsumer factory methods");
    println!("{}", "-".repeat(50));

    // noop
    println!("noop - does nothing:");
    let noop = BoxConsumer::<i32>::noop();
    let value = 42;
    noop.accept(&value);
    println!("Value: {}\n", value);

    // print
    print!("print - prints value: ");
    let print = BoxConsumer::new(|x: &i32| println!("{}", x));
    let value = 42;
    print.accept(&value);
    println!();

    // print with prefix
    let print_with = BoxConsumer::new(|x: &i32| println!("Value is: {}", x));
    let value = 42;
    print_with.accept(&value);
    println!();

    // ========================================================================
    // Example 5: Conditional Consumer
    // ========================================================================
    println!("Example 5: Conditional Consumer");
    println!("{}", "-".repeat(50));

    // when
    let mut check_positive =
        BoxStatefulConsumer::new(|x: &i32| println!("Positive: {}", x)).when(|x: &i32| *x > 0);

    let positive = 5;
    let negative = -5;
    print!("Check {}: ", positive);
    check_positive.accept(&positive);
    print!("Check {}: ", negative);
    check_positive.accept(&negative);
    println!("(negative numbers not printed)\n");

    // when().or_else()
    let mut categorize = BoxStatefulConsumer::new(|x: &i32| println!("Positive: {}", x))
        .when(|x: &i32| *x > 0)
        .or_else(|x: &i32| println!("Non-positive: {}", x));

    let positive = 10;
    let negative = -10;
    categorize.accept(&positive);
    categorize.accept(&negative);
    println!();

    // ========================================================================
    // Example 6: ArcConsumer - multi-threaded sharing
    // ========================================================================
    println!("Example 6: ArcConsumer - multi-threaded sharing");
    println!("{}", "-".repeat(50));

    let shared = ArcConsumer::new(|x: &i32| println!("Processing value: {}", x * 2));

    // Clone for another thread
    let shared_clone = shared.clone();
    let handle = thread::spawn(move || {
        let value = 5;
        let consumer = shared_clone;
        consumer.accept(&value);
        value
    });

    // Use in main thread
    let value = 3;
    let consumer = shared;
    consumer.accept(&value);

    let thread_result = handle.join().expect("thread should not panic");
    println!("Thread result: {}\n", thread_result);

    // ========================================================================
    // Example 7: ArcConsumer composition (does not consume original consumer)
    // ========================================================================
    println!("Example 7: ArcConsumer composition (borrowing &self)");
    println!("{}", "-".repeat(50));

    let double = ArcConsumer::new(|x: &i32| println!("double: {}", x * 2));
    let add_ten = ArcConsumer::new(|x: &i32| println!("add_ten: {}", x + 10));

    // Composition does not consume original consumer
    let pipeline1 = double.and_then(add_ten.clone());
    let pipeline2 = add_ten.and_then(double.clone());

    let value1 = 5;
    let p1 = pipeline1;
    print!("pipeline1 processing 5: ");
    p1.accept(&value1);

    let value2 = 5;
    let p2 = pipeline2;
    print!("pipeline2 processing 5: ");
    p2.accept(&value2);

    // double and add_ten are still available
    let value3 = 10;
    let d = double;
    print!("Original double still available, processing 10: ");
    d.accept(&value3);
    println!();

    // ========================================================================
    // Example 8: RcConsumer - single-threaded sharing
    // ========================================================================
    println!("Example 8: RcConsumer - single-threaded sharing");
    println!("{}", "-".repeat(50));

    let rc_consumer = RcConsumer::new(|x: &i32| println!("Processing: {}", x * 2));

    // Clone multiple copies
    let clone1 = rc_consumer.clone();
    let clone2 = rc_consumer.clone();

    let value1 = 5;
    let c1 = clone1;
    print!("clone1 processing 5: ");
    c1.accept(&value1);

    let value2 = 3;
    let c2 = clone2;
    print!("clone2 processing 3: ");
    c2.accept(&value2);

    let value3 = 7;
    let c3 = rc_consumer;
    print!("Original processing 7: ");
    c3.accept(&value3);
    println!();

    // ========================================================================
    // Example 9: RcConsumer composition (borrowing &self)
    // ========================================================================
    println!("Example 9: RcConsumer composition (borrowing &self)");
    println!("{}", "-".repeat(50));

    let double = RcConsumer::new(|x: &i32| println!("double: {}", x * 2));
    let add_ten = RcConsumer::new(|x: &i32| println!("add_ten: {}", x + 10));

    let pipeline1 = double.and_then(add_ten.clone());
    let pipeline2 = add_ten.and_then(double.clone());

    let value1 = 5;
    let p1 = pipeline1;
    print!("pipeline1 processing 5: ");
    p1.accept(&value1);

    let value2 = 5;
    let p2 = pipeline2;
    print!("pipeline2 processing 5: ");
    p2.accept(&value2);
    println!();

    // ========================================================================
    // Example 10: Unified Consumer trait
    // ========================================================================
    println!("Example 10: Unified Consumer trait");
    println!("{}", "-".repeat(50));

    fn log_all<C: Consumer<i32>>(consumer: &mut C, values: &[i32]) {
        for value in values.iter() {
            consumer.accept(value);
        }
    }

    let values = vec![1, 2, 3, 4, 5];

    let mut box_con = BoxConsumer::new(|x: &i32| print!("{} ", x * 2));
    print!("BoxConsumer processing {:?}: ", values);
    log_all(&mut box_con, &values);
    println!();

    let mut arc_con = ArcConsumer::new(|x: &i32| print!("{} ", x * 2));
    print!("ArcConsumer processing {:?}: ", values);
    log_all(&mut arc_con, &values);
    println!();

    let mut rc_con = RcConsumer::new(|x: &i32| print!("{} ", x * 2));
    print!("RcConsumer processing {:?}: ", values);
    log_all(&mut rc_con, &values);
    println!();

    let mut closure = |x: &i32| print!("{} ", x * 2);
    print!("Closure processing {:?}: ", values);
    log_all(&mut closure, &values);
    println!("\n");

    // ========================================================================
    // Example 11: Data validation and logging
    // ========================================================================
    println!("Example 11: Data validation and logging");
    println!("{}", "-".repeat(50));

    let validator = BoxConsumer::new(|x: &i32| {
        let status = if *x >= 0 && *x <= 100 {
            "valid"
        } else {
            "out of range"
        };
        println!("Validate {}: {}", x, status);
    });

    let logger = BoxConsumer::new(|x: &i32| {
        println!("Log to file: value={}, square={}", x, x * x);
    });

    let pipeline = validator.and_then(logger);

    let test_values = vec![-50, 30, 200];
    for value in test_values {
        pipeline.accept(&value);
    }
    println!();

    // ========================================================================
    // Example 12: String analysis
    // ========================================================================
    println!("Example 12: String analysis");
    println!("{}", "-".repeat(50));

    let string_analyzer = BoxConsumer::new(|s: &String| {
        println!("Length: {}", s.len());
    })
    .and_then(|s: &String| {
        println!("Lowercase: {}", s.to_lowercase());
    })
    .and_then(|s: &String| {
        println!("Uppercase: {}", s.to_uppercase());
    })
    .and_then(|s: &String| {
        let word_count = s.split_whitespace().count();
        println!("Word count: {}", word_count);
    });

    let text = String::from("Hello World");
    println!("Analyzing text: \"{}\"", text);
    string_analyzer.accept(&text);
    println!("Original text: \"{}\" (not modified)\n", text);

    // ========================================================================
    // Example 13: Type conversion
    // ========================================================================
    println!("Example 13: Type conversion");
    println!("{}", "-".repeat(50));

    // Closure -> BoxConsumer
    let closure = |x: &i32| print!("Processing: {} ", x * 2);
    let box_con = Consumer::into_box(closure);
    let value = 5;
    print!("Closure -> BoxConsumer: ");
    box_con.accept(&value);
    println!();

    // Closure -> RcConsumer
    let closure = |x: &i32| print!("Processing: {} ", x * 2);
    let rc_con = Consumer::into_rc(closure);
    let value = 5;
    print!("Closure -> RcConsumer: ");
    rc_con.accept(&value);
    println!();

    // Closure -> ArcConsumer
    let closure = |x: &i32| print!("Processing: {} ", x * 2);
    let arc_con = Consumer::into_arc(closure);
    let value = 5;
    print!("Closure -> ArcConsumer: ");
    arc_con.accept(&value);
    println!();

    // BoxConsumer -> RcConsumer
    let box_con = BoxConsumer::new(|x: &i32| print!("Processing: {} ", x * 2));
    let rc_con = box_con.into_rc();
    let value = 5;
    print!("BoxConsumer -> RcConsumer: ");
    rc_con.accept(&value);
    println!();

    // RcConsumer -> BoxConsumer
    let rc_con = RcConsumer::new(|x: &i32| print!("Processing: {} ", x * 2));
    let box_con = rc_con.into_box();
    let value = 5;
    print!("RcConsumer -> BoxConsumer: ");
    box_con.accept(&value);
    println!("\n");

    // ========================================================================
    // Example 14: Custom types
    // ========================================================================
    println!("Example 14: Custom types");
    println!("{}", "-".repeat(50));

    #[derive(Debug, Clone)]
    struct Point {
        x: i32,
        y: i32,
    }

    let analyzer = BoxConsumer::new(|p: &Point| {
        println!("Point coordinates: ({}, {})", p.x, p.y);
    })
    .and_then(|p: &Point| {
        let distance = ((p.x * p.x + p.y * p.y) as f64).sqrt();
        println!("Distance from origin: {:.2}", distance);
    })
    .and_then(|p: &Point| {
        let quadrant = match (p.x >= 0, p.y >= 0) {
            (true, true) => "First quadrant",
            (false, true) => "Second quadrant",
            (false, false) => "Third quadrant",
            (true, false) => "Fourth quadrant",
        };
        println!("Quadrant: {}", quadrant);
    });

    let point = Point { x: 3, y: 4 };
    println!("Analyzing point: {:?}", point);
    analyzer.accept(&point);
    println!("Original point: {:?} (not modified)\n", point);

    // ========================================================================
    // Example 15: Data collection and statistics
    // ========================================================================
    println!("Example 15: Data collection and statistics");
    println!("{}", "-".repeat(50));

    let sum = Arc::new(Mutex::new(0));
    let count = Arc::new(Mutex::new(0));
    let sum_clone = sum.clone();
    let count_clone = count.clone();

    let collector = BoxConsumer::new(move |x: &i32| {
        *sum_clone.lock().expect("mutex should not be poisoned") += *x;
        *count_clone.lock().expect("mutex should not be poisoned") += 1;
    });

    let numbers = vec![10, 20, 30, 40, 50];
    println!("Numbers: {:?}", numbers);
    for num in &numbers {
        collector.accept(num);
    }

    let total = *sum.lock().expect("mutex should not be poisoned");
    let cnt = *count.lock().expect("mutex should not be poisoned");
    println!("Sum: {}", total);
    println!("Count: {}", cnt);
    println!("Average: {:.2}\n", total as f64 / cnt as f64);

    println!("=== All examples completed ===");
    println!("\nTip: For value modification functionality, please refer to mutator_demo.rs");
}

Trait Implementations

impl<T> Clone for ArcConsumer<T>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<T> Consumer<T> for ArcConsumer<T>

fn accept(&self, value: &T)

Execute non-mutating consumption operation

fn into_box(self) -> BoxConsumer<T>

where Self: 'static,

Convert to BoxConsumer

fn into_rc(self) -> RcConsumer<T>

where Self: 'static,

Convert to RcConsumer

fn into_arc(self) -> ArcConsumer<T>

Convert to ArcConsumer

fn into_fn(self) -> impl Fn(&T)

Convert to closure

fn into_once(self) -> BoxConsumerOnce<T>

where Self: 'static,

Convert to ConsumerOnce

fn to_box(&self) -> BoxConsumer<T>

where Self: 'static,

Non-consuming conversion to BoxConsumer

fn to_rc(&self) -> RcConsumer<T>

where Self: 'static,

Non-consuming conversion to RcConsumer

fn to_arc(&self) -> ArcConsumer<T>

Non-consuming conversion to ArcConsumer

fn to_fn(&self) -> impl Fn(&T)

Non-consuming conversion to a boxed closure

fn to_once(&self) -> BoxConsumerOnce<T>

where Self: 'static,

Convert to ConsumerOnce without consuming self

impl<T> Debug for ArcConsumer<T>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<T> Display for ArcConsumer<T>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.