pipecircuit-cli 0.0.2

use app_core::{domain_primitives::Count, state_machines::Counter};
use app_program::{
    CounterProgram, counting,
    pipes::{Client, CounterAuthority, Counting, Reply, Request},
};
use core::{convert::Infallible, num::NonZeroUsize};
use pipecircuit::{
    AdmissionFailure, DriveFailure, Inline, InlineSubmission, InstalledPipe, Pipecircuit,
};
use pipecircuit_mio::{
    MioRegistrationId, MioTimedReactor,
    host::{
        MioAssembly, MioConductor, MioHost, MioHostSettings, MioHostedCircuit, MioPipeSet,
        MioPreparationFailure, MioProgress,
    },
    spsc::{
        AcceptRetained, BoundedPendingCapacity, FixedSizeCharge, PendingAdmissionFailure,
        RetainingSpscFailure, RetainingSpscPipe, RetainingSubmissionFailure,
        RetentionCreationError, SpscCapacity, SpscEndpoint, SpscPipe, pair,
    },
};
use std::io;

pub type ClientDriver = RetainingSpscPipe<Client, Reply, Request, FixedSizeCharge<Reply>>;
pub type CountingDriver = Inline<CounterAuthority, Counting>;

#[derive(MioPipeSet)]
#[mio_roster(counting)]
pub struct InstalledPipes {
    /// Owned queues connect an ordinary CLI peer to this program.
    pub client: InstalledPipe<Client, ClientDriver>,
    /// The counter authority executes inline in this program.
    pub counter: InstalledPipe<Counting, CountingDriver>,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum Completion {
    // A physical disconnect makes no claim that the peer consumed queued replies.
    ClientDisconnected,
}

pub type SubmissionFailure =
    counting::SubmissionFailure<RetainingSubmissionFailure<Reply>, Infallible>;
pub type TurnFailure = DriveFailure<
    RetainingSpscFailure,
    AdmissionFailure<PendingAdmissionFailure, SubmissionFailure>,
>;
pub type InstalledCircuit<Conduct> = MioHostedCircuit<MioAssembly<InstalledPipes, Conduct>>;
pub type InstalledProgram<Conduct> = Pipecircuit<CounterProgram, InstalledCircuit<Conduct>>;

pub struct Configuration {
    pub requests: NonZeroUsize,
    pub replies: NonZeroUsize,
    pub retained_replies: NonZeroUsize,
    pub dispatch_quantum: NonZeroUsize,
}

#[derive(Debug)]
pub enum OpenFailure {
    Host(io::Error),
    Io(io::Error),
    LocalWake(pipecircuit_mio::MioWakerOpenFailure),
    PeerWake(pipecircuit_mio::MioWakerOpenFailure),
    ReplyStorage(RetentionCreationError),
    Installation(MioPreparationFailure),
}

pub struct Application<Conduct>
where
    Conduct: MioConductor<CounterProgram, InstalledPipes>,
{
    pub pipecircuit: InstalledProgram<Conduct>,
    pub peer: SpscEndpoint<Request, Reply>,
    pub peer_reactor: MioTimedReactor,
}

pub fn open(
    configuration: Configuration,
) -> Result<
    Application<
        impl MioConductor<
            CounterProgram,
            InstalledPipes,
            Completion = Completion,
            Failure = TurnFailure,
        >,
    >,
    OpenFailure,
> {
    let host = MioHost::prepare(MioHostSettings {
        poll_capacity: NonZeroUsize::MIN,
        dispatch_quantum: configuration.dispatch_quantum,
    })
    .map_err(OpenFailure::Host)?;
    let local_wake = host
        .try_open_waker(MioRegistrationId::new(1))
        .map_err(OpenFailure::LocalWake)?;
    let peer_reactor = MioTimedReactor::try_open().map_err(OpenFailure::Io)?;
    let peer_wake = peer_reactor
        .try_open_waker(MioRegistrationId::new(1))
        .map_err(OpenFailure::PeerWake)?;
    let (local, peer) = pair(
        SpscCapacity {
            to_peer: configuration.replies,
            from_peer: configuration.requests,
        },
        local_wake,
        peer_wake,
    );
    let authority = Counter::new(Count::ZERO);
    let pipes = InstalledPipes::from(counting::Pipes {
        client: InstalledPipe::new(
            SpscPipe::attach(local)
                .with_fixed_size_retention(configuration.retained_replies)
                .map_err(|failure| OpenFailure::ReplyStorage(failure.failure))?,
        ),
        counter: InstalledPipe::new(Inline::for_pipe::<Counting>(CounterAuthority::new(
            authority,
        ))),
    });
    let pipecircuit = host
        .install_program(
            CounterProgram,
            pipes.with_submission_policies(
                counting::SubmissionPolicies {
                    client: AcceptRetained,
                    counter: InlineSubmission,
                },
                |source| {
                    let InstalledPipesSourceTurn::Client(turn) = source;
                    turn.with_fallible_admission(|_, _, capacity: BoundedPendingCapacity| {
                        // The installed inline route produces one fixed-size owned reply.
                        capacity.admission()
                    })
                    .conduct(|| MioProgress::Continue)
                    .complete_on_physical_failure(|failure| match failure {
                        RetainingSpscFailure::IncomingClosed => Ok(Completion::ClientDisconnected),
                        failure @ (RetainingSpscFailure::Pending(_)
                        | RetainingSpscFailure::CapacityWake(_)) => Err(failure),
                    })
                },
            ),
        )
        .map_err(|failure| OpenFailure::Installation(failure.reason))?;
    Ok(Application {
        pipecircuit,
        peer,
        peer_reactor,
    })
}

impl core::fmt::Display for OpenFailure {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        match self {
            Self::Host(failure) => write!(f, "host creation: {failure:?}"),
            Self::Io(failure) => write!(f, "peer reactor: {failure}"),
            Self::LocalWake(failure) => write!(f, "local wake: {failure:?}"),
            Self::PeerWake(failure) => write!(f, "peer wake: {failure:?}"),
            Self::ReplyStorage(failure) => write!(f, "reply storage: {failure:?}"),
            Self::Installation(failure) => write!(f, "installation: {failure:?}"),
        }
    }
}