From 0f8446a35850642ae35cef6e43681d9eed910936 Mon Sep 17 00:00:00 2001
From: Pavel Borzenkov <pavel@fly.io>
Date: Tue, 17 Sep 2024 13:31:31 +0200
Subject: [PATCH] corro-client: expose subscription hash to the clients

---
 crates/corro-client/src/lib.rs | 11 +++++++++++
 crates/corro-client/src/sub.rs |  7 +++++++
 2 files changed, 18 insertions(+)

diff --git a/crates/corro-client/src/lib.rs b/crates/corro-client/src/lib.rs
index e23bd017..b5019320 100644
--- a/crates/corro-client/src/lib.rs
+++ b/crates/corro-client/src/lib.rs
@@ -137,9 +137,14 @@ impl CorrosionApiClient {
             .get(HeaderName::from_static("corro-query-id"))
             .and_then(|v| v.to_str().ok().and_then(|v| v.parse().ok()))
             .ok_or(Error::ExpectedQueryId)?;
+        let hash = res
+            .headers()
+            .get(HeaderName::from_static("corro-query-hash"))
+            .and_then(|v| v.to_str().map(ToOwned::to_owned).ok());
 
         Ok(SubscriptionStream::new(
             id,
+            hash,
             self.api_client.clone(),
             self.api_addr,
             res.into_body(),
@@ -189,8 +194,14 @@ impl CorrosionApiClient {
             return Err(Error::UnexpectedStatusCode(res.status()));
         }
 
+        let hash = res
+            .headers()
+            .get(HeaderName::from_static("corro-query-hash"))
+            .and_then(|v| v.to_str().map(ToOwned::to_owned).ok());
+
         Ok(SubscriptionStream::new(
             id,
+            hash,
             self.api_client.clone(),
             self.api_addr,
             res.into_body(),
diff --git a/crates/corro-client/src/sub.rs b/crates/corro-client/src/sub.rs
index 25fb4bcd..6d21a475 100644
--- a/crates/corro-client/src/sub.rs
+++ b/crates/corro-client/src/sub.rs
@@ -56,6 +56,7 @@ type FramedBody = FramedRead<IoBodyStreamReader, LinesBytesCodec>;
 
 pub struct SubscriptionStream<T> {
     id: Uuid,
+    hash: Option<String>,
     client: hyper::Client<HttpConnector, Body>,
     api_addr: SocketAddr,
     observed_eoq: bool,
@@ -91,6 +92,7 @@ where
 {
     pub fn new(
         id: Uuid,
+        hash: Option<String>,
         client: hyper::Client<HttpConnector, Body>,
         api_addr: SocketAddr,
         body: hyper::Body,
@@ -98,6 +100,7 @@ where
     ) -> Self {
         Self {
             id,
+            hash,
             client,
             api_addr,
             observed_eoq: change_id.is_some(),
@@ -117,6 +120,10 @@ where
         self.id
     }
 
+    pub fn hash(&self) -> Option<&str> {
+        self.hash.as_deref()
+    }
+
     pub fn api_addr(&self) -> SocketAddr {
         self.api_addr
     }