ruststream 0.7.0

//! The [`Broker`] / [`ConnectedBroker`] ladder: the entry point of any broker implementation.

use std::{error::Error as StdError, future::Future};

/// An unconnected broker: configuration captured, no I/O performed yet.
///
/// `Broker` is the entry point of any broker crate (`ruststream-nats`, `ruststream-kafka`, ...).
/// The lifecycle is a ladder of consuming transitions, so each state is a distinct type and
/// out-of-order calls do not compile:
///
/// ```text
/// B::new(config)                      unconnected: sync, I/O-free construction
/// broker.connect(self)  -> Connected  the live connection, a typed witness
/// connected.shutdown(self) -> Closed  the terminal witness (may carry diagnostics)
/// ```
///
/// Subscribing is described separately by a [`SubscriptionSource`](crate::SubscriptionSource)
/// (or the [`Subscribe`](crate::Subscribe) capability for the by-name case), resolved against the
/// [`Connected`](Self::Connected) form. Publishers are likewise produced by broker-specific
/// constructors.
///
/// `Send + Sync` is required so the runtime can move the broker across tasks.
///
/// # Lazy startup contract
///
/// Implementations MUST be constructible **synchronously**, without performing I/O: expose a plain
/// `new(..)` constructor that only captures configuration (addresses, credentials). All network
/// setup happens in [`connect`], which the runtime calls once at startup, after the synchronous
/// `#[ruststream::app]` builder has run. This is what lets a service be assembled with the app
/// macro regardless of broker. A broker that can only be built by connecting (an `async` "connect
/// and return the handle" constructor) does not satisfy this contract. Each broker also ships a
/// [`SubscriptionSource`](crate::SubscriptionSource) for its subjects, resolved against the
/// connected form.
/// [`conformance::harness::lifecycle`](crate::conformance::harness::lifecycle) checks the whole
/// ladder: synchronous construction, `connect`, subscribe through the source, deliver, ack,
/// `shutdown`, and the post-shutdown behaviour of aliased handles below.
///
/// # Internal tasks run on the runtime the broker connected on
///
/// A task the broker starts on its own behalf (a delayed-redelivery timer, a reply dispatcher,
/// a commit window) runs on the runtime [`connect`] ran on. The connected form keeps
/// `tokio::runtime::Handle::current()` from `connect` and spawns every such task through it,
/// whichever thread publishes, settles or requests. A handler on a dedicated thread publishes and
/// settles from that thread's own runtime: a task spawned there waits behind the handler's
/// computation and stops with that runtime, while the connection it serves lives on.
/// [`conformance::harness::lifecycle`](crate::conformance::harness::lifecycle) publishes and
/// settles with a delay from a runtime that stops at once, and
/// [`conformance::capabilities::request_reply`](crate::conformance::capabilities::request_reply)
/// requests from one; both then expect the work to complete.
///
/// ```
/// use std::time::Duration;
///
/// use tokio::runtime::Handle;
/// use tokio::sync::mpsc::UnboundedSender;
/// use tokio::time::sleep;
///
/// struct ConnectedExample {
///     // Captured in `connect`.
///     runtime: Handle,
/// }
///
/// impl ConnectedExample {
///     fn requeue_after(&self, back: UnboundedSender<Vec<u8>>, delivery: Vec<u8>, delay: Duration) {
///         self.runtime.spawn(async move {
///             sleep(delay).await;
///             let _ = back.send(delivery);
///         });
///     }
/// }
/// # let _ = ConnectedExample::requeue_after;
/// ```
///
/// # Shutdown is a type, not a flag
///
/// [`ConnectedBroker::shutdown`] consumes the connected broker, so misuse by the owner of the
/// handle (a publish or subscribe after shutdown) is a compile error, not a runtime one. What
/// remains dynamic is transport reality, not contract bookkeeping: handles aliasing the
/// connection (publishers created before shutdown, clones of a shareable broker) MUST surface an
/// error when used after the connection closed, never a silent success against a dead
/// connection. The conformance lifecycle check verifies that aliased-handle behaviour.
///
/// # Examples
///
/// ```
/// use ruststream::{Broker, ConnectedBroker};
///
/// async fn ladder<B: Broker>(broker: B) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
///     let connected = broker.connect().await?;
///     let _closed = connected.shutdown().await?;
///     Ok(())
/// }
/// ```
///
/// [`connect`]: Self::connect
pub trait Broker: Send + Sync + Sized {
    /// The error type returned by broker-level operations.
    type Error: StdError + Send + Sync + 'static;

