doc: add consume example (#29)

* doc: add consume example

* chore: add version

Author: alissa-tung

Cargo.lock

@@ -16,7 +16,7 @@ num-bigint = "0.4.3"
 num-traits = "0.2.15"
 md-5 = "0.10.1"
 
-hstreamdb-pb = { path = "../hstreamdb-pb/" }
+hstreamdb-pb = { version = "0.1.0", path = "../hstreamdb-pb/" }
 prost-types = "0.11.1"
 
 prost = "0.11.0"

src/hstreamdb/Cargo.toml

@@ -5,6 +5,7 @@ pub use hstreamdb_pb::{CompressionType, SpecialOffset, Stream};
 use num_bigint::ParseBigIntError;
 use tonic::transport;
 
+#[derive(Debug)]
 pub struct Subscription {
     pub subscription_id: String,
     pub stream_name: String,

src/hstreamdb/src/common.rs

@@ -1,6 +1,7 @@
 //! Rust client library for [HStreamDB](https://hstream.io/)
 
 //! ## Write Data to Streams
+//!
 //! ```
 //! use std::env;
 //!
@@ -67,8 +68,57 @@
 //!     Ok(())
 //! }
 //! ```
+//!
+//! ## Read Data from Subscriptions
+//!
+//! ```
+//! use std::env;
+//!
+//! use hstreamdb::client::Client;
+//! use hstreamdb::{SpecialOffset, Subscription};
+//! use tokio_stream::StreamExt;
+//!
+//! async fn consume_example() -> anyhow::Result<()> {
+//!     let addr = env::var("TEST_SERVER_ADDR").unwrap();
+//!     let mut client = Client::new(addr).await.unwrap();
+//!
+//!     let stream_name = "test_stream";
+//!     let subscription_id = "test_subscription";
+//!
+//!     client
+//!         .create_subscription(Subscription {
+//!             subscription_id: subscription_id.to_string(),
+//!             stream_name: stream_name.to_string(),
+//!             ack_timeout_seconds: 60 * 60,
+//!             max_unacked_records: 1000,
+//!             offset: SpecialOffset::Earliest,
+//!         })
+//!         .await?;
+//!     println!("{:?}", client.list_subscriptions().await?);
+//!
+//!     let mut stream = client
+//!         .streaming_fetch("test_consumer".to_string(), subscription_id.to_string())
+//!         .await
+//!         .unwrap();
+//!     let mut records = Vec::new();
+//!     while let Some((record, ack)) = stream.next().await {
+//!         println!("{record:?}");
+//!         records.push(record);
+//!         ack().unwrap();
+//!         if records.len() == 10 * 100 {
+//!             println!("done");
+//!             break;
+//!         }
+//!     }
+//!
+//!     client
+//!         .delete_subscription(subscription_id.to_string(), true)
+//!         .await?;
+//!
+//!     Ok(())
+//! }
+//! ```
 
-#![feature(try_blocks)]
 #![feature(box_syntax)]
 #![feature(default_free_fn)]
 
@@ -80,4 +130,6 @@ pub mod consumer;
 pub mod producer;
 pub mod utils;
 
-pub use common::{CompressionType, Error, Payload, Record, Result, Stream, Subscription};
+pub use common::{
+    CompressionType, Error, Payload, Record, Result, SpecialOffset, Stream, Subscription,
+};