refactor(SyncEngine): make it easier to test

Author: davidatsurge

rustc-ice-2025-02-13T11_42_50-54470.txt

@@ -1,12 +1,13 @@
-use std::fmt::Debug;
+use std::{fmt::Debug, future::Future};
 
 /// # Panics
 ///
 /// Panics if `n` = 0.
-pub async fn try_n_times<T, E: Debug>(
-    func: impl AsyncFn() -> Result<T, E>,
-    mut n: usize,
-) -> Result<T, E> {
+pub async fn try_n_times<Fut, T, E>(func: impl Fn() -> Fut, mut n: usize) -> Result<T, E>
+where
+    Fut: Future<Output = Result<T, E>>,
+    E: Debug,
+{
     if n == 0 {
         panic!("try_n_times() received a non-positive integer.");
     }

shared/src/retry.rs

@@ -1,7 +1,8 @@
 use optimistic::db::{DbWithOptimisticChanges, ReactivityTrackers};
 use registry::Registry;
 use send_wrapper::SendWrapper;
-use std::{marker::PhantomData, sync::Arc};
+use url::Url;
+use std::{future::Future, pin::Pin, sync::Arc};
 pub use websocket_updates::transport::*;
 mod conversions;
 mod initial_sync;
@@ -40,11 +41,12 @@ mod new_defn;
 
 pub use new_defn::DbStoreMarkers;
 
-pub struct SyncEngine<Transport, GithubApi> {
+pub struct SyncEngine<Transport: TransportTrait, GithubApi> {
     pub db: Rc<DbWithOptimisticChanges<DbStoreMarkers>>,
     pub db_subscriptions: SendWrapper<Rc<Registry<DbSubscription>>>,
     endpoint_client: EndpointClient,
-    _transport: PhantomData<(Transport, GithubApi)>,
+    _github_api: GithubApi,
+    make_transport: Rc<dyn Fn(Url) -> Pin<Box<dyn Future<Output = Result<Transport, Transport::TransportError>>>>>,
 }
 
 const MAX_PER_PAGE: i32 = 100;

shared/src/sync_engine/mod.rs

@@ -1,7 +1,7 @@
+use std::future::Future;
+use std::pin::Pin;
 /// Without this isolation, our `impl` definition for the `DbStoreMarkers` type will not have one
 /// "defining use."
-
-use std::marker::PhantomData;
 use std::rc::Rc;
 
 use crate::types::label::Label;
@@ -19,6 +19,7 @@ use crate::{
 };
 use send_wrapper::SendWrapper;
 use typesafe_idb::{StoreMarker, TypesafeDb};
+use url::Url;
 
 use super::optimistic::db::DbWithOptimisticChanges;
 use super::registry::Registry;
@@ -41,8 +42,19 @@ pub type DbStoreMarkers = impl StoreMarker<IssueCommentsInitialSyncStatus>
     + StoreMarker<LastWebhookUpdateAt>
     + Default;
 
-impl<W: TransportTrait, GithubApi> SyncEngine<W, GithubApi> {
-    pub async fn new(endpoint_client: EndpointClient) -> SyncResult<Self, W> {
+impl<Transport: TransportTrait, GithubApi> SyncEngine<Transport, GithubApi> {
+    pub async fn new<F, Fut>(
+        endpoint_client: EndpointClient,
+        make_transport: F,
+        github_api: GithubApi,
+    ) -> SyncResult<Self, Transport>
+    where
+        F: Fn(Url) -> Fut + 'static,
+        Fut: Future<Output = Result<Transport, Transport::TransportError>> + 'static,
+    {
+        // Convert the nice generic function into the boxed version we need internally
+        let make_transport = Rc::new(move |url| Box::pin(make_transport(url)) as Pin<Box<dyn Future<Output = _>>>);
+
         let db = TypesafeDb::builder("heimisch".into())
             .with_store::<Issue>()
             .with_store::<User>()
@@ -72,14 +84,15 @@ impl<W: TransportTrait, GithubApi> SyncEngine<W, GithubApi> {
                     })
                     .for_each(|sub| (sub.func)());
             }),
-        ).await?;
+        )
+        .await?;
 
         Ok(Self {
             db: Rc::new(db),
             db_subscriptions: SendWrapper::new(db_subscriptions),
             endpoint_client,
-            _transport: PhantomData,
+            _github_api: github_api,
+            make_transport,
         })
     }
 }
