ruststream 0.7.3

//! Workers: a subscription that waits on I/O handles several deliveries at once, and `by_key`
//! keeps the deliveries of one key in order. Each scenario is a test.
//!
//! ```text
//! cargo test --example workers --features testing,macros,memory,json
//! ```

// A plain build compiles the services; the tests that use them exist only in the test build.
#![cfg_attr(not(test), allow(dead_code, unused_imports))]

use std::time::Duration;

use serde::{Deserialize, Serialize};

#[derive(Debug, PartialEq, Deserialize, Serialize, ruststream::Outgoing)]
struct Order {
    id: u64,
}

/// Stands in for a call to a payment provider: a second of waiting on I/O.
async fn charge(order: &Order) {
    tokio::time::sleep(Duration::from_secs(1)).await;
    tracing::info!(order.id, "charged");
}

mod pool {
    use std::error::Error;

    use futures::future::join_all;
    use ruststream::memory::prelude::*;
    use ruststream::testing::TestApp;
    use tokio::time::Instant;

    use super::{Order, charge};

    // --8<-- [start:workers]
    /// Up to four orders are charged at once; the next one waits for a free worker.
    #[subscriber("orders", workers(4))]
    async fn bill(order: &Order) -> HandlerOutcome {
        charge(order).await;
        HandlerOutcome::ack()
    }
    // --8<-- [end:workers]

    fn app() -> impl App {
        RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
            b.include(bill);
        })
    }

    #[tokio::test]
    async fn four_orders_are_charged_together() -> Result<(), Box<dyn Error>> {
        tokio::time::pause();
        let tb = TestApp::start(app()).await?;
        let broker = tb.broker::<MemoryBroker>();
        let orders: Vec<Order> = (1..=4).map(|id| Order { id }).collect();

        let started = Instant::now();
        for published in join_all(
            orders
                .iter()
                .map(|order| broker.message(order).to("orders").publish()),
        )
        .await
        {
            published?;
        }

        // Four charges of a second each, side by side: one second passed, not four.
        assert_eq!(started.elapsed().as_secs(), 1);
        broker
            .subscriber("orders")
            .assert_called(4)
            .settled(HandlerOutcome::ack());
        Ok(())
    }
}

mod by_key {
    use std::error::Error;

    use futures::future::join_all;
    use ruststream::HeaderMap;
    use ruststream::memory::PARTITION_KEY_HEADER;
    use ruststream::memory::prelude::*;
    use ruststream::testing::TestApp;
    use tokio::time::Instant;

    use super::{Order, charge};

    // --8<-- [start:workers_by_key]
    /// Four lanes by the message's partition key: one customer's orders are charged in order,
    /// different customers' at once.
    #[subscriber("orders", workers(4, by_key))]
    async fn bill(order: &Order) -> HandlerOutcome {
        charge(order).await;
        HandlerOutcome::ack()
    }
    // --8<-- [end:workers_by_key]

    fn app() -> impl App {
        RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
            b.include(bill);
        })
    }

    #[tokio::test]
    async fn one_customer_waits_for_its_own_orders_only() -> Result<(), Box<dyn Error>> {
        tokio::time::pause();
        let tb = TestApp::start(app()).await?;
        let broker = tb.broker::<MemoryBroker>();
        // Two orders of each customer; the in-memory broker reads the key from a header.
        let orders: Vec<(Order, HeaderMap)> = [(1, "alice"), (2, "bob"), (3, "alice"), (4, "bob")]
            .into_iter()
            .map(|(id, customer)| {
                let mut headers = HeaderMap::new();
                headers.insert(PARTITION_KEY_HEADER, customer);
                (Order { id }, headers)
            })
            .collect();

        let started = Instant::now();
        for published in join_all(orders.iter().map(|(order, headers)| {
            broker
                .message(order)
                .with_headers(headers.clone())
                .to("orders")
                .publish()
        }))
        .await
        {
            published?;
        }

        // Each customer's two charges ran one after the other, beside the other customer's.
        assert_eq!(started.elapsed().as_secs(), 2);
        broker
            .subscriber("orders")
            .assert_called(4)
            .settled(HandlerOutcome::ack());
        Ok(())
    }
}

// The scenarios are the tests above; the command in the header runs them.
fn main() {}