pipecircuit-cli 0.0.2

use app_core::{domain_primitives::Amount, state_machines::AddOutcome};
use app_program::pipes::{Reply, Request};
use pipecircuit::Reactor as _;
use pipecircuit_mio::{
    MioTimedReaction, MioTimedReactor, MioTimedReactorFailure,
    spsc::{QueueReception, QueueSubmission, SpscEndpoint},
};
use std::io::{self, Write as _};

pub struct Client {
    pub endpoint: SpscEndpoint<Request, Reply>,
    pub reactor: MioTimedReactor,
    pub additions: Vec<Amount>,
}

#[derive(Debug)]
pub enum Failure {
    Rejected(fibre::TrySendError<Request>),
    AdmittedButWakeFailed(io::Error),
    ReceivedButWakeFailed { reply: Reply, failure: io::Error },
    Wait(MioTimedReactorFailure),
    UnexpectedTimer,
    Disconnected,
    Output(io::Error),
}

#[derive(Debug)]
pub struct Exit {
    pub exchange: Result<(), Failure>,
    pub close: io::Result<()>,
}

impl Client {
    pub fn run(mut self) -> Exit {
        let additions = core::mem::take(&mut self.additions);
        let exchange = additions
            .into_iter()
            .map(Request::Add)
            .chain(core::iter::once(Request::Read))
            .try_for_each(|request| self.exchange(request));
        Exit {
            exchange,
            close: self.endpoint.close(),
        }
    }

    fn exchange(&mut self, request: Request) -> Result<(), Failure> {
        // This peer permits one outstanding request. A full queue is a refusal,
        // never a retry that could duplicate an already admitted operation.
        match self.endpoint.submit(request) {
            QueueSubmission::Admitted => {}
            QueueSubmission::Rejected(rejected) => return Err(Failure::Rejected(rejected)),
            QueueSubmission::AdmittedButWakeFailed(failure) => {
                return Err(Failure::AdmittedButWakeFailed(failure));
            }
        }
        loop {
            match self.endpoint.receive() {
                QueueReception::Received {
                    output: reply,
                    capacity_wake,
                } => {
                    capacity_wake
                        .map_err(|failure| Failure::ReceivedButWakeFailed { reply, failure })?;
                    let mut output = io::stdout().lock();
                    return match reply {
                        Reply::Added(AddOutcome::Added { before, after }) => {
                            writeln!(output, "count: {} (was {})", after.get(), before.get())
                        }
                        Reply::Added(AddOutcome::Overflow { current, requested }) => writeln!(
                            output,
                            "overflow: {} + {}; count unchanged",
                            current.get(),
                            requested.get()
                        ),
                        Reply::Snapshot(count) => writeln!(output, "snapshot: {}", count.get()),
                    }
                    .map_err(Failure::Output);
                }
                QueueReception::Closed => return Err(Failure::Disconnected),
                QueueReception::Empty => {
                    match self.reactor.wait_for_reaction().map_err(Failure::Wait)? {
                        MioTimedReaction::Readiness(_) => {}
                        MioTimedReaction::TimerElapsed => return Err(Failure::UnexpectedTimer),
                    }
                }
            }
        }
    }
}

impl core::fmt::Display for Failure {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        match self {
            Self::Rejected(rejected) => write!(f, "request rejected: {rejected:?}"),
            Self::AdmittedButWakeFailed(failure) => {
                write!(f, "request admitted, wake failed: {failure}")
            }
            Self::ReceivedButWakeFailed { reply, failure } => {
                write!(f, "received {reply:?}, capacity wake failed: {failure}")
            }
            Self::Wait(failure) => write!(f, "reactor wait failed: {failure:?}"),
            Self::UnexpectedTimer => f.write_str("unexpected timer without a configured deadline"),
            Self::Disconnected => f.write_str("counter disconnected before replying"),
            Self::Output(failure) => write!(f, "output failed: {failure}"),
        }
    }
}