feat: add receiver queue size configuration to consumer (#371)

Author: crossoverJie

src/consumer/mod.rs

@@ -563,6 +563,8 @@ mod tests {
             .unwrap();
 
         assert!(&consumer_1.check_connection().await.is_ok());
+        let receive_queue_size = consumer_1.options().receiver_queue_size;
+        assert_eq!(receive_queue_size, None);
 
         let mut consumer_2: Consumer<TestData, _> = builder
             .clone()
@@ -677,6 +679,40 @@ mod tests {
         assert!(consumer_1_exclusive.is_ok());
     }
 
+    #[tokio::test]
+    #[cfg(any(
+        feature = "tokio-runtime",
+        feature = "tokio-rustls-runtime-aws-lc-rs",
+        feature = "tokio-rustls-runtime-ring"
+    ))]
+    async fn consumer_zero_receiver_queue_size() {
+        let _result = log::set_logger(&TEST_LOGGER);
+        log::set_max_level(LevelFilter::Debug);
+        let addr = "pulsar://127.0.0.1:6650";
+        let topic = format!(
+            "consumer_zero_receiver_queue_size_{}",
+            rand::random::<u16>()
+        );
+
+        let client: Pulsar<_> = Pulsar::builder(addr, TokioExecutor).build().await.unwrap();
+        let consumer: Consumer<TestData, _> = client
+            .consumer()
+            .with_topic(&topic)
+            .with_subscription("dropped_ack")
+            .with_subscription_type(SubType::Shared)
+            // get earliest messages
+            .with_options(
+                ConsumerOptions::default()
+                    .with_receiver_queue_size(0)
+                    .with_initial_position(InitialPosition::Earliest),
+            )
+            .build()
+            .await
+            .unwrap();
+        let size = consumer.options().receiver_queue_size.unwrap();
+        assert_eq!(size, 1000);
+    }
+
     #[tokio::test]
     #[cfg(any(
         feature = "tokio-runtime",
@@ -715,15 +751,18 @@ mod tests {
                 .with_subscription("dropped_ack")
                 .with_subscription_type(SubType::Shared)
                 // get earliest messages
-                .with_options(ConsumerOptions {
-                    initial_position: InitialPosition::Earliest,
-                    ..Default::default()
-                })
+                .with_options(
+                    ConsumerOptions::default()
+                        .with_receiver_queue_size(2000)
+                        .with_initial_position(InitialPosition::Earliest),
+                )
                 .build()
                 .await
                 .unwrap();
 
             println!("created consumer");
+            let receive_queue_size = consumer.options().receiver_queue_size;
+            assert_eq!(receive_queue_size, Some(2000));
 
             // consumer.next().await
             let msg: Message<TestData> = timeout(Duration::from_secs(1), consumer.next())

src/consumer/options.rs

@@ -20,6 +20,8 @@ pub struct ConsumerOptions {
     pub metadata: BTreeMap<String, String>,
     pub read_compacted: Option<bool>,
     pub schema: Option<Schema>,
+    /// size of the receiver queue
+    pub receiver_queue_size: Option<u32>,
     /// Signal whether the subscription will initialize on latest
     /// or earliest message (default on latest)
     ///
@@ -76,4 +78,11 @@ impl ConsumerOptions {
         self.initial_position = initial_position;
         self
     }
+
+    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
+    pub fn with_receiver_queue_size(mut self, size: u32) -> Self {
+        // todo: support zero_queue_size consumer
+        self.receiver_queue_size = Some(if size == 0 { 1000 } else { size });
+        self
+    }
 }

src/consumer/topic.rs

@@ -103,7 +103,8 @@ impl<T: DeserializeMessage, Exe: Executor> TopicConsumer<T, Exe> {
                 return Err(Error::Executor);
             }
         }
-        let (tx, rx) = mpsc::channel(1000);
+        let receiver_queue_size = options.receiver_queue_size.unwrap_or(1000);
+        let (tx, rx) = mpsc::channel(receiver_queue_size as usize);
         let mut c = ConsumerEngine::new(
             client.clone(),
             connection.clone(),