Resolved issue #6: Stream query in release mode. (#7)

* Updated Readme.

Signed-off-by: Marvin Hansen <marvin.hansen@gmail.com>

* Enabled relase-plz

Signed-off-by: Marvin Hansen <marvin.hansen@gmail.com>

* Added client without compression to resolve issue with handling streaming data in release mode.

Added new method fetch_stream to ,well, fetch data from a continuous query of a stream.

Signed-off-by: Marvin Hansen <marvin.hansen@gmail.com>

* Fixed Docstring test in fetch_stream.

Signed-off-by: Marvin Hansen <marvin.hansen@gmail.com>

* Update src/lib/stream.rs

Co-authored-by: Gimi Liang <gimi.liang@timeplus.io>

* Update src/lib/mod.rs

Co-authored-by: Gimi Liang <gimi.liang@timeplus.io>

---------

Signed-off-by: Marvin Hansen <marvin.hansen@gmail.com>
Co-authored-by: Jove Zhong <jove@timeplus.io>
Co-authored-by: Gimi Liang <gimi.liang@timeplus.io>

Author: 3 people

src/lib/mod.rs

@@ -7,12 +7,17 @@ mod getters;
 mod insert;
 pub mod prelude;
 mod query;
+mod stream;
 
-use clickhouse::Client;
+use clickhouse::{Client, Compression};
 
 #[derive(Clone)]
 pub struct ProtonClient {
     client: Client,
+    // A separate client without compression is currently necessary to enable
+    // streaming queries in release builds. See issue #6 on Github:
+    // https://github.com/timeplus-io/proton-rust-client/issues/6
+    client_without_compression: Client,
     url: String,
 }
 
@@ -36,7 +41,13 @@ impl ProtonClient {
     /// ```
     pub fn new(url: &str) -> Self {
         let client = Client::default().with_url(url.to_string());
+        let client_without_compression =
+            Client::with_compression(client.clone(), Compression::None);
         let url = url.to_string();
-        Self { client, url }
+        Self {
+            client,
+            client_without_compression,
+            url,
+        }
     }
 }

src/lib/stream.rs

@@ -0,0 +1,51 @@
+use crate::prelude::{ProtonClient, ProtonClientError, Result};
+use clickhouse::query::RowCursor;
+use clickhouse::Row;
+use serde::Deserialize;
+
+impl ProtonClient {
+    ///
+    /// Executes a streaming query, returning a [`RowCursor`] to obtain results
+    /// as they become available from the stream. The key difference compared to fetch is that,
+    /// for streaming query, the returned result is a unbounded stream. Also,
+    /// a fetch_stream query will keep running continuously returning fresh data
+    /// until the application terminates..
+    ///
+    /// # Example
+    ///
+    /// ```no_run
+    ///  use proton_client::ProtonClient;
+    ///  use proton_client::prelude::Result;
+    ///
+    ///  async fn example() -> Result<()> {
+    ///
+    /// #[derive(Debug, clickhouse::Row, serde::Deserialize)]
+    /// struct MyRow {
+    ///     no: u32,
+    ///     name: String,
+    /// }
+    ///
+    /// let client = ProtonClient::new("http://localhost:3218");
+    ///
+    ///  let mut cursor = client
+    ///     .fetch_stream::<MyRow>("SELECT ?fields from (test_stream) WHERE no BETWEEN 500 AND 504")
+    ///     .await
+    ///     .expect("[main/fetch]: Failed to fetch stream data");
+    ///
+    /// while let Some(MyRow { name, no }) = cursor.next().await.expect("[main/fetch]: Failed to fetch data") {
+    ///     println!("{name}: {no}");
+    /// }
+    /// # Ok(()) }
+    /// ```
+    pub async fn fetch_stream<T>(&self, query: &str) -> Result<RowCursor<T>>
+    where
+        T: Row + for<'b> Deserialize<'b>,
+    {
+        // Here we use the client without compression. For details, see:
+        // https://github.com/timeplus-io/proton-rust-client/issues/6
+        match self.client_without_compression.query(query).fetch::<T>() {
+            Ok(cursor) => Ok(cursor),
+            Err(e) => Err(ProtonClientError::FetchFailed(e.to_string())),
+        }
+    }
+}