fix(client): align with the server contract

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>

Author: yordis

trogon-eventstore/protos/streams.proto

@@ -105,34 +105,20 @@ message ReadResp {
 
 	// The $all or stream subscription has caught up and become live.
 	message CaughtUp {
-		// Current time in the server when the subscription caught up
-		google.protobuf.Timestamp timestamp = 1;
-
-		// Checkpoint for resuming a stream subscription.
-		// For stream subscriptions it is populated unless the stream is empty.
-		// For $all subscriptions it is not populated.
-		optional int64 stream_revision = 2;
-
-		// Checkpoint for resuming a $all subscription.
-		// For stream subscriptions it is not populated.
-		// For $all subscriptions it is populated unless the database is empty.
-		optional Position position = 3;
+		oneof position {
+			uint64 stream_position = 1;
+			event_store.client.AllStreamPosition all_stream_position = 2;
+			event_store.client.Empty no_position = 3;
+		}
 	}
 
 	// The $all or stream subscription has fallen back into catchup mode and is no longer live.
 	message FellBehind {
-		// Current time in the server when the subscription fell behind
-		google.protobuf.Timestamp timestamp = 1;
-
-		// Checkpoint for resuming a stream subscription.
-		// For stream subscriptions it is populated unless the stream is empty.
-		// For $all subscriptions it is not populated.
-		optional int64 stream_revision = 2;
-
-		// Checkpoint for resuming a $all subscription.
-		// For stream subscriptions it is not populated.
-		// For $all subscriptions it is populated unless the database is empty.
-		optional Position position = 3;
+		oneof position {
+			uint64 stream_position = 1;
+			event_store.client.AllStreamPosition all_stream_position = 2;
+			event_store.client.Empty no_position = 3;
+		}
 	}
 
 	message ReadEvent {
@@ -165,11 +151,6 @@ message ReadResp {
 		google.protobuf.Timestamp timestamp = 3;
 	}
 
-	message Position {
-		uint64 commit_position = 1;
-		uint64 prepare_position = 2;
-	}
-
 	message StreamNotFound {
 		event_store.client.StreamIdentifier stream_identifier = 1;
 	}

trogon-eventstore/src/commands.rs

@@ -1,9 +1,8 @@
 #![allow(clippy::large_enum_variant)]
 //! Commands this client supports.
 use std::ops::Add;
-use std::time::{Duration, SystemTime, UNIX_EPOCH};
+use std::time::{Duration, SystemTime};
 
-use chrono::{DateTime, Utc};
 use futures::TryStreamExt;
 use nom::AsBytes;
 use tokio::sync::mpsc;
@@ -840,26 +839,46 @@ impl Subscription {
                                 }
 
                                 streams::read_resp::Content::CaughtUp(args) => {
-                                    let args = args.timestamp.map(|t| crate::CaughtUp {
-                                        date: timestamp_to_datetime(t),
-                                        stream_revision: args.stream_revision.map(|x| x as u64),
-                                        position: args.position.map(|x| Position {
-                                            commit: x.commit_position,
-                                            prepare: x.prepare_position,
-                                        }),
+                                    use streams::read_resp::caught_up::Position::*;
+                                    let args = args.position.map(|position| match position {
+                                        StreamPosition(revision) => crate::CaughtUp {
+                                            stream_revision: Some(revision),
+                                            position: None,
+                                        },
+                                        AllStreamPosition(position) => crate::CaughtUp {
+                                            stream_revision: None,
+                                            position: Some(Position {
+                                                commit: position.commit_position,
+                                                prepare: position.prepare_position,
+                                            }),
+                                        },
+                                        NoPosition(()) => crate::CaughtUp {
+                                            stream_revision: None,
+                                            position: None,
+                                        },
                                     });
 
                                     return Ok(SubscriptionEvent::CaughtUp(args));
                                 }
 
                                 streams::read_resp::Content::FellBehind(args) => {
-                                    let args = args.timestamp.map(|t| crate::FellBehind {
-                                        date: timestamp_to_datetime(t),
-                                        stream_revision: args.stream_revision.map(|x| x as u64),
-                                        position: args.position.map(|x| Position {
-                                            commit: x.commit_position,
-                                            prepare: x.prepare_position,
-                                        }),
+                                    use streams::read_resp::fell_behind::Position::*;
+                                    let args = args.position.map(|position| match position {
+                                        StreamPosition(revision) => crate::FellBehind {
+                                            stream_revision: Some(revision),
+                                            position: None,
+                                        },
+                                        AllStreamPosition(position) => crate::FellBehind {
+                                            stream_revision: None,
+                                            position: Some(Position {
+                                                commit: position.commit_position,
+                                                prepare: position.prepare_position,
+                                            }),
+                                        },
+                                        NoPosition(()) => crate::FellBehind {
+                                            stream_revision: None,
+                                            position: None,
+                                        },
                                     });
 
                                     return Ok(SubscriptionEvent::FellBehind(args));
@@ -927,10 +946,6 @@ impl Subscription {
     }
 }
 
-fn timestamp_to_datetime(t: prost_types::Timestamp) -> DateTime<Utc> {
-    (UNIX_EPOCH + Duration::new(t.seconds as u64, t.nanos as u32)).into()
-}
-
 /// Runs the subscription command.
 pub fn subscribe_to_stream(
     connection: GrpcClient,

trogon-eventstore/src/event_store/generated/streams.rs

@@ -194,36 +194,38 @@ pub mod read_resp {
     /// The $all or stream subscription has caught up and become live.
     #[derive(Clone, Copy, PartialEq, ::prost::Message)]
     pub struct CaughtUp {
-        /// Current time in the server when the subscription caught up
-        #[prost(message, optional, tag = "1")]
-        pub timestamp: ::core::option::Option<::prost_types::Timestamp>,
-        /// Checkpoint for resuming a stream subscription.
-        /// For stream subscriptions it is populated unless the stream is empty.
-        /// For $all subscriptions it is not populated.
-        #[prost(int64, optional, tag = "2")]
-        pub stream_revision: ::core::option::Option<i64>,
-        /// Checkpoint for resuming a $all subscription.
-        /// For stream subscriptions it is not populated.
-        /// For $all subscriptions it is populated unless the database is empty.
-        #[prost(message, optional, tag = "3")]
-        pub position: ::core::option::Option<Position>,
+        #[prost(oneof = "caught_up::Position", tags = "1, 2, 3")]
+        pub position: ::core::option::Option<caught_up::Position>,
+    }
+    /// Nested message and enum types in `CaughtUp`.
+    pub mod caught_up {
+        #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
+        pub enum Position {
+            #[prost(uint64, tag = "1")]
+            StreamPosition(u64),
+            #[prost(message, tag = "2")]
+            AllStreamPosition(crate::event_store::generated::common::AllStreamPosition),
+            #[prost(message, tag = "3")]
+            NoPosition(()),
+        }
     }
     /// The $all or stream subscription has fallen back into catchup mode and is no longer live.
     #[derive(Clone, Copy, PartialEq, ::prost::Message)]
     pub struct FellBehind {
-        /// Current time in the server when the subscription fell behind
-        #[prost(message, optional, tag = "1")]
-        pub timestamp: ::core::option::Option<::prost_types::Timestamp>,
-        /// Checkpoint for resuming a stream subscription.
-        /// For stream subscriptions it is populated unless the stream is empty.
-        /// For $all subscriptions it is not populated.
-        #[prost(int64, optional, tag = "2")]
-        pub stream_revision: ::core::option::Option<i64>,
-        /// Checkpoint for resuming a $all subscription.
-        /// For stream subscriptions it is not populated.
-        /// For $all subscriptions it is populated unless the database is empty.
-        #[prost(message, optional, tag = "3")]
-        pub position: ::core::option::Option<Position>,
+        #[prost(oneof = "fell_behind::Position", tags = "1, 2, 3")]
+        pub position: ::core::option::Option<fell_behind::Position>,
+    }
+    /// Nested message and enum types in `FellBehind`.
+    pub mod fell_behind {
+        #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
+        pub enum Position {
+            #[prost(uint64, tag = "1")]
+            StreamPosition(u64),
+            #[prost(message, tag = "2")]
+            AllStreamPosition(crate::event_store::generated::common::AllStreamPosition),
+            #[prost(message, tag = "3")]
+            NoPosition(()),
+        }
     }
     #[derive(Clone, PartialEq, ::prost::Message)]
     pub struct ReadEvent {
@@ -283,13 +285,6 @@ pub mod read_resp {
         #[prost(message, optional, tag = "3")]
         pub timestamp: ::core::option::Option<::prost_types::Timestamp>,
     }
-    #[derive(Clone, Copy, PartialEq, ::prost::Message)]
-    pub struct Position {
-        #[prost(uint64, tag = "1")]
-        pub commit_position: u64,
-        #[prost(uint64, tag = "2")]
-        pub prepare_position: u64,
-    }
     #[derive(Clone, PartialEq, ::prost::Message)]
     pub struct StreamNotFound {
         #[prost(message, optional, tag = "1")]

trogon-eventstore/src/operations/gossip.rs

@@ -1,9 +1,8 @@
+use crate::ClientSettings;
 use crate::event_store::client::gossip as wire;
 use crate::grpc::HyperClient;
-use crate::http::http_configure_auth;
 use crate::request::build_request_metadata;
 use crate::types::Endpoint;
-use crate::{ClientSettings, grpc};
 use serde::{Deserialize, Serialize};
 use tonic::{Request, Status};
 use uuid::Uuid;
@@ -71,57 +70,6 @@ pub async fn read(
     Ok(members)
 }
 
-pub(crate) async fn http_read(
-    setts: &ClientSettings,
-    handle: grpc::Handle,
-) -> Result<Vec<MemberInfo>, Box<dyn std::error::Error>> {
-    let client = reqwest::Client::builder()
-        .danger_accept_invalid_certs(!setts.tls_verify_cert)
-        .build()?;
-
-    let default_auth = setts
-        .default_user_name
-        .as_ref()
-        .map(|c| crate::Authentication::Basic(c.clone()));
-
-    let resp = http_configure_auth(
-        client.get(format!("{}/gossip", handle.url())),
-        default_auth.as_ref(),
-    )
-    .send()
-    .await?;
-
-    let gossip = resp.json::<Gossip>().await?;
-
-    Ok(gossip
-        .members
-        .into_iter()
-        .map(|i| MemberInfo {
-            instance_id: i.instance_id,
-            time_stamp: i.time_stamp.timestamp(),
-            state: i.state,
-            is_alive: i.is_alive,
-            http_end_point: Endpoint {
-                host: i.external_http_ip,
-                port: i.external_http_port as u32,
-            },
-            last_commit_position: i.last_commit_position,
-            writer_checkpoint: i.writer_checkpoint,
-            chaser_checkpoint: i.chaser_checkpoint,
-            epoch_position: i.epoch_position,
-            epoch_number: i.epoch_number,
-            epoch_id: i.epoch_id,
-            node_priority: i.node_priority,
-        })
-        .collect())
-}
-
-#[derive(Serialize, Deserialize, Debug)]
-#[serde(rename_all = "camelCase")]
-struct Gossip {
-    members: Vec<HttpMemberInfo>,
-}
-
 #[derive(Debug, Clone)]
 pub struct MemberInfo {
     pub instance_id: Uuid,
@@ -138,31 +86,6 @@ pub struct MemberInfo {
     pub node_priority: i64,
 }
 
-#[derive(Deserialize, Serialize, Debug)]
-#[serde(rename_all = "camelCase")]
-pub struct HttpMemberInfo {
-    pub instance_id: Uuid,
-    pub time_stamp: chrono::DateTime<chrono::Utc>,
-    pub state: VNodeState,
-    pub is_alive: bool,
-    pub internal_tcp_ip: String,
-    pub internal_tcp_port: u16,
-    pub internal_secure_tcp_port: u16,
-    pub external_tcp_ip: String,
-    pub external_secure_tcp_port: u16,
-    #[serde(rename = "httpEndPointIp")]
-    pub external_http_ip: String,
-    #[serde(rename = "httpEndPointPort")]
-    pub external_http_port: u16,
-    pub last_commit_position: i64,
-    pub writer_checkpoint: i64,
-    pub chaser_checkpoint: i64,
-    pub epoch_position: i64,
-    pub epoch_number: i64,
-    pub epoch_id: Uuid,
-    pub node_priority: i64,
-}
-
 #[derive(Copy, Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
 #[serde(rename_all = "PascalCase")]
 pub enum VNodeState {

trogon-eventstore/src/operations/mod.rs

@@ -73,12 +73,9 @@ impl Client {
     pub async fn read_gossip(&self) -> crate::Result<Vec<gossip::MemberInfo>> {
         let handle = self.inner.current_selected_node().await?;
 
-        // We currently use the http endpoint instead of the gRPC one because at that time
-        // 04-25-2022, the public gRPC endpoint doesn't return all the gossip info like current
-        // epoch and other checkpoints.
-        gossip::http_read(self.inner.connection_settings(), handle)
+        gossip::read(self.inner.connection_settings(), &handle.client, handle.uri)
             .await
-            .map_err(|e| crate::Error::IllegalStateError(e.to_string()))
+            .map_err(crate::Error::from_grpc)
     }
 
     pub async fn stats(&self, options: &StatsOptions) -> crate::Result<Stats> {

trogon-eventstore/src/types.rs

@@ -1127,14 +1127,12 @@ pub enum SubscriptionEvent {
 
 #[derive(Debug)]
 pub struct CaughtUp {
-    pub date: DateTime<Utc>,
     pub stream_revision: Option<u64>,
     pub position: Option<Position>,
 }
 
 #[derive(Debug)]
 pub struct FellBehind {
-    pub date: DateTime<Utc>,
     pub stream_revision: Option<u64>,
     pub position: Option<Position>,
 }

trogon-eventstore/tests/api/operations.rs

@@ -1,5 +1,6 @@
 use std::time::Duration;
 use tracing::debug;
+use trogon_eventstore::Credentials;
 use trogon_eventstore::operations;
 use trogon_eventstore::operations::StatsOptions;
 
@@ -186,12 +187,15 @@ async fn test_change_user_password(
         )
         .await?;
 
+    let options = operations::OperationalOptions::default()
+        .authenticated(Credentials::new(login.clone(), password.clone()));
+
     client
         .change_user_password(
             login.as_str(),
-            password,
+            password.as_str(),
             names.next().unwrap(),
-            &Default::default(),
+            &options,
         )
         .await?;
 
@@ -242,10 +246,7 @@ async fn test_op_restart_persistent_subscription_subsystem(
 }
 
 async fn test_scavenge(client: &operations::Client) -> trogon_eventstore::Result<()> {
-    let result = client.start_scavenge(1, 0, &Default::default()).await?;
-    let result = client.stop_scavenge(result.id(), &Default::default()).await;
-
-    assert!(result.is_ok());
+    client.start_scavenge(1, 0, &Default::default()).await?;
 
     Ok(())
 }