Add polling strategy parameter to poll_messages (#28)

Closes #23

python_examples/consumer.py

@@ -2,7 +2,7 @@
 from loguru import logger
 
 # Assuming there's a Python module for iggy with similar functionalities.
-from iggy_py import IggyClient, ReceiveMessage
+from iggy_py import IggyClient, ReceiveMessage, PollingStrategy
 
 STREAM_NAME = "sample-stream"
 TOPIC_NAME = "sample-topic"
@@ -36,6 +36,7 @@ async def consume_messages(client: IggyClient):
                 stream=STREAM_NAME,
                 topic=TOPIC_NAME,
                 partition_id=PARTITION_ID,
+                polling_strategy=PollingStrategy.Next(),
                 count=messages_per_batch,
                 auto_commit=True
             )

src/client.rs

@@ -7,15 +7,15 @@ use iggy::clients::client::IggyClient as RustIggyClient;
 use iggy::compression::compression_algorithm::CompressionAlgorithm;
 use iggy::consumer::Consumer as RustConsumer;
 use iggy::identifier::Identifier;
-use iggy::messages::poll_messages::PollingStrategy;
+use iggy::messages::poll_messages::PollingStrategy as RustPollingStrategy;
 use iggy::messages::send_messages::{Message as RustMessage, Partitioning};
 use iggy::utils::expiry::IggyExpiry;
 use iggy::utils::topic_size::MaxTopicSize;
 use pyo3::prelude::*;
 use pyo3::types::PyList;
 use tokio::runtime::{Builder, Runtime};
 
-use crate::receive_message::ReceiveMessage;
+use crate::receive_message::{PollingStrategy, ReceiveMessage};
 use crate::send_message::SendMessage;
 use crate::stream::StreamDetails;
 use crate::topic::TopicDetails;
@@ -164,7 +164,11 @@ impl IggyClient {
     /// Gets topic by stream and id.
     ///
     /// Returns Option of topic details or a PyRuntimeError on failure.
-    fn get_topic(&self, stream_id: PyIdentifier, topic_id: PyIdentifier) -> PyResult<Option<TopicDetails>> {
+    fn get_topic(
+        &self,
+        stream_id: PyIdentifier,
+        topic_id: PyIdentifier,
+    ) -> PyResult<Option<TopicDetails>> {
         let stream_id = Identifier::from(stream_id);
         let topic_id = Identifier::from(topic_id);
         let topic_future = self.inner.get_topic(&stream_id, &topic_id);
@@ -215,13 +219,14 @@ impl IggyClient {
         stream: PyIdentifier,
         topic: PyIdentifier,
         partition_id: u32,
+        polling_strategy: &PollingStrategy,
         count: u32,
         auto_commit: bool,
     ) -> PyResult<Vec<ReceiveMessage>> {
         let consumer = RustConsumer::default();
         let stream = Identifier::from(stream);
         let topic = Identifier::from(topic);
-        let strategy = PollingStrategy::next();
+        let strategy: RustPollingStrategy = (*polling_strategy).into();
 
         let poll_messages = self.inner.poll_messages(
             &stream,

src/lib.rs

@@ -6,7 +6,7 @@ mod topic;
 
 use client::IggyClient;
 use pyo3::prelude::*;
-use receive_message::ReceiveMessage;
+use receive_message::{PollingStrategy, ReceiveMessage};
 use send_message::SendMessage;
 use stream::StreamDetails;
 use topic::TopicDetails;
@@ -19,5 +19,6 @@ fn iggy_py(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<IggyClient>()?;
     m.add_class::<StreamDetails>()?;
     m.add_class::<TopicDetails>()?;
+    m.add_class::<PollingStrategy>()?;
     Ok(())
 }

src/receive_message.rs

@@ -1,3 +1,4 @@
+use iggy::messages::poll_messages::PollingStrategy as RustPollingStrategy;
 use iggy::models::messages::PolledMessage as RustReceiveMessage;
 use pyo3::prelude::*;
 use pyo3::types::PyBytes;
@@ -35,3 +36,25 @@ impl ReceiveMessage {
         self.inner.offset
     }
 }
+
+#[derive(Clone, Copy)]
+#[pyclass]
+pub enum PollingStrategy {
+    Offset { value: u64 },
+    Timestamp { value: u64 },
+    First {},
+    Last {},
+    Next {},
+}
+
+impl From<PollingStrategy> for RustPollingStrategy {
+    fn from(value: PollingStrategy) -> Self {
+        match value {
+            PollingStrategy::Offset { value } => RustPollingStrategy::offset(value),
+            PollingStrategy::Timestamp { value } => RustPollingStrategy::timestamp(value.into()),
+            PollingStrategy::First {} => RustPollingStrategy::first(),
+            PollingStrategy::Last {} => RustPollingStrategy::last(),
+            PollingStrategy::Next {} => RustPollingStrategy::next(),
+        }
+    }
+}

src/stream.rs

@@ -8,7 +8,7 @@ pub struct StreamDetails {
 
 impl From<RustStreamDetails> for StreamDetails {
     fn from(stream_details: RustStreamDetails) -> Self {
-        StreamDetails {
+        Self {
             inner: stream_details,
         }
     }

src/topic.rs

@@ -8,7 +8,7 @@ pub struct TopicDetails {
 
 impl From<RustTopicDetails> for TopicDetails {
     fn from(topic_details: RustTopicDetails) -> Self {
-        TopicDetails {
+        Self {
             inner: topic_details,
         }
     }
@@ -36,4 +36,3 @@ impl TopicDetails {
         self.inner.partitions_count
     }
 }
-