Use default buffer size for old APIs

Also adding test case

Author: erikbosch

databroker/src/broker.rs

@@ -4241,8 +4241,7 @@ pub mod tests {
         }
     }
 
-    #[tokio::test]
-    async fn test_subscribe_and_get() {
+    async fn test_subscribe_and_get_buffer_size(buffer_size: Option<usize>) {
         let broker = DataBroker::default();
         let broker = broker.authorized_access(&permissions::ALLOW_ALL);
 
@@ -4264,7 +4263,7 @@ pub mod tests {
         let mut stream = broker
             .subscribe(
                 HashMap::from([(id1, HashSet::from([Field::Datapoint]))]),
-                None,
+                buffer_size,
             )
             .await
             .expect("subscription should succeed");
@@ -4342,6 +4341,49 @@ pub mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn test_subscribe_and_get() {
+        // None and 0-1000 is valid range
+        test_subscribe_and_get_buffer_size(None).await;
+        test_subscribe_and_get_buffer_size(Some(0)).await;
+        test_subscribe_and_get_buffer_size(Some(1000)).await;
+    }
+
+    #[tokio::test]
+    async fn test_subscribe_buffersize_out_of_range() {
+        let broker = DataBroker::default();
+        let broker = broker.authorized_access(&permissions::ALLOW_ALL);
+
+        let id1 = broker
+            .add_entry(
+                "test.datapoint1".to_owned(),
+                DataType::Int32,
+                ChangeType::OnChange,
+                EntryType::Sensor,
+                "Test datapoint 1".to_owned(),
+                None, // min
+                None, // max
+                None,
+                None,
+            )
+            .await
+            .expect("Register datapoint should succeed");
+
+        match broker
+            .subscribe(
+                HashMap::from([(id1, HashSet::from([Field::Datapoint]))]),
+                // 1001 is just outside valid range 0-1000
+                Some(1001),
+            )
+            .await
+        {
+            Err(SubscriptionError::InvalidBufferSize) => {}
+            _ => {
+                panic!("expected it to fail with InvalidBufferSize");
+            }
+        }
+    }
+
     #[tokio::test]
     async fn test_metadata_for_each() {
         let db = DataBroker::default();

databroker/src/grpc/kuksa_val_v1/val.rs

@@ -600,7 +600,7 @@ impl proto::val_server::Val for broker::DataBroker {
             }
         }
 
-        match broker.subscribe(entries, Some(1)).await {
+        match broker.subscribe(entries, None).await {
             Ok(stream) => {
                 let stream = convert_to_proto_stream(stream);
                 Ok(tonic::Response::new(Box::pin(stream)))

databroker/src/viss/v2/server.rs

@@ -271,7 +271,7 @@ impl Viss for Server {
             });
         };
 
-        match broker.subscribe(entries, Some(1)).await {
+        match broker.subscribe(entries, None).await {
             Ok(stream) => {
                 let subscription_id = SubscriptionId::new();