From b364f13f78a789896dec34f2f2424b06a9f59f64 Mon Sep 17 00:00:00 2001
From: George Danezis <gdanezis@users.noreply.github.com>
Date: Fri, 22 Apr 2022 09:29:33 +0200
Subject: [PATCH] Add a safe client function that augments the follower API
 with downloading the transactions information (#1446)

Co-authored-by: George Danezis <george@danez.is>
---
 sui_core/src/safe_client.rs | 41 +++++++++++++++++++++++++++++++++++++
 1 file changed, 41 insertions(+)

diff --git a/sui_core/src/safe_client.rs b/sui_core/src/safe_client.rs
index 4dd06c4bc93f2..5e7faf7136aeb 100644
--- a/sui_core/src/safe_client.rs
+++ b/sui_core/src/safe_client.rs
@@ -221,6 +221,47 @@ impl<C> SafeClient<C> {
     }
 }
 
+impl<C> SafeClient<C>
+where
+    C: AuthorityAPI + Send + Sync + Clone + 'static,
+{
+    /// Uses the follower API and augments each digest received with a full transactions info structure.
+    pub async fn handle_transaction_info_request_to_transaction_info(
+        &self,
+        request: BatchInfoRequest,
+    ) -> Result<
+        impl futures::Stream<Item = Result<(u64, TransactionInfoResponse), SuiError>> + '_,
+        SuiError,
+    > {
+        let new_stream = self
+            .handle_batch_stream(request)
+            .await
+            .map_err(|err| SuiError::GenericAuthorityError {
+                error: format!("Stream error: {:?}", err),
+            })?
+            .filter_map(|item| {
+                let _client = self.clone();
+                async move {
+                    match &item {
+                        Ok(BatchInfoResponseItem(UpdateItem::Batch(_signed_batch))) => None,
+                        Ok(BatchInfoResponseItem(UpdateItem::Transaction((seq, digest)))) => {
+                            // Download the full transaction info
+                            let transaction_info_request = TransactionInfoRequest::from(*digest);
+                            let res = _client
+                                .handle_transaction_info_request(transaction_info_request)
+                                .await
+                                .map(|v| (*seq, v));
+                            Some(res)
+                        }
+                        Err(err) => Some(Err(err.clone())),
+                    }
+                }
+            });
+
+        Ok(new_stream)
+    }
+}
+
 #[async_trait]
 impl<C> AuthorityAPI for SafeClient<C>
 where