kafka_http/

lib.rs

use reqwest::Client;
use crate::api::{commit, create_consumer, poll, produce, subscribe};
use crate::error::Error;
use crate::error::ErrorType::InvalidInput;
use crate::types::{CreateConsumerParams, PartitionOffsetCommitParams, ProduceParams, Record, SubscribeParams};

pub(crate) mod api;
pub mod error;
pub mod types;


/// A client for interacting with Kafka REST Proxy over HTTP.
///
/// This client provides methods for creating consumers, subscribing to topics,
/// polling for records, producing messages, and committing offsets using the Kafka REST API.
///
/// # Examples
///
/// ```no_run
/// use kafka_http::KafkaHttpClient;
///
/// let mut client = KafkaHttpClient::new("http://localhost:8082");
/// client.set_timeout_ms(5000);
/// ```
#[derive(Debug, Clone)]
pub struct KafkaHttpClient {
    pub(crate) client: Client,
    pub(crate) base_uri: String,
    pub(crate) consumer_uri: Option<String>,
    pub(crate) timeout_ms: u64,
    pub(crate) target_topic: Option<String>
}

impl KafkaHttpClient {
    /// Creates a new `KafkaHttpClient` instance.
    ///
    /// # Arguments
    ///
    /// * `base_uri` - The base URI of the Kafka REST Proxy (e.g., "http://localhost:8082")
    ///
    /// # Returns
    ///
    /// A new `KafkaHttpClient` with default timeout of 1000ms
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use kafka_http::KafkaHttpClient;
    ///
    /// let client = KafkaHttpClient::new("http://localhost:8082");
    /// ```
    pub fn new(base_uri: &str) -> Self {
        Self {
            client: reqwest::Client::new(),
            base_uri: base_uri.to_string(),
            target_topic: None,
            consumer_uri: None,
            timeout_ms: 1000,
        }
    }

    /// Helper util to rebuild and set the uri if a conflict exists
    fn rebuild_consumer_uri(&self, group: &str, consumer_id: &str) -> String {
        // "http://localhost:18082/consumers/demo-group/instances/consumer-1"
        format!("{}/consumers/{group}/instances/{consumer_id}", self.base_uri)
    }

    /// Sets the consumer URI for this client.
    ///
    /// This is typically called internally after creating a consumer, but can be used
    /// to manually set a consumer URI if needed.
    ///
    /// # Arguments
    ///
    /// * `uri` - The consumer instance URI to set
    pub fn set_consumer_uri(&mut self, uri: &String) {
        tracing::debug!("Setting consumer URI to {uri}");
        self.consumer_uri = Some(uri.clone());
    }

    /// Sets the timeout duration for polling operations.
    ///
    /// # Arguments
    ///
    /// * `timeout_ms` - The timeout duration in milliseconds
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use kafka_http::KafkaHttpClient;
    ///
    /// let mut client = KafkaHttpClient::new("http://localhost:8082");
    /// client.set_timeout_ms(5000);  // Set 5 second timeout
    /// ```
    pub fn set_timeout_ms(&mut self, timeout_ms: u64) {
        tracing::debug!("Setting timeout to {timeout_ms} ms");
        self.timeout_ms = timeout_ms;
    }

    /// Creates a new consumer in the specified consumer group.
    ///
    /// This method will fail if the consumer already exists. Use `try_create_consumer`
    /// if you want to handle existing consumers gracefully.
    ///
    /// # Arguments
    ///
    /// * `group` - The consumer group name
    /// * `params` - Parameters for creating the consumer (name, format, etc.)
    ///
    /// # Returns
    ///
    /// * `Ok(String)` - The consumer instance URI on success
    /// * `Err(Error)` - An error if the consumer creation fails or URI is not returned
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use kafka_http::{KafkaHttpClient, types::CreateConsumerParams};
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let mut client = KafkaHttpClient::new("http://localhost:8082");
    /// let params = CreateConsumerParams {
    ///     name: "consumer-1".to_string(),
    ///     format: "json".to_string(),
    ///     ..Default::default()
    /// };
    /// let consumer_uri = client.create_consumer("my-group", &params).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn create_consumer(&mut self, group: &str, params: &CreateConsumerParams) -> Result<String, Error> {
        tracing::debug!("Creating consumer for group {group} - params: {params:?}");
        let try_consumer_uri = create_consumer(&self.client, &self.base_uri, group, params).await?;

        let Some(consumer_uri) = try_consumer_uri else {
            tracing::error!("failed to create consumer, consumer_uri is not set");
            return Err(Error::new(InvalidInput, "failed to create consumer, consumer_uri is not set"));
        };

