ruststream-rdkafka 0.7.1

//! Named partitions of one topic, assigned without joining a consumer group.

#[cfg(feature = "asyncapi")]
use ruststream::asyncapi::Bindings;
use ruststream::runtime::IntoSource;
use ruststream::{NamedCopies, SubscriptionSource};

use super::{Commit, GroupSettings, LaneKey, Reader, StartOffset, SubscriptionPlan};
use crate::broker::ConnectedKafkaBroker;
use crate::error::KafkaError;
use crate::subscriber::KafkaSubscriber;

/// A subscription that takes exactly the partitions it names, without joining a consumer group
/// and without rebalancing.
///
/// That fits a reader pinned to a partition, an inspection or replay tool, and a
/// one-consumer-per-partition deployment. A group stays optional and changes only how offsets
/// are handled: with one named the consumer commits into the group without joining it, and
/// without one commits are off, so the start offset has to be explicit.
///
/// This form reads a subset of a topic, and the partitioner places a published record wherever
/// the key sends it, so no publish addresses it: a registration that retries names the
/// destination itself with `.out_retry(policy).to("orders.retry")`. Reading the whole topic is
/// [`KafkaTopic`](crate::KafkaTopic), which addresses its own copies.
///
/// # Examples
///
/// An inspection reader pinned to partition 0, with no group side effects:
///
/// ```
/// use ruststream_rdkafka::context::keys::Offset;
/// use ruststream_rdkafka::prelude::*;
/// # #[derive(serde::Deserialize)]
/// # struct Order { id: u64 }
///
/// #[subscriber(KafkaPartitions::new("orders", [0]).start(StartOffset::Earliest))]
/// async fn inspect(order: &Order, Ctx(offset): Ctx<Offset>) -> HandlerOutcome {
///     tracing::info!(order.id, %offset, "read from partition 0");
///     HandlerOutcome::ack()
/// }
///
/// fn app() -> RustStream {
///     RustStream::new(AppInfo::new("inspector", "0.1.0"))
///         .with_broker(KafkaBroker::new(["localhost:9092"]), |b| {
///             b.include(inspect);
///         })
/// }
/// # let _ = app;
/// ```
#[derive(Debug, Clone)]
pub struct KafkaPartitions {
    topic: String,
    partitions: Vec<i32>,
    settings: GroupSettings,
}

impl KafkaPartitions {
    /// Describes a reader over exactly `partitions` of `topic`.
    #[must_use]
    pub fn new(topic: impl Into<String>, partitions: impl IntoIterator<Item = i32>) -> Self {
        Self {
            topic: topic.into(),
            partitions: partitions.into_iter().collect(),
            settings: GroupSettings::default(),
        }
    }

    /// The consumer group whose committed offsets this reader uses, overriding
    /// [`KafkaBroker::default_group`](crate::KafkaBroker::default_group). The reader commits
    /// into the group without ever joining it.
    #[must_use]
    pub fn group(mut self, group: impl Into<String>) -> Self {
        self.settings.group = Some(group.into());
        self
    }

    /// Where the reader starts (see [`StartOffset`]). Without a group,
    /// [`StartOffset::Committed`] has nothing to resume from and is a startup error.
    #[must_use]
    pub fn start(mut self, start: StartOffset) -> Self {
        self.settings.start = start;
        self
    }

    /// How processed deliveries are committed (see [`Commit`]).
    /// [`Commit::Transactional`] is a startup error here: an exactly-once pipeline commits
    /// through the consumer-group protocol this form does not take part in.
    #[must_use]
    pub fn commit(mut self, commit: Commit) -> Self {
        self.settings.commit = commit;
        self
    }

    /// What drives keyed worker lanes for this reader (see [`LaneKey`]). Under the default each
    /// assigned partition gets a lane of its own.
    #[must_use]
    pub fn lane_key(mut self, lane_key: LaneKey) -> Self {
        self.settings.lane_key = lane_key;
        self
    }

    /// Raw librdkafka consumer property passthrough, applied last (it wins over the typed
    /// options and the broker-wide config).
    ///
    /// A property the subscription's [`Commit`] mode owns is the exception; see
    /// [`KafkaTopic::config`](crate::KafkaTopic::config).
    #[must_use]
    pub fn config(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
        self.settings.config.set(key.into(), value.into());
        self
    }

    /// The topic these partitions belong to, which is also the handler metadata name.
    #[must_use]
    pub fn topic(&self) -> &str {
        &self.topic
    }

    /// The partitions this reader takes.
    #[must_use]
    pub fn partitions(&self) -> &[i32] {
        &self.partitions
    }

    /// What this reader adds to its channel in the generated `AsyncAPI` document: the Kafka
    /// topic its partitions belong to.
    #[cfg(feature = "asyncapi")]
    fn channel_bindings(&self) -> Bindings {
        crate::bindings::channel(&self.topic)
    }

    /// What this reader adds to its `receive` operation: the group it commits into, and the
    /// client id where the raw passthrough names one.
    #[cfg(feature = "asyncapi")]
    fn operation_bindings(&self) -> Bindings {
        crate::bindings::operation(self.settings.group.as_deref(), &self.settings.config)
    }

    fn into_plan(self) -> Result<SubscriptionPlan, KafkaError> {
        super::reject_empty(&self.topic)?;
        super::reject_pattern(&self.topic, "`KafkaPartitions`")?;
        if self.partitions.is_empty() {
            return Err(KafkaError::InvalidOptions(format!(
                "manual assignment of {:?} names no partition, so the reader would take \
                 nothing; name the partitions it reads",
                self.topic,
            )));
        }
        super::reject_commit_mode_clash(&self.topic, &self.settings.commit, &self.settings.config)?;
        Ok(SubscriptionPlan {
            name: self.topic.clone(),
            reader: Reader::Assigned {
                topic: self.topic,
                partitions: self.partitions,
            },
            settings: self.settings,
        })
    }
}

// The manual path binds a `Handle` body to its source with `subscriber(source, body)`, which
// takes the descriptor as it is, the way the `#[subscriber(..)]` attribute does.
impl IntoSource for KafkaPartitions {
    type Source = Self;

    fn into_source(self) -> Self {
        self
    }
}

impl SubscriptionSource<ConnectedKafkaBroker> for KafkaPartitions {
    type Subscriber = KafkaSubscriber;
    type Copies = NamedCopies;

    fn name(&self) -> &str {
        &self.topic
    }

    async fn subscribe(
        self,
        connected: &ConnectedKafkaBroker,
    ) -> Result<Self::Subscriber, KafkaError> {
        connected.open(self.into_plan()?).await
    }

    #[cfg(feature = "asyncapi")]
    fn channel_bindings(&self) -> Bindings {
        Self::channel_bindings(self)
    }

    #[cfg(feature = "asyncapi")]
    fn operation_bindings(&self) -> Bindings {
        Self::operation_bindings(self)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn a_reader_plans_the_partitions_it_names() {
        let plan = KafkaPartitions::new("orders", [0, 2])
            .start(StartOffset::Earliest)
            .into_plan()
            .expect("two partitions of one topic are a valid assignment");
        assert_eq!(plan.name, "orders");
        let Reader::Assigned { topic, partitions } = plan.reader else {
            panic!("a manual assignment must plan an assigned reader");
        };
        assert_eq!(topic, "orders");
        assert_eq!(partitions, [0, 2]);
    }

    #[test]
    fn naming_no_partition_is_refused() {
        let err = KafkaPartitions::new("orders", [])
            .into_plan()
            .expect_err("a reader over no partition reads nothing");
        assert!(matches!(err, KafkaError::InvalidOptions(_)));
    }
}