chris/ruwuma
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Use stream::try_unfold from futures 0.3.2

Browse Source
This commit is contained in:
Jonas Platte 2020-02-04 19:20:43 +01:00
parent e4cf0fbf1e
commit aee5693fd8
No known key found for this signature in database
GPG Key ID: 7D261D771D915378
1 changed files with 5 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
src/lib.rs
View File

@ -301,14 +301,6 @@ where
+ TryStream<Ok = api::r0::sync::sync_events::IncomingResponse, Error = Error> {
use api::r0::sync::sync_events;
// TODO: Is this really the way TryStreams are supposed to work?
#[derive(Debug, PartialEq, Eq)]
enum State {
InitialSync,
Since(String),
Errored,
}
let client = self.clone();
let set_presence = if set_presence {
None
@ -316,23 +308,12 @@ where
Some(sync_events::SetPresence::Offline)
};
let initial_state = match since {
Some(s) => State::Since(s),
None => State::InitialSync,
};
stream::unfold(initial_state, move |state| {
stream::try_unfold(since, move |since| {
let client = client.clone();
let filter = filter.clone();
async move {
let since = match state {
State::Errored => return None,
State::Since(s) => Some(s),
State::InitialSync => None,
};
let res = client
let response = client
.request(sync_events::Request {
filter,
since,
@ -340,15 +321,10 @@ where
set_presence,
timeout: None,
})
.await;
.await?;
match res {
Ok(response) => {
let next_batch_clone = response.next_batch.clone();
Some((Ok(response), State::Since(next_batch_clone)))
}
Err(e) => Some((Err(e), State::Errored)),
}
let next_batch_clone = response.next_batch.clone();
Ok(Some((response, Some(next_batch_clone))))
}
})
}