scaling.rs

use bastion::prelude::*;
use futures_timer::Delay;
use std::{sync::Arc, time::Duration};

#[derive(Debug)]
struct Start(usize);

async fn run_producer(ctx: BastionContext) -> Result<(), ()> {
    let runners = BroadcastTarget::Group("runners".into());
    let mut i = 0;
    loop {
        ctx.broadcast_message(runners.clone(), Start(i));
        println!("Pulled task {}", i);
        i += 1;
        Delay::new(Duration::from_millis(500)).await;
    }
}

async fn run_runner(ctx: BastionContext) -> Result<(), ()> {
    let producers = BroadcastTarget::Group("producers".into());
    println!("Runner has started!");
    loop {
        msg! { ctx.recv().await?,
            raw_message: Arc<SignedMessage> => {
                let msg = Arc::try_unwrap(raw_message).unwrap();
                msg! { msg,
                    ref start: Start => {
                        println!("Starting task {} from {}", start.0, ctx.current().id());
                        Delay::new(Duration::from_millis(5000)).await;
                    };
                    unhandled:_ => println!("Couldn't handle message {:?}", unhandled);
                }
            };
            _: _ => ();
        }
    }
}

fn setup_system() {
    Bastion::supervisor(|producers| {
        producers
            .with_strategy(SupervisionStrategy::OneForAll)
            .children(|producer| {
                producer
                    .with_dispatcher(Dispatcher::with_type(DispatcherType::Named(
                        "producers".into(),
                    )))
                    .with_resizer(OptimalSizeExploringResizer::default())
                    .with_exec(run_producer)
            })
    })
    .expect("Could not start Producers supervisor");

    Bastion::supervisor(|runners| {
        runners
            .with_strategy(SupervisionStrategy::OneForOne)
            .children(|runner| {
                runner
                    .with_dispatcher(Dispatcher::with_type(DispatcherType::Named(
                        "runners".into(),
                    )))
                    .with_resizer(
                        OptimalSizeExploringResizer::default()
                            .with_lower_bound(2)
                            .with_upper_bound(UpperBound::Limit(10)),
                    )
                    .with_exec(run_runner)
            })
    })
    .expect("Could not start Runners supervisor");
}

fn main() {
    Bastion::init();
    setup_system();
    Bastion::start();
    Bastion::block_until_stopped();
}