Upgrade pravega client

Signed-off-by: Luis Liu <[email protected]>

Author: vangork

Cargo.lock

@@ -33,9 +33,9 @@ gst-rtsp-server = { package = "gstreamer-rtsp-server", git = "https://gitlab.fre
 gst-sdp = { package = "gstreamer-sdp", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs" }
 gstreamer-video = { git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs" }
 gtk = { git = "https://github.com/gtk-rs/gtk-rs" }
-pravega-client = { git = "https://github.com/pravega/pravega-client-rust", rev = "08338688fbea790f9bc947cd8cfb9a90f13b6498" }
-pravega-client-config = { git = "https://github.com/pravega/pravega-client-rust", package = "pravega-client-config", rev = "08338688fbea790f9bc947cd8cfb9a90f13b6498" }
-pravega-client-shared = { git = "https://github.com/pravega/pravega-client-rust", package = "pravega-client-shared", rev = "08338688fbea790f9bc947cd8cfb9a90f13b6498" }
+pravega-client = { git = "https://github.com/pravega/pravega-client-rust", rev = "17deb48bbdb9b0180e93942d5e0e9218b553f77b" }
+pravega-client-config = { git = "https://github.com/pravega/pravega-client-rust", package = "pravega-client-config", rev = "17deb48bbdb9b0180e93942d5e0e9218b553f77b" }
+pravega-client-shared = { git = "https://github.com/pravega/pravega-client-rust", package = "pravega-client-shared", rev = "17deb48bbdb9b0180e93942d5e0e9218b553f77b" }
 pravega-video = { path = "../pravega-video" }
 log = "0.4"
 serde = "1"

apps/Cargo.toml

@@ -17,8 +17,8 @@ use pravega_client::client_factory::ClientFactory;
 use pravega_client_config::ClientConfigBuilder;
 use pravega_client_shared::{Scope, Stream, ScopedStream};
 use pravega_video::index::{IndexSearcher, SearchMethod, get_index_stream_name};
-use pravega_video::utils::parse_controller_uri;
 use pravega_video::timestamp::PravegaTimestamp;
+use pravega_video::utils::{parse_controller_uri, SyncByteReader};
 
 /// Tools to manage Pravega streams.
 #[derive(Clap)]
@@ -80,14 +80,14 @@ fn truncate_stream(controller: String, scope_name: String, stream_name: String,
         scope: scope.clone(),
         stream: stream.clone(),
     };
-    let writer = client_factory.create_byte_writer(scoped_stream);
+    let writer = runtime.block_on(client_factory.create_byte_writer(scoped_stream));
     let index_scoped_stream = ScopedStream {
         scope: scope.clone(),
         stream: index_stream.clone(),
     };
-    let index_writer = client_factory.create_byte_writer(index_scoped_stream.clone());
-    let index_reader = client_factory.create_byte_reader(index_scoped_stream.clone());
-    let mut index_searcher = IndexSearcher::new(index_reader);
+    let index_writer = runtime.block_on(client_factory.create_byte_writer(index_scoped_stream.clone()));
+    let index_reader = runtime.block_on(client_factory.create_byte_reader(index_scoped_stream.clone()));
+    let mut index_searcher = IndexSearcher::new(SyncByteReader::new(index_reader, client_factory.runtime_handle()));
     let index_record = index_searcher.search_timestamp_and_return_index_offset(
         truncate_at_timestamp, SearchMethod::Before).unwrap();
     println!("Truncating prior to {:?}", index_record);

apps/src/bin/pravega-tools.rs

@@ -12,7 +12,6 @@ use clap::Clap;
 use log::info;
 
 use std::convert::TryInto;
-use std::io::{Write};
 use uuid::Uuid;
 
 use pravega_client::client_factory::ClientFactory;
@@ -79,10 +78,9 @@ fn main() {
         let num_events: u64 = 3;
 
         if opts.use_byte_stream_writer {
-            let client_factory = client_factory.clone();
             let scoped_stream = scoped_stream.clone();
-            runtime.spawn_blocking(move || {
-                let mut writer = client_factory.create_byte_writer(scoped_stream);
+            let mut writer = client_factory.create_byte_writer(scoped_stream).await;
+            runtime.spawn(async move{
                 for i in 0..num_events {
                     let payload = format!("event {}", i).into_bytes();
                     let payload_length = payload.len();
@@ -92,7 +90,7 @@ fn main() {
                     bytes_to_write[4..8].copy_from_slice(&event_length.to_be_bytes()[..]);
                     bytes_to_write[8..8+payload_length].copy_from_slice(&payload[..]);
                     info!("bytes_to_write={:?}", bytes_to_write);
-                    writer.write_all(&bytes_to_write).unwrap();
+                    writer.write(&bytes_to_write).await.unwrap();
                 }
             });
         } else {
@@ -108,11 +106,11 @@ fn main() {
 
         // create event stream reader
         let reader_group_name = format!("rg{}", uuid::Uuid::new_v4()).to_string();
-        let rg = client_factory.create_reader_group(scope, reader_group_name, scoped_stream).await;
+        let rg = client_factory.create_reader_group(reader_group_name, scoped_stream).await;
         let mut reader = rg.create_reader("r1".to_string()).await;
 
         // read from segment
-        let mut slice = reader.acquire_segment().await.expect("acquire segment");
+        let mut slice = reader.acquire_segment().await.expect("acquire segment").unwrap();
         for i in 0..num_events {
             let read_event = slice.next();
             info!("read_event={:?}", read_event);

apps/src/bin/pravega_event_test1.rs

@@ -33,7 +33,7 @@ struct Opts {
     stream: String,
 }
 
-/// Demonstrate ability to write using the byte stream writer and read using the event reader.
+/// Demonstrate ability to write using the event stream writer and read using the event reader.
 fn main() {
     env_logger::init();
     let opts: Opts = Opts::parse();
@@ -96,11 +96,11 @@ fn main() {
     runtime.block_on(async {
         // create event stream reader
         let reader_group_name = format!("rg{}", uuid::Uuid::new_v4()).to_string();
-        let rg = client_factory.create_reader_group(scope, reader_group_name, scoped_stream).await;
+        let rg = client_factory.create_reader_group(reader_group_name, scoped_stream).await;
         let mut reader = rg.create_reader("r1".to_string()).await;
 
         // read from segment
-        let mut slice = reader.acquire_segment().await.expect("acquire segment");
+        let mut slice = reader.acquire_segment().await.expect("acquire segment").unwrap();
         for _ in 0..num_events {
             let read_event = slice.next();
             info!("read_event={:?}", read_event);

apps/src/bin/pravega_event_test2.rs

@@ -15,8 +15,9 @@ use std::{thread, time};
 use pravega_client::client_factory::ClientFactory;
 use pravega_client_shared::{Scope, Stream, ScopedStream};
 
+use pravega_video::index::IndexSearcher;
 use pravega_video::utils;
-use pravega_video::index::{IndexSearcher};
+use pravega_video::utils::SyncByteReader;
 
 #[derive(Clap)]
 struct Opts {
@@ -56,8 +57,9 @@ fn main() {
         scope: scope,
         stream: stream,
     };
-    let index_reader = client_factory.create_byte_reader(index_scoped_stream);
-    let mut index_searcher = IndexSearcher::new(index_reader);
+    let runtime = client_factory.runtime();
+    let index_reader = runtime.block_on(client_factory.create_byte_reader(index_scoped_stream));
+    let mut index_searcher = IndexSearcher::new(SyncByteReader::new(index_reader, client_factory.runtime_handle()));
     let check_period = time::Duration::from_secs(opts.check_period);
 
     info!("Checking period is {} seconds",  opts.check_period);

apps/src/bin/pravega_retention_checker.rs

@@ -18,9 +18,9 @@ edition = "2018"
 
 [dependencies]
 anyhow = "1"
-pravega-client = { git = "https://github.com/pravega/pravega-client-rust", rev = "08338688fbea790f9bc947cd8cfb9a90f13b6498" }
-pravega-client-config = { git = "https://github.com/pravega/pravega-client-rust", package = "pravega-client-config", rev = "08338688fbea790f9bc947cd8cfb9a90f13b6498" }
-pravega-client-shared = { git = "https://github.com/pravega/pravega-client-rust", package = "pravega-client-shared", rev = "08338688fbea790f9bc947cd8cfb9a90f13b6498" }
+pravega-client = { git = "https://github.com/pravega/pravega-client-rust", rev = "17deb48bbdb9b0180e93942d5e0e9218b553f77b" }
+pravega-client-config = { git = "https://github.com/pravega/pravega-client-rust", package = "pravega-client-config", rev = "17deb48bbdb9b0180e93942d5e0e9218b553f77b" }
+pravega-client-shared = { git = "https://github.com/pravega/pravega-client-rust", package = "pravega-client-shared", rev = "17deb48bbdb9b0180e93942d5e0e9218b553f77b" }
 pravega-video = { path = "../../pravega-video" }
 tracing = "0.1"
 tracing-subscriber = "0.2"

deepstream/pravega_protocol_adapter/Cargo.toml

@@ -21,6 +21,7 @@ use std::fmt;
 use std::os::raw::c_char;
 use std::ptr;
 use std::sync::{Arc, Once};
+use tokio::runtime::Handle;
 use tokio::sync::Mutex;
 use tracing::{debug, error, info, trace};
 use tracing_subscriber::fmt::format::FmtSpan;
@@ -134,16 +135,16 @@ impl fmt::Debug for EventWriterPool {
 }
 
 pub struct NvDsPravegaClientHandle {
-    pub client_factory: ClientFactory,
+    pub runtime_handle: Handle,
     pub writer_pool: EventWriterPool,
     pub routing_key_method: RoutingKeyMethod,
 }
 
 impl NvDsPravegaClientHandle {
     pub fn new(client_factory: ClientFactory, routing_key_method: RoutingKeyMethod) -> Self {
         NvDsPravegaClientHandle {
-            client_factory: client_factory.clone(),
-            writer_pool: EventWriterPool::new(client_factory.clone()),
+            runtime_handle: client_factory.runtime_handle(),
+            writer_pool: EventWriterPool::new(client_factory),
             routing_key_method,
         }
     }
@@ -255,13 +256,12 @@ pub extern "C" fn nvds_msgapi_send(h_ptr: *mut NvDsPravegaClientHandle, topic: *
     let payload_string = String::from_utf8_lossy(payload);
     trace!("nvds_msgapi_send: payload_string={}", payload_string);
     let scoped_stream = client_handle.resolve_topic(topic).unwrap();
-    let runtime = client_handle.client_factory.runtime();
     let routing_key_method = client_handle.routing_key_method.clone();
     let routing_key = match routing_key_method {
         RoutingKeyMethod::Fixed { routing_key } => routing_key,
     };
     debug!("nvds_msgapi_send: routing_key={:?}", routing_key);
-    let result = runtime.block_on(async {
+    let result = client_handle.runtime_handle.block_on(async {
         // Get a reference to the writer for this topic from the writer pool.
         let writer = client_handle.writer_pool.get_or_create(scoped_stream).await;
         // Get the mutex for this writer so we can use it.
@@ -312,15 +312,14 @@ pub extern "C" fn nvds_msgapi_send_async(
     // Convert unsafe payload to a vector. This also copies the payload which is critical to avoid memory corruption.
     let event = payload.to_vec();
     let scoped_stream = client_handle.resolve_topic(topic).unwrap();
-    let runtime = client_handle.client_factory.runtime();
     let routing_key_method = client_handle.routing_key_method.clone();
     let routing_key = match routing_key_method {
         RoutingKeyMethod::Fixed { routing_key } => routing_key,
     };
     debug!("nvds_msgapi_send_async: routing_key={:?}", routing_key);
     // Spawn a task in the Tokio runtime that will write the event, wait for it to be durably persisted,
     // and then call the callback function.
-    runtime.spawn(async move {
+    client_handle.runtime_handle.spawn(async move {
         // Get a reference to the writer for this topic from the writer pool.
         let writer = client_handle.writer_pool.get_or_create(scoped_stream).await;
         // Get the mutex for this writer so we can use it.

deepstream/pravega_protocol_adapter/src/lib.rs

@@ -23,12 +23,13 @@ glib = { git = "https://github.com/gtk-rs/gtk-rs" }
 gst = { package = "gstreamer", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs" }
 gst-base = { package = "gstreamer-base", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs" }
 once_cell = "1"
-pravega-client = { git = "https://github.com/pravega/pravega-client-rust", rev = "08338688fbea790f9bc947cd8cfb9a90f13b6498" }
-pravega-client-config = { git = "https://github.com/pravega/pravega-client-rust", package = "pravega-client-config", rev = "08338688fbea790f9bc947cd8cfb9a90f13b6498" }
-pravega-client-shared = { git = "https://github.com/pravega/pravega-client-rust", package = "pravega-client-shared", rev = "08338688fbea790f9bc947cd8cfb9a90f13b6498" }
+pravega-client = { git = "https://github.com/pravega/pravega-client-rust", rev = "17deb48bbdb9b0180e93942d5e0e9218b553f77b" }
+pravega-client-config = { git = "https://github.com/pravega/pravega-client-rust", package = "pravega-client-config", rev = "17deb48bbdb9b0180e93942d5e0e9218b553f77b" }
+pravega-client-shared = { git = "https://github.com/pravega/pravega-client-rust", package = "pravega-client-shared", rev = "17deb48bbdb9b0180e93942d5e0e9218b553f77b" }
 pravega-video = { path = "../pravega-video" }
 serde = "1"
 serde_json = "1"
+tokio = { version = "1", features = ["full"] }
 
 [lib]
 name = "gstpravega"