fluxion-stream 0.8.0

// Copyright 2025 Umberto Gotti <umberto.gotti@umbertogotti.dev>
// Licensed under the Apache License, Version 2.0
// http://www.apache.org/licenses/LICENSE-2.0

//! Error propagation tests for `map_ordered` operator.

use fluxion_core::{FluxionError, StreamItem};

use fluxion_stream::{CombineWithPreviousExt, MapOrderedExt};
use fluxion_test_utils::{test_channel_with_errors, unwrap_stream, Sequenced};

#[tokio::test]
async fn test_map_ordered_propagates_errors() -> anyhow::Result<()> {
    // Arrange
    let (tx, stream) = test_channel_with_errors::<Sequenced<i32>>();

    // Use combine_with_previous then map to string
    let mut result = stream
        .combine_with_previous()
        .map_ordered(|x| Sequenced::new(format!("Current: {}", x.current.value)));

    // Act & Assert: Send value
    tx.unbounded_send(StreamItem::Value(Sequenced::new(1)))?;
    assert!(
        matches!(unwrap_stream(&mut result, 100).await, StreamItem::Value(ref v) if v.value == "Current: 1")
    );

    // Send error
    tx.unbounded_send(StreamItem::Error(FluxionError::stream_error("Error")))?;
    assert!(matches!(
        unwrap_stream(&mut result, 100).await,
        StreamItem::Error(_)
    ));

    // Continue
    tx.unbounded_send(StreamItem::Value(Sequenced::new(2)))?;
    assert!(matches!(
        unwrap_stream(&mut result, 100).await,
        StreamItem::Value(_)
    ));

    drop(tx);

    Ok(())
}

#[tokio::test]
async fn test_map_ordered_transformation_after_error() -> anyhow::Result<()> {
    // Arrange
    let (tx, stream) = test_channel_with_errors::<Sequenced<i32>>();

    // Use combine_with_previous then map
    let mut result = stream
        .combine_with_previous()
        .map_ordered(|x| Sequenced::new(x.current.value * 2));

    // Act & AssertSend values
    tx.unbounded_send(StreamItem::Value(Sequenced::with_timestamp(10, 1)))?;
    assert!(matches!(
        unwrap_stream(&mut result, 100).await,
        StreamItem::Value(v) if v.value == 20
    ));

    tx.unbounded_send(StreamItem::Value(Sequenced::with_timestamp(20, 2)))?;
    assert!(matches!(
        unwrap_stream(&mut result, 100).await,
        StreamItem::Value(v) if v.value == 40
    ));

    // Send error
    tx.unbounded_send(StreamItem::Error(FluxionError::stream_error("Error")))?;
    assert!(matches!(
        unwrap_stream(&mut result, 100).await,
        StreamItem::Error(_)
    ));

    // Continue after error
    tx.unbounded_send(StreamItem::Value(Sequenced::with_timestamp(40, 4)))?;
    assert!(matches!(
        unwrap_stream(&mut result, 100).await,
        StreamItem::Value(v) if v.value == 80
    ));

    drop(tx);

    Ok(())
}

#[tokio::test]
async fn test_map_ordered_preserves_error_passthrough() -> anyhow::Result<()> {
    // Arrange
    let (tx, stream) = test_channel_with_errors::<Sequenced<i32>>();

    let mut result = stream
        .combine_with_previous()
        .map_ordered(|x| Sequenced::new(x.current.value * 100));

    // Act & Assert: Error immediately
    tx.unbounded_send(StreamItem::Error(FluxionError::stream_error("Error")))?;

    assert!(matches!(
        unwrap_stream(&mut result, 100).await,
        StreamItem::Error(_)
    ),);

    // Continue with value
    tx.unbounded_send(StreamItem::Value(Sequenced::with_timestamp(2, 2)))?;

    assert!(matches!(
        unwrap_stream(&mut result, 100).await,
        StreamItem::Value(v) if v.value == 200
    ));

    drop(tx);

    Ok(())
}

#[tokio::test]
async fn test_map_ordered_chain_after_error() -> anyhow::Result<()> {
    let (tx, stream) = test_channel_with_errors::<Sequenced<i32>>();

    // Chain combine_with_previous and map
    let mut result = stream
        .combine_with_previous()
        .map_ordered(|x| Sequenced::new(x.current.value * 2));

    // Act & Assert: Send value
    tx.unbounded_send(StreamItem::Value(Sequenced::with_timestamp(5, 1)))?;
    assert!(matches!(
        unwrap_stream(&mut result, 100).await,
        StreamItem::Value(v) if v.value == 10
    ));

    // Act & Assert: Send error
    tx.unbounded_send(StreamItem::Error(FluxionError::stream_error("Error")))?;
    assert!(matches!(
        unwrap_stream(&mut result, 100).await,
        StreamItem::Error(_)
    ));

    // Continue
    tx.unbounded_send(StreamItem::Value(Sequenced::with_timestamp(15, 3)))?;
    assert!(matches!(
        unwrap_stream(&mut result, 100).await,
        StreamItem::Value(v) if v.value == 30
    ));

    drop(tx);

    Ok(())
}