layered 0.3.3

Build composable async services with layered middleware.
Documentation 
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

// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.

#![allow(missing_docs, clippy::unwrap_used, reason = "sample code")]

//! Tower interoperability examples.
//!
//! Shows three ways to combine layered and Tower services:
//! - Layered stack with Tower execution (requires polling)
//! - Layered stack with `tower_layer()` adapter (no polling)
//! - Tower `ServiceBuilder` with layered layers

use std::future::poll_fn;

use layered::prelude::*;
use layered::tower::tower_layer;
use layered::{Execute, Intercept};
use tower::limit::GlobalConcurrencyLimitLayer;
use tower_service::Service as TowerService;

#[tokio::main]
async fn main() {
    println!("=== Layered with Tower Execution ===");
    example_oxidizer().await;

    println!("\n=== Layered with tower_layer() ===");
    example_oxidizer_native().await;

    println!("\n=== Tower ServiceBuilder ===");
    example_tower().await;
}

// Layered stack with Tower execution model (requires polling)
async fn example_oxidizer() {
    let stack = (
        GlobalConcurrencyLimitLayer::new(1),
        Intercept::layer().on_input(|input| println!("outer input: {input}")),
        GlobalConcurrencyLimitLayer::new(1),
        Intercept::layer().on_input(|input| println!("inner input: {input}")),
        Execute::new(execute),
    );

    let mut service = stack.into_service();
    poll_fn(|cx| service.poll_ready(cx)).await.unwrap();
    service.call("hello world from Oxidizer".to_string()).await.unwrap();
}

// Layered stack with tower_layer() adapter (no polling needed)
async fn example_oxidizer_native() {
    let stack = (
        tower_layer(GlobalConcurrencyLimitLayer::new(1)),
        Intercept::layer().on_input(|input| println!("outer input: {input}")),
        tower_layer(GlobalConcurrencyLimitLayer::new(1)),
        Intercept::layer().on_input(|input| println!("inner input: {input}")),
        Execute::new(execute),
    );

    let service = stack.into_service();

    // Direct execution - no polling required
    service.execute("hello world from Oxidizer Native".to_string()).await.unwrap();
}

// Tower ServiceBuilder with layered layers
async fn example_tower() {
    let mut service = tower::ServiceBuilder::new()
        .concurrency_limit(10)
        .layer(Intercept::layer().on_input(|input: &String| println!("outer input: {input}")))
        .concurrency_limit(1)
        .layer(Intercept::layer().on_input(|input: &String| println!("inner input: {input}")))
        .service_fn(execute);

    // Tower services must be polled for readiness first
    poll_fn(|cx| service.poll_ready(cx)).await.unwrap();
    service.call("hello world from Tower".to_string()).await.unwrap();
}

async fn execute(data: String) -> Result<String, String> {
    println!("executing input: {data}");
    Ok::<_, String>(data)
}