        tracing::debug!("setting stored consumer_uri: {consumer_uri}");
        self.set_consumer_uri(&consumer_uri);
        Ok(consumer_uri.clone())
    }

    /// Attempts to create a new consumer, handling the case where the consumer already exists.
    ///
    /// If the consumer already exists, this method will reconstruct the consumer URI based on
    /// the group and consumer name, and return it wrapped in `Some`. This allows reconnecting
    /// to existing consumers.
    ///
    /// # Arguments
    ///
    /// * `group` - The consumer group name
    /// * `params` - Parameters for creating the consumer (name, format, etc.)
    ///
    /// # Returns
    ///
    /// * `Ok(Some(String))` - The consumer instance URI (new or reconstructed)
    /// * `Ok(None)` - Should not occur in current implementation
    /// * `Err(Error)` - An error if the operation fails
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use kafka_http::{KafkaHttpClient, types::CreateConsumerParams};
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let mut client = KafkaHttpClient::new("http://localhost:8082");
    /// let params = CreateConsumerParams {
    ///     name: "consumer-1".to_string(),
    ///     format: "json".to_string(),
    ///     ..Default::default()
    /// };
    /// // This won't fail if consumer already exists
    /// let consumer_uri = client.try_create_consumer("my-group", &params).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn try_create_consumer(&mut self, group: &str, params: &CreateConsumerParams) -> Result<Option<String>, Error> {
        tracing::debug!("Creating consumer for group {group} - params: {params:?}");
        let consumer_uri = create_consumer(&self.client, &self.base_uri, group, params).await?;
        match consumer_uri {
            Some(new_uri) => {
                tracing::debug!("setting stored consumer_uri: {new_uri}");
                self.set_consumer_uri(&new_uri);
                Ok(Some(new_uri))
            }
            None => {
                tracing::debug!("consumer already exists");
                let group = group.to_string();
                let consumer_id = &params.name;
                let built_consumer_uri = self.rebuild_consumer_uri(&group, consumer_id);

                tracing::debug!("setting stored consumer_uri: {built_consumer_uri}");

                self.set_consumer_uri(&built_consumer_uri);
                Ok(self.consumer_uri.clone())
            }
        }
    }

    /// Subscribes the consumer to one or more topics.
    ///
    /// A consumer must be created before calling this method. The consumer URI must be set
    /// (either by creating a consumer or manually setting it).
    ///
    /// # Arguments
    ///
    /// * `group` - The consumer group name (used for logging)
    /// * `params` - Subscription parameters including topics to subscribe to
    ///
    /// # Returns
    ///
    /// * `Ok(())` - Subscription was successful
    /// * `Err(Error)` - An error if consumer URI is not set or subscription fails
    ///
    /// # Examples    /// Polls for new records from subscribed topics.
    ///
    /// This method uses the timeout configured via `set_timeout_ms()`. A consumer must be
    /// created and subscribed to topics before polling.
    ///
    /// # Returns
    ///
    /// * `Ok(Vec<Record>)` - A vector of records (may be empty if no records available)
    /// * `Err(Error)` - An error if consumer URI is not set or polling fails
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use kafka_http::KafkaHttpClient;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let mut client = KafkaHttpClient::new("http://localhost:8082");
    /// // ... create consumer and subscribe ...
    /// let records = client.poll().await?;
    /// for record in records {
    ///     println!("Received: {:?}", record);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// ```no_run
    /// use kafka_http::{KafkaHttpClient, types::{CreateConsumerParams, SubscribeParams}};
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let mut client = KafkaHttpClient::new("http://localhost:8082");
    /// // ... create consumer first ...
    /// let subscribe_params = SubscribeParams {
    ///     topics: vec!["my-topic".to_string()],
    ///     ..Default::default()
    /// };
    /// client.subscribe("my-group", &subscribe_params).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn subscribe(&self, group: &str, params: &SubscribeParams) -> Result<(), Error> {
        let Some(consumer_uri) = &self.consumer_uri else {
            tracing::error!("Consumer URI is not set");
            return Err(Error::new(InvalidInput, "Consumer URI is not set"));
        };
        tracing::debug!("Subscribing to {group} - params: {params:?}");
        subscribe(&self.client, &consumer_uri, params).await
    }

    /// Polls for new records from subscribed topics.
    ///
    /// This method uses the timeout configured via `set_timeout_ms()`. A consumer must be
    /// created and subscribed to topics before polling.
    ///
    /// # Returns
    ///
    /// * `Ok(Vec<Record>)` - A vector of records (may be empty if no records available)
    /// * `Err(Error)` - An error if consumer URI is not set or polling fails
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use kafka_http::KafkaHttpClient;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let mut client = KafkaHttpClient::new("http://localhost:8082");
    /// // ... create consumer and subscribe ...
    /// let records = client.poll().awaait?;
    /// for record in records {
    ///     println!("Received: {:?}", record);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn poll(&self) -> Result<Vec<Record>, Error> {
        let Some(consumer_uri) = &self.consumer_uri else {
            tracing::error!("Consumer URI is not set");
            return Err(Error::new(InvalidInput, "Consumer URI is not set"));
        };
        tracing::debug!("Polling");
        poll(&self.client, &consumer_uri, self.timeout_ms).await
    }

    /// Produces records to a Kafka topic.
    ///
    /// # Arguments
    ///
    /// * `topic` - The name of the topic to produce to
    /// * `params` - Parameters containing the records to produce
    ///
    /// # Returns
    ///
    /// * `Ok(())` - Records were successfully produced
    /// * `Err(Error)` - An error if production fails
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use kafka_http::{KafkaHttpClient, types::ProduceParams};
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let client = KafkaHttpClient::new("http://localhost:8082");
    /// let params = ProduceParams {
    ///     records: vec![/* records */],
    ///     ..Default::default()
    /// };
    /// client.produce_to_topic("my-topic", &params).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn produce_to_topic(&self, topic: &str, params: &ProduceParams) -> Result<(), Error> {
        tracing::debug!("producing records");
        produce(&self.client, &self.base_uri, topic, params).await
    }

    /// Sets the target topic for subsequent produce operations.
    ///
    /// This allows using the `produce()` method without specifying a topic each time.
    /// The topic will be used for all subsequent calls to `produce()` until changed.
    ///
    /// # Arguments
    ///
    /// * `topic` - The name of the topic to use for future produce operations
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use kafka_http::KafkaHttpClient;
    ///
    /// let mut client = KafkaHttpClient::new("http://localhost:8082");
    /// client.set_target_topic("my-topic");
    /// // Now can use produce() without specifying topic
    /// ```
    pub fn set_target_topic(&mut self, topic: &str) {
        self.target_topic = Some(topic.to_string());
    }

    /// Produces records to the previously set target topic.
    ///
    /// This is a convenience method that uses the topic set via `set_target_topic()`.
    /// If no target topic has been set, this method will return an error.
    ///
    /// # Arguments
    ///
    /// * `params` - Parameters containing the records to produce
    ///
    /// # Returns
    ///
    /// * `Ok(())` - Records were successfully produced
    /// * `Err(Error)` - An error if target topic is not set or production fails
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use kafka_http::{KafkaHttpClient, types::ProduceParams};
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let mut client = KafkaHttpClient::new("http://localhost:8082");
    /// client.set_target_topic("my-topic");
    ///
    /// let params = ProduceParams {
    ///     records: vec![/* records */],
    ///     ..Default::default()
    /// };
    /// client.produce(&params).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn produce(&self, params: &ProduceParams) -> Result<(), Error> {
        let Some(topic) = &self.target_topic else {
            tracing::error!("Target topic is not set");
            return Err(Error::new(InvalidInput, "Target topic is not set"));
        };
        tracing::debug!("producing records to topic: {topic}");
        produce(&self.client, &self.base_uri, topic, params).await
    }
    /// Commits offsets for consumed messages.
    ///
    /// A consumer must be created before calling this method. The consumer URI must be set.
    ///
    /// # Arguments
    ///
    /// * `params` - Parameters specifying which partition offsets to commit
    ///
    /// # Returns
    ///
    /// * `Ok(())` - Offsets were successfully committed
    /// * `Err(Error)` - An error if consumer URI is not set or commit fails
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use kafka_http::{KafkaHttpClient, types::PartitionOffsetCommitParams};
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let mut client = KafkaHttpClient::new("http://localhost:8082");
    /// // ... create consumer, subscribe, and poll ...
    /// let params = PartitionOffsetCommitParams {
    ///     partitions: vec![/* partition offsets */],
    ///     ..Default::default()
    /// };
    /// client.commit(&params).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn commit(&self, params: &PartitionOffsetCommitParams) -> Result<(), Error> {
        let Some(consumer_uri) = &self.consumer_uri else {
            tracing::error!("Consumer URI is not set");
            return Err(Error::new(InvalidInput, "Consumer URI is not set"));
        };
        tracing::debug!("commiting offsets");
        commit(&self.client, &consumer_uri, params).await
    }
}