    /// The connected form of this broker: the typed witness that [`connect`](Self::connect)
    /// succeeded.
    type Connected: ConnectedBroker;

    /// Establishes the connection to the broker, consuming the unconnected form.
    ///
    /// The returned [`Connected`](Self::Connected) value is the only way to reach the
    /// connection-bound surface (subscriptions, and for shareable brokers the live handles), so
    /// "not connected" is not representable on the owner's path.
    ///
    /// # Errors
    ///
    /// Returns [`Self::Error`] when the broker is unreachable, authentication fails, or the
    /// configuration is invalid.
    fn connect(self) -> impl Future<Output = Result<Self::Connected, Self::Error>> + Send;

    /// Wraps the broker for publisher-token minting before registration: the returned
    /// [`Bindable`](crate::runtime::Bindable) hands out
    /// [`Bound`](crate::runtime::Bound) tokens via its `bind`, and is itself what
    /// [`with_broker`](crate::runtime::RustStream::with_broker) takes.
    ///
    /// Provided for every broker; implementations do not override it.
    ///
    /// # Examples
    ///
    /// ```
    /// # #[cfg(all(feature = "macros", feature = "memory", feature = "json"))]
    /// # mod demo {
    /// use ruststream::memory::prelude::*;
    /// use serde::{Deserialize, Serialize};
    ///
    /// #[derive(Deserialize)]
    /// struct Order {
    ///     id: u64,
    /// }
    ///
    /// #[derive(Serialize, Outgoing)]
    /// #[outgoing(name = "orders.accepted")]
    /// struct Accepted {
    ///     id: u64,
    /// }
    ///
    /// #[subscriber("orders", reply)]
    /// async fn accept(order: &Order) -> Accepted {
    ///     Accepted { id: order.id }
    /// }
    ///
    /// fn app() -> RustStream {
    ///     // Orders arrive on one bus and their answers leave on another.
    ///     let egress = MemoryBroker::new().bindable();
    ///     let to_egress = egress.bind(Publish);
    ///     RustStream::new(AppInfo::new("bridge", "0.1.0"))
    ///         .with_broker_labeled("egress", egress, |_b| {})
    ///         .with_broker_labeled("ingress", MemoryBroker::new(), |b| {
    ///             b.include(accept).out_reply(to_egress);
    ///         })
    /// }
    /// # }
    /// # fn main() {}
    /// ```
    #[must_use]
    fn bindable(self) -> crate::runtime::Bindable<Self>
    where
        Self: 'static,
    {
        crate::runtime::Bindable::new(self)
    }
}

/// A connected broker: the typed witness of a live connection.
///
/// Obtained only from [`Broker::connect`]. The `'static` supertrait keeps the connected form an
/// owned value the runtime can hold and erase.
///
/// # Examples
///
/// ```
/// use ruststream::ConnectedBroker;
///
/// async fn stop<C: ConnectedBroker>(connected: C) -> Result<C::Closed, C::Error> {
///     connected.shutdown().await
/// }
/// ```
pub trait ConnectedBroker: Send + Sync + Sized + 'static {
    /// The error type returned by connected-broker operations.
    type Error: StdError + Send + Sync + 'static;

    /// The terminal witness returned by [`shutdown`](Self::shutdown).
    ///
    /// A closed broker has no publish or subscribe surface; the witness may carry teardown
    /// diagnostics (flush results, drop counts) as plain data.
    type Closed: Send;

    /// Closes the broker connection, flushing in-flight publishes and stopping background tasks,
    /// consuming the connected form.
    ///
    /// Consuming `self` makes a second shutdown, or any operation after shutdown, a compile
    /// error for the owner of the handle. Aliased handles (publishers handed out earlier,
    /// clones of a shareable broker) surface the transport's own error when used afterwards;
    /// see the [`Broker`] contract.
    ///
    /// # Errors
    ///
    /// Returns [`Self::Error`] when the broker rejects the disconnect or a background flush
    /// fails to complete.
    fn shutdown(self) -> impl Future<Output = Result<Self::Closed, Self::Error>> + Send;
}

/// Shorthand for a broker's connected form, so bounds read
/// `S: SubscriptionSource<Connected<B>>` instead of spelling the associated type projection.
pub type Connected<B> = <B as Broker>::Connected;