-

shared/src/sync_engine/new_defn.rs

@@ -8,9 +8,7 @@ use wasm_bindgen_test::wasm_bindgen_test;
 
 use crate::{
     endpoints::endpoint_client::EndpointClient,
-    sync_engine::{
-        websocket_updates::tests::MockTransport, DbStoreMarkers, DbSubscription, SyncEngine,
-    },
+    sync_engine::{tests::MockTransport, DbStoreMarkers, DbSubscription, SyncEngine},
     types::{
         issue::{Issue, RepositoryIdIndex},
         repository::Repository,
@@ -20,10 +18,11 @@ use crate::{
 use super::{TxnBuilderWithOptimisticChanges, TxnWithOptimisticChanges};
 
 async fn get_sync_engine() -> SyncEngine<MockTransport, ()> {
-    SyncEngine::<MockTransport, ()>::new(EndpointClient::new(
-        |_| (),
-        Url::parse("https://www.example.com/").unwrap(),
-    ))
+    SyncEngine::<MockTransport, ()>::new(
+        EndpointClient::new(|_| (), Url::parse("https://www.example.com/").unwrap()),
+        |_| async { Ok(MockTransport::new().0) },
+        (),
+    )
     .await
     .unwrap()
 }
@@ -263,7 +262,7 @@ pub async fn get_all_no_optimisim_create_overlapping() {
         .with_txn_2(async |txn| {
             txn.object_store::<Issue>()
                 .unwrap()
-                .create(Default::default(), async {Ok( Default::default()) });
+                .create(Default::default(), async { Ok(Default::default()) });
         })
         .should_overlap(true)
         .call()

shared/src/sync_engine/optimistic/db/tests.rs

@@ -67,5 +67,5 @@ pub trait BinaryTransportTrait:
     type ConnError: Debug;
 
     #[allow(async_fn_in_trait)]
-    async fn establish_conn(url: &Url) -> Result<Self, Self::ConnError>;
+    async fn establish_conn(url: Url) -> Result<Self, Self::ConnError>;
 }

shared/src/sync_engine/websocket_updates/binary_transport.rs

@@ -1,8 +1,6 @@
 pub mod applying_error;
 pub mod binary_transport;
 
-#[cfg(test)]
-pub mod tests;
 pub mod transport;
 
 use applying_error::{ApplyingError, ApplyingResult};
@@ -48,7 +46,7 @@ where
             })
             .expect(""),
         ));
-        let websocket_conn = try_n_times(async || Transport::establish(&url).await, 3)
+        let websocket_conn = try_n_times(|| (self.make_transport)(url.clone()), 3)
             .await
             .map_err(|e| SyncErrorSrc::Transport(e))?;
         pin_mut!(websocket_conn);

shared/src/sync_engine/websocket_updates/mod.rs

@@ -1,17 +1,17 @@
-use codee::{Decoder, Encoder};
-use futures::{Sink, Stream};
-use pin_project::pin_project;
-use std::fmt::Debug;
-use std::task::{ready, Poll};
-use url::Url;
-
 use crate::{
     endpoints::defns::api::websocket_updates::{ClientMsg, ServerMsg},
     sync_engine::websocket_updates::binary_transport::{
         BinaryTransportError, BinaryTransportTrait, ConnOrClosedError, JsonSerdeToBinaryCodec,
     },
 };
+use codee::{Decoder, Encoder};
+use futures::{Sink, Stream};
+use pin_project::pin_project;
+use std::fmt::Debug;
+use std::task::{ready, Poll};
+use url::Url;
 
+#[derive(Clone)]
 #[pin_project]
 pub struct Transport<I> {
     #[pin]
@@ -100,18 +100,12 @@ pub trait TransportTrait:
     Sized + Sink<ClientMsg> + Stream<Item = Result<ServerMsg, Self::TransportError>>
 {
     type TransportError: Debug;
-
-    #[allow(async_fn_in_trait)]
-    async fn establish(url: &Url) -> Result<Self, Self::TransportError>;
 }
 
-impl<T> TransportTrait for Transport<T>
-where
-    T: BinaryTransportTrait,
-{
-    type TransportError = BinaryTransportError<<T as Sink<Vec<u8>>>::Error>;
-
-    async fn establish(url: &Url) -> Result<Self, Self::TransportError> {
+impl<T: BinaryTransportTrait> Transport<T> {
+    pub async fn new(
+        url: Url,
+    ) -> Result<Self, <Transport<T> as TransportTrait>::TransportError> {
         let inner = T::establish_conn(url)
             .await
             .map_err(BinaryTransportError::Conn)?;
@@ -120,45 +114,69 @@ where
     }
 }
 
+impl<T> TransportTrait for Transport<T>
+where
+    T: BinaryTransportTrait,
+{
+    type TransportError = BinaryTransportError<<T as Sink<Vec<u8>>>::Error>;
+}
+
 #[cfg(test)]
 pub mod tests {
-    use std::{pin::Pin, task::{Context, Poll}};
+    use std::{
+        pin::Pin,
+        task::{Context, Poll},
+    };
 
     use futures::{channel::mpsc, Sink, Stream};
 
     use crate::endpoints::defns::api::websocket_updates::{ClientMsg, ServerMsg};
 
+    use super::TransportTrait;
+
     pub struct MockTransportHandler {
         send: mpsc::Sender<ServerMsg>,
         recv: mpsc::Receiver<ClientMsg>,
     }
 
-    pub struct TestTransport {
+    pub struct MockTransport {
         recv: mpsc::Receiver<ServerMsg>,
         send: mpsc::Sender<ClientMsg>,
     }
 
-    impl TestTransport {
+    impl TransportTrait for MockTransport {
+        type TransportError = mpsc::SendError;
+    }
+
+    impl MockTransport {
         pub fn new() -> (Self, MockTransportHandler) {
             let (server_msg_sender, server_msg_receiver) = mpsc::channel(100);
             let (client_msg_sender, client_msg_receiver) = mpsc::channel(100);
 
             (
-                Self { recv: server_msg_receiver, send: client_msg_sender },
-                MockTransportHandler { send: server_msg_sender, recv: client_msg_receiver },
+                Self {
+                    recv: server_msg_receiver,
+                    send: client_msg_sender,
+                },
+                MockTransportHandler {
+                    send: server_msg_sender,
+                    recv: client_msg_receiver,
+                },
             )
         }
     }
 
-    impl Stream for TestTransport {
+    impl Stream for MockTransport {
         type Item = Result<ServerMsg, mpsc::SendError>;
 
         fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-            Pin::new(&mut self.recv).poll_next(cx).map(|opt| opt.map(Ok))
+            Pin::new(&mut self.recv)
+                .poll_next(cx)
+                .map(|opt| opt.map(Ok))
         }
     }
 
-    impl Sink<ClientMsg> for TestTransport {
+    impl Sink<ClientMsg> for MockTransport {
         type Error = mpsc::SendError;
 
         fn poll_ready(
@@ -186,4 +204,4 @@ pub mod tests {
             Pin::new(&mut self.send).poll_close(cx)
         }
     }
-}
+}

shared/src/sync_engine/websocket_updates/tests.rs

@@ -17,6 +17,7 @@ use std::rc::Rc;
 use leptos::prelude::*;
 
 use routing::Routed;
+use shared::sync_engine::Transport;
 use sync_engine_provider::SyncEngine;
 
 use crate::consts::ENDPOINT_CLIENT;
@@ -25,9 +26,13 @@ use crate::consts::ENDPOINT_CLIENT;
 pub fn App() -> impl IntoView {
     let sync_engine = LocalResource::new(move || async move {
         Rc::new(
-            SyncEngine::new(ENDPOINT_CLIENT.with(|e| e.clone()))
-                .await
-                .unwrap(),
+            SyncEngine::new(
+                ENDPOINT_CLIENT.with(|e| e.clone()),
+                |url| async { Transport::new(url).await },
+                github_api::github_api_trait::GithubApi,
+            )
+            .await
+            .unwrap(),
         )
     });