client: Replace futures_util::stream::try_unfold with async_stream::try_stream!

ruma-client/Cargo.toml

@@ -17,8 +17,8 @@ version = "0.5.0-alpha.1"
 
 [dependencies]
 assign = "1.1.1"
+async-stream = "0.3.0"
 futures-core = "0.3.8"
-futures-util = "0.3.8"
 http = "0.2.2"
 hyper = { version = "0.14.2", features = ["client", "tcp"] }
 hyper-tls = { version = "0.5.0", optional = true }
@@ -34,6 +34,7 @@ serde_json = "1.0.61"
 
 [dev-dependencies]
 anyhow = "1.0.37"
+futures-util = "0.3.8"
 ruma = { version = "0.0.2", path = "../ruma", features = ["client-api"] }
 tokio = { version = "1.0.1", features = ["macros", "rt"] }
 

ruma-client/src/lib.rs

@@ -107,8 +107,8 @@ use std::{
 };
 
 use assign::assign;
-use futures_core::stream::{Stream, TryStream};
-use futures_util::stream;
+use async_stream::try_stream;
+use futures_core::stream::Stream;
 use http::{uri::Uri, Response as HttpResponse};
 use hyper::client::{Client as HyperClient, HttpConnector};
 use ruma_api::{AuthScheme, OutgoingRequest};
@@ -302,17 +302,13 @@ impl Client {
     pub fn sync<'a>(
         &self,
         filter: Option<&'a SyncFilter<'a>>,
-        since: String,
+        mut since: String,
         set_presence: &'a PresenceState,
         timeout: Option<Duration>,
-    ) -> impl Stream<Item = Result<SyncResponse, Error<ruma_client_api::Error>>>
-           + TryStream<Ok = SyncResponse, Error = Error<ruma_client_api::Error>>
-           + 'a {
+    ) -> impl Stream<Item = Result<SyncResponse, Error<ruma_client_api::Error>>> + 'a {
         let client = self.clone();
-        stream::try_unfold(since, move |since| {
-            let client = client.clone();
-
-            async move {
+        try_stream! {
+            loop {
                 let response = client
                     .request(assign!(SyncRequest::new(), {
                         filter,
@@ -322,10 +318,10 @@ impl Client {
                     }))
                     .await?;
 
-                let next_batch_clone = response.next_batch.clone();
-                Ok(Some((response, next_batch_clone)))
+                since = response.next_batch.clone();
+                yield response;
             }
-        })
+        }
     }
 
     /// Makes a request to a Matrix API endpoint.