From f9a55a2b2152ac6621f45765e8ce2a4c4df6147c Mon Sep 17 00:00:00 2001
From: Arpad Borsos <arpad.borsos@sentry.io>
Date: Tue, 19 Dec 2023 16:13:51 +0100
Subject: [PATCH] Make `poll_queue` non-generic

This eagerly resolves the `Into<Timeout>`, making the function non-generic, which can theoretically help with compile times.
---
 src/consumer/base_consumer.rs   | 9 ++++-----
 src/consumer/stream_consumer.rs | 5 +++--
 2 files changed, 7 insertions(+), 7 deletions(-)

diff --git a/src/consumer/base_consumer.rs b/src/consumer/base_consumer.rs
index 08ec51b78..eaab7520c 100644
--- a/src/consumer/base_consumer.rs
+++ b/src/consumer/base_consumer.rs
@@ -115,16 +115,15 @@ where
     ///
     /// The returned message lives in the memory of the consumer and cannot outlive it.
     pub fn poll<T: Into<Timeout>>(&self, timeout: T) -> Option<KafkaResult<BorrowedMessage<'_>>> {
-        self.poll_queue(self.get_queue(), timeout)
+        self.poll_queue(self.get_queue(), timeout.into())
     }
 
-    pub(crate) fn poll_queue<T: Into<Timeout>>(
+    pub(crate) fn poll_queue(
         &self,
         queue: &NativeQueue,
-        timeout: T,
+        mut timeout: Timeout,
     ) -> Option<KafkaResult<BorrowedMessage<'_>>> {
         let now = Instant::now();
-        let mut timeout = timeout.into();
         let min_poll_interval = self.context().main_queue_min_poll_interval();
         loop {
             let op_timeout = std::cmp::min(timeout, min_poll_interval);
@@ -795,7 +794,7 @@ where
     /// associated consumer regularly, even if no messages are expected, to
     /// serve events.
     pub fn poll<T: Into<Timeout>>(&self, timeout: T) -> Option<KafkaResult<BorrowedMessage<'_>>> {
-        self.consumer.poll_queue(&self.queue, timeout)
+        self.consumer.poll_queue(&self.queue, timeout.into())
     }
 
     /// Sets a callback that will be invoked whenever the queue becomes
diff --git a/src/consumer/stream_consumer.rs b/src/consumer/stream_consumer.rs
index 5a7f60552..f4a50eadc 100644
--- a/src/consumer/stream_consumer.rs
+++ b/src/consumer/stream_consumer.rs
@@ -123,10 +123,11 @@ impl<'a, C: ConsumerContext> MessageStream<'a, C> {
     }
 
     fn poll(&self) -> Option<KafkaResult<BorrowedMessage<'a>>> {
+        let timeout: Timeout = Duration::ZERO.into();
         if let Some(queue) = self.partition_queue {
-            self.consumer.poll_queue(queue, Duration::ZERO)
+            self.consumer.poll_queue(queue, timeout)
         } else {
-            self.consumer.poll(Duration::ZERO)
+            self.consumer.poll(timeout)
         }
     }
 }