saddle-runtime 0.3.24

//! Service-shaped pre-poll factory handoff. No protocol parser or listener is
//! implemented here; the validated identity future is supplied by the entry.
//! This compile consumer does not claim full Service layout or HTTP acceptance.
#![allow(dead_code)]
use saddle_observability::{CallContext, EventContext, Observer};
use saddle_runtime::{
    profusegw::{
        ProfuseGwAdmissionEvent, ProfuseGwProcessLease, ReservedDispatchOutcome,
        ReservedDispatchOwner, finish_profusegw_without_database,
    },
    request_task::reserved::{
        ReservedBorrowedFuture, ReservedRequestFailure, ReservedTaskContext,
        dispatch_storage_bytes_for,
    },
};
use std::{alloc::Layout, future::Future};
type Owner = ReservedDispatchOwner<tokio::net::TcpStream>;
struct NormalFactory<I> {
    observer: Observer,
    identity: I,
}
impl<I: Future<Output = Option<(CallContext, EventContext)>> + Send> NormalFactory<I> {
    async fn body(
        self,
        owner: &mut Owner,
        _task: ReservedTaskContext,
    ) -> Result<bool, ReservedRequestFailure> {
        // A Pending identity read never takes the original event out of owner.
        // Abort therefore preserves it with the supervising ticket.
        let Some((call, event)) = self.identity.await else {
            return Ok(false);
        };
        let original = owner.2.take().expect("single original admission event");
        let dispatch = owner.0.take().expect("same admitted owner");
        owner.0 = Some(dispatch.bind_observation(self.observer, call, event, original));
        // Service continues its original DB/handler/delivery loop here.
        // This example stops before that logic; binding is not written proof.
        Ok(true)
    }
}
fn boxed<I: Future<Output = Option<(CallContext, EventContext)>> + Send + 'static>(
    factory: NormalFactory<I>,
) -> impl for<'a> FnOnce(&'a mut Owner, ReservedTaskContext) -> ReservedBorrowedFuture<'a, bool>
+ Send
+ 'static {
    move |owner, task| Box::pin(factory.body(owner, task))
}
fn result_layout<A, R>(_: impl FnOnce(A) -> R) -> Layout {
    Layout::new::<R>()
}
fn body_layout<I: Future<Output = Option<(CallContext, EventContext)>> + Send + 'static>() -> Layout
{
    result_layout(
        |(factory, owner, task): (NormalFactory<I>, &'static mut Owner, ReservedTaskContext)| {
            factory.body(owner, task)
        },
    )
}
struct Recovered {
    socket: tokio::net::TcpStream,
    /// Returned untouched when identity was absent or polling was cancelled.
    /// Caller must retain/handle it explicitly; no fake trace and no silent
    /// fallback consumption is hidden in this adapter.
    unbound: Option<ProfuseGwAdmissionEvent>,
    bound: bool,
}
async fn consume<I: Future<Output = Option<(CallContext, EventContext)>> + Send + 'static>(
    lease: &ProfuseGwProcessLease,
    socket: tokio::net::TcpStream,
    factory: NormalFactory<I>,
    cancel_before_poll: bool,
) -> Result<
    Recovered,
    ReservedDispatchOutcome<
        tokio::net::TcpStream,
        bool,
        impl for<'a> FnOnce(&'a mut Owner, ReservedTaskContext) -> ReservedBorrowedFuture<'a, bool>
        + Send
        + 'static,
    >,
> {
    let make = boxed(factory);
    // Same concrete capture/body/owner that will be consumed; no factory call
    // or task/Arc allocation precedes Runtime's reservation.
    let _required = dispatch_storage_bytes_for::<tokio::net::TcpStream, bool, _>(
        &make,
        body_layout::<I>(),
        &[],
    )
    .unwrap();
    let (root, future, mut ticket) = match lease.try_reserved_dispatch(
        saddle_core::request_context::ContextLabel::checked("app").unwrap(),
        None,
        body_layout::<I>(),
        &[],
        socket,
        make,
    ) {
        ReservedDispatchOutcome::Ready {
            root,
            future,
            ticket,
        } => (root, future, ticket),
        failure => return Err(failure), // Includes original input, uncalled make and event.
    };
    let mut tasks = tokio::task::JoinSet::new();
    let handle = tasks.spawn(future);
    ticket.bind(handle.id()).unwrap();
    if cancel_before_poll {
        handle.abort();
    }
    let joined = ticket
        .complete(tasks.join_next().await.unwrap())
        .ok()
        .unwrap();
    let mut recovered = joined.recover(Default::default()).ok().unwrap();
    let bound = recovered
        .result
        .take()
        .and_then(Result::ok)
        .unwrap_or(false);
    // No DB was entered by this compile consumer. Real Service retains its
    // actual physical completion until delivery/cleanup; this is not that test.
    finish_profusegw_without_database(recovered.owner.0.take().unwrap(), ())
        .ok()
        .unwrap();
    drop(root);
    Ok(Recovered {
        socket: recovered.owner.1,
        unbound: recovered.owner.2,
        bound,
    })
}
fn main() {
    println!("AR Service-shaped factory compiles; listener/HTTP/DB NOT_RUN");
}