datum-core 0.9.0

//! Port-bundle shapes describing a graph's (or stage's) external ports.
//!
//! A [`Shape`] is the contract returned from a `GraphDsl` builder closure and
//! declared by every [`GraphStage`](super::GraphStage): it enumerates the open
//! inlets and outlets. The builder's `finish` step checks that the result
//! shape's ports exactly match the graph's unconnected ports. Shapes mirror
//! Akka's `Shape` hierarchy (`SourceShape`, `FlowShape`, `FanInShape`, …).

use super::*;

/// A bundle of typed external ports. Implementors expose their inlets/outlets
/// in a stable order; `inlets()`/`outlets()` return the type-erased views the
/// builder validates against. Cheap to clone (ports are id + name).
pub trait Shape: Clone + Send + Sync + 'static {
    fn inlets(&self) -> Vec<AnyInlet>;
    fn outlets(&self) -> Vec<AnyOutlet>;
}

/// One outlet, no inlets — a graph that behaves as a `Source`.
#[derive(Debug, PartialEq, Eq)]
pub struct SourceShape<Out: 'static> {
    outlet: Outlet<Out>,
}

impl<Out: 'static> Clone for SourceShape<Out> {
    fn clone(&self) -> Self {
        Self {
            outlet: self.outlet.clone(),
        }
    }
}

impl<Out: 'static> SourceShape<Out> {
    #[must_use]
    pub fn new(outlet: Outlet<Out>) -> Self {
        Self { outlet }
    }

    #[must_use]
    pub fn outlet(&self) -> Outlet<Out> {
        self.outlet.clone()
    }
}

impl<Out: 'static> Shape for SourceShape<Out> {
    fn inlets(&self) -> Vec<AnyInlet> {
        Vec::new()
    }

    fn outlets(&self) -> Vec<AnyOutlet> {
        vec![self.outlet.erase()]
    }
}

/// One inlet, no outlets — a graph that behaves as a `Sink`.
#[derive(Debug, PartialEq, Eq)]
pub struct SinkShape<In: 'static> {
    inlet: Inlet<In>,
}

impl<In: 'static> Clone for SinkShape<In> {
    fn clone(&self) -> Self {
        Self {
            inlet: self.inlet.clone(),
        }
    }
}

impl<In: 'static> SinkShape<In> {
    #[must_use]
    pub fn new(inlet: Inlet<In>) -> Self {
        Self { inlet }
    }

    #[must_use]
    pub fn inlet(&self) -> Inlet<In> {
        self.inlet.clone()
    }
}

impl<In: 'static> Shape for SinkShape<In> {
    fn inlets(&self) -> Vec<AnyInlet> {
        vec![self.inlet.erase()]
    }

    fn outlets(&self) -> Vec<AnyOutlet> {
        Vec::new()
    }
}

/// One inlet and one outlet — a graph that behaves as a `Flow`. Re-exported as
/// `GraphFlowShape` from the crate root.
#[derive(Debug, PartialEq, Eq)]
pub struct FlowShape<In: 'static, Out: 'static> {
    inlet: Inlet<In>,
    outlet: Outlet<Out>,
}

impl<In: 'static, Out: 'static> Clone for FlowShape<In, Out> {
    fn clone(&self) -> Self {
        Self {
            inlet: self.inlet.clone(),
            outlet: self.outlet.clone(),
        }
    }
}

impl<In: 'static, Out: 'static> FlowShape<In, Out> {
    #[must_use]
    pub fn new(inlet: Inlet<In>, outlet: Outlet<Out>) -> Self {
        Self { inlet, outlet }
    }

    #[must_use]
    pub fn inlet(&self) -> Inlet<In> {
        self.inlet.clone()
    }

    #[must_use]
    pub fn outlet(&self) -> Outlet<Out> {
        self.outlet.clone()
    }
}

impl<In: 'static, Out: 'static> Shape for FlowShape<In, Out> {
    fn inlets(&self) -> Vec<AnyInlet> {
        vec![self.inlet.erase()]
    }

    fn outlets(&self) -> Vec<AnyOutlet> {
        vec![self.outlet.erase()]
    }
}

/// N inlets and one outlet — the shape of merge-style junctions (`Merge`,
/// `Concat`, `Interleave`, `OrElse`, `MergeSorted`, …). `Out` defaults to `In`;
/// `MergeLatest` uses `Out = Vec<In>`.
#[derive(Debug, PartialEq, Eq)]
pub struct FanInShape<In: 'static, Out: 'static = In> {
    inlets: Vec<Inlet<In>>,
    outlet: Outlet<Out>,
}

impl<In: 'static, Out: 'static> Clone for FanInShape<In, Out> {
    fn clone(&self) -> Self {
        Self {
            inlets: self.inlets.clone(),
            outlet: self.outlet.clone(),
        }
    }
}

impl<In: 'static, Out: 'static> FanInShape<In, Out> {
    #[must_use]
    pub fn new(inlets: Vec<Inlet<In>>, outlet: Outlet<Out>) -> Self {
        Self { inlets, outlet }
    }

    #[must_use]
    pub fn inlet_count(&self) -> usize {
        self.inlets.len()
    }

    pub fn inlet(&self, index: usize) -> StreamResult<Inlet<In>> {
        self.inlets
            .get(index)
            .cloned()
            .ok_or_else(|| StreamError::GraphValidation(format!("fan-in inlet {index} is missing")))
    }

    #[must_use]
    pub fn inlets_vec(&self) -> Vec<Inlet<In>> {
        self.inlets.clone()
    }

    #[must_use]
    pub fn outlet(&self) -> Outlet<Out> {
        self.outlet.clone()
    }
}

impl<In: 'static, Out: 'static> Shape for FanInShape<In, Out> {
    fn inlets(&self) -> Vec<AnyInlet> {
        self.inlets.iter().map(Inlet::erase).collect()
    }

    fn outlets(&self) -> Vec<AnyOutlet> {
        vec![self.outlet.erase()]
    }
}

/// One preferred inlet, N secondary inlets, and one outlet — the shape of
/// `MergePreferred`. Explicit-only in the wiring DSL (use `.preferred()` /
/// `.secondary(i)`).
#[derive(Debug, PartialEq, Eq)]
pub struct MergePreferredShape<T: 'static> {
    preferred: Inlet<T>,
    secondary: Vec<Inlet<T>>,
    outlet: Outlet<T>,
}

impl<T: 'static> Clone for MergePreferredShape<T> {
    fn clone(&self) -> Self {
        Self {
            preferred: self.preferred.clone(),
            secondary: self.secondary.clone(),
            outlet: self.outlet.clone(),
        }
    }
}

impl<T: 'static> MergePreferredShape<T> {
    #[must_use]
    pub fn new(preferred: Inlet<T>, secondary: Vec<Inlet<T>>, outlet: Outlet<T>) -> Self {
        Self {
            preferred,
            secondary,
            outlet,
        }
    }

    #[must_use]
    pub fn preferred(&self) -> Inlet<T> {
        self.preferred.clone()
    }

    #[must_use]
    pub fn secondary_count(&self) -> usize {
        self.secondary.len()
    }

    pub fn secondary(&self, index: usize) -> StreamResult<Inlet<T>> {
        self.secondary.get(index).cloned().ok_or_else(|| {
            StreamError::GraphValidation(format!(
                "merge-preferred secondary inlet {index} is missing"
            ))
        })
    }

    #[must_use]
    pub fn secondary_vec(&self) -> Vec<Inlet<T>> {
        self.secondary.clone()
    }

    #[must_use]
    pub fn outlet(&self) -> Outlet<T> {
        self.outlet.clone()
    }
}

impl<T: 'static> Shape for MergePreferredShape<T> {
    fn inlets(&self) -> Vec<AnyInlet> {
        std::iter::once(self.preferred.erase())
            .chain(self.secondary.iter().map(Inlet::erase))
            .collect()
    }

    fn outlets(&self) -> Vec<AnyOutlet> {
        vec![self.outlet.erase()]
    }
}

/// One inlet and N outlets — the shape of fan-out junctions (`Broadcast`,
/// `Balance`, `Partition`). `Out` defaults to `In`.
#[derive(Debug, PartialEq, Eq)]
pub struct FanOutShape<In: 'static, Out: 'static = In> {
    inlet: Inlet<In>,
    outlets: Vec<Outlet<Out>>,
}

impl<In: 'static, Out: 'static> Clone for FanOutShape<In, Out> {
    fn clone(&self) -> Self {
        Self {
            inlet: self.inlet.clone(),
            outlets: self.outlets.clone(),
        }
    }
}

impl<In: 'static, Out: 'static> FanOutShape<In, Out> {
    #[must_use]
    pub fn new(inlet: Inlet<In>, outlets: Vec<Outlet<Out>>) -> Self {
        Self { inlet, outlets }
    }

    #[must_use]
    pub fn inlet(&self) -> Inlet<In> {
        self.inlet.clone()
    }

    #[must_use]
    pub fn outlet_count(&self) -> usize {
        self.outlets.len()
    }

    pub fn outlet(&self, index: usize) -> StreamResult<Outlet<Out>> {
        self.outlets.get(index).cloned().ok_or_else(|| {
            StreamError::GraphValidation(format!("fan-out outlet {index} is missing"))
        })
    }

    #[must_use]
    pub fn outlets_vec(&self) -> Vec<Outlet<Out>> {
        self.outlets.clone()
    }
}

impl<In: 'static, Out: 'static> Shape for FanOutShape<In, Out> {
    fn inlets(&self) -> Vec<AnyInlet> {
        vec![self.inlet.erase()]
    }

    fn outlets(&self) -> Vec<AnyOutlet> {
        self.outlets.iter().map(Outlet::erase).collect()
    }
}

/// One inlet and two heterogeneously-typed outlets — the shape of `Unzip` and
/// `UnzipWith`. Outlets are reached via `.out0()` / `.out1()`.
#[derive(Debug, PartialEq, Eq)]
pub struct FanOutShape2<In: 'static, Out0: 'static, Out1: 'static> {
    inlet: Inlet<In>,
    out0: Outlet<Out0>,
    out1: Outlet<Out1>,
}

impl<In: 'static, Out0: 'static, Out1: 'static> Clone for FanOutShape2<In, Out0, Out1> {
    fn clone(&self) -> Self {
        Self {
            inlet: self.inlet.clone(),
            out0: self.out0.clone(),
            out1: self.out1.clone(),
        }
    }
}

impl<In: 'static, Out0: 'static, Out1: 'static> FanOutShape2<In, Out0, Out1> {
    #[must_use]
    pub fn new(inlet: Inlet<In>, out0: Outlet<Out0>, out1: Outlet<Out1>) -> Self {
        Self { inlet, out0, out1 }
    }

    #[must_use]
    pub fn inlet(&self) -> Inlet<In> {
        self.inlet.clone()
    }

    #[must_use]
    pub fn out0(&self) -> Outlet<Out0> {
        self.out0.clone()
    }

    #[must_use]
    pub fn out1(&self) -> Outlet<Out1> {
        self.out1.clone()
    }
}

impl<In: 'static, Out0: 'static, Out1: 'static> Shape for FanOutShape2<In, Out0, Out1> {
    fn inlets(&self) -> Vec<AnyInlet> {
        vec![self.inlet.erase()]
    }

    fn outlets(&self) -> Vec<AnyOutlet> {
        vec![self.out0.erase(), self.out1.erase()]
    }
}

/// Two heterogeneously-typed inlets and one `(Left, Right)` outlet — the shape
/// of `Zip`. Inlets are reached via `.in0()` / `.in1()`.
#[derive(Debug, PartialEq, Eq)]
pub struct ZipShape<Left: 'static, Right: 'static> {
    left: Inlet<Left>,
    right: Inlet<Right>,
    outlet: Outlet<(Left, Right)>,
}

impl<Left: 'static, Right: 'static> Clone for ZipShape<Left, Right> {
    fn clone(&self) -> Self {
        Self {
            left: self.left.clone(),
            right: self.right.clone(),
            outlet: self.outlet.clone(),
        }
    }
}

impl<Left: 'static, Right: 'static> ZipShape<Left, Right> {
    #[must_use]
    pub fn new(left: Inlet<Left>, right: Inlet<Right>, outlet: Outlet<(Left, Right)>) -> Self {
        Self {
            left,
            right,
            outlet,
        }
    }

    #[must_use]
    pub fn in0(&self) -> Inlet<Left> {
        self.left.clone()
    }

    #[must_use]
    pub fn in1(&self) -> Inlet<Right> {
        self.right.clone()
    }

    #[must_use]
    pub fn outlet(&self) -> Outlet<(Left, Right)> {
        self.outlet.clone()
    }
}

impl<Left: 'static, Right: 'static> Shape for ZipShape<Left, Right> {
    fn inlets(&self) -> Vec<AnyInlet> {
        vec![self.left.erase(), self.right.erase()]
    }

    fn outlets(&self) -> Vec<AnyOutlet> {
        vec![self.outlet.erase()]
    }
}

/// Two inlets and two outlets arranged as two independent directions — the
/// shape of a bidirectional flow (top: `I1 → O1`, bottom: `I2 → O2`).
/// Explicit-only in the wiring DSL.
#[derive(Debug, PartialEq, Eq)]
pub struct BidiShape<I1: 'static, O1: 'static, I2: 'static, O2: 'static> {
    in1: Inlet<I1>,
    out1: Outlet<O1>,
    in2: Inlet<I2>,
    out2: Outlet<O2>,
}

impl<I1: 'static, O1: 'static, I2: 'static, O2: 'static> Clone for BidiShape<I1, O1, I2, O2> {
    fn clone(&self) -> Self {
        Self {
            in1: self.in1.clone(),
            out1: self.out1.clone(),
            in2: self.in2.clone(),
            out2: self.out2.clone(),
        }
    }
}

impl<I1: 'static, O1: 'static, I2: 'static, O2: 'static> BidiShape<I1, O1, I2, O2> {
    #[must_use]
    pub fn new(in1: Inlet<I1>, out1: Outlet<O1>, in2: Inlet<I2>, out2: Outlet<O2>) -> Self {
        Self {
            in1,
            out1,
            in2,
            out2,
        }
    }

    #[must_use]
    pub fn in1(&self) -> Inlet<I1> {
        self.in1.clone()
    }

    #[must_use]
    pub fn out1(&self) -> Outlet<O1> {
        self.out1.clone()
    }

    #[must_use]
    pub fn in2(&self) -> Inlet<I2> {
        self.in2.clone()
    }

    #[must_use]
    pub fn out2(&self) -> Outlet<O2> {
        self.out2.clone()
    }

    #[must_use]
    pub fn from_flows<In1: 'static, Out1: 'static, In2: 'static, Out2: 'static>(
        top: &FlowShape<In1, Out1>,
        bottom: &FlowShape<In2, Out2>,
    ) -> BidiShape<In1, Out1, In2, Out2> {
        BidiShape::new(top.inlet(), top.outlet(), bottom.inlet(), bottom.outlet())
    }
}

impl<I1: 'static, O1: 'static, I2: 'static, O2: 'static> Shape for BidiShape<I1, O1, I2, O2> {
    fn inlets(&self) -> Vec<AnyInlet> {
        vec![self.in1.erase(), self.in2.erase()]
    }

    fn outlets(&self) -> Vec<AnyOutlet> {
        vec![self.out1.erase(), self.out2.erase()]
    }
}

/// Hands a [`GraphStage`](super::GraphStage) fresh, uniquely-identified ports
/// when its shape is allocated. Each `inlet`/`outlet` call draws a new global
/// [`PortId`](super::PortId); the allocator itself is stateless.
#[derive(Debug, Default)]
pub struct PortAllocator;

impl PortAllocator {
    #[must_use]
    pub fn inlet<T: 'static>(&mut self, name: impl Into<String>) -> Inlet<T> {
        Inlet::with_id(next_port_id(), name)
    }

    pub(super) fn inlet_arc<T: 'static>(&mut self, name: Arc<str>) -> Inlet<T> {
        Inlet::with_arc_name(next_port_id(), name)
    }

    #[must_use]
    pub fn outlet<T: 'static>(&mut self, name: impl Into<String>) -> Outlet<T> {
        Outlet::with_id(next_port_id(), name)
    }

    pub(super) fn outlet_arc<T: 'static>(&mut self, name: Arc<str>) -> Outlet<T> {
        Outlet::with_arc_name(next_port_id(), name)
    }
}