Skip to content

Commit afeb980

Browse files
jbrjbr
committed
rebase and clippy
1 parent 9ddd226 commit afeb980

File tree

7 files changed

+187
-162
lines changed

7 files changed

+187
-162
lines changed

‎src/clients/aggregator_client.rs‎

Lines changed: 6 additions & 141 deletions
Original file line numberDiff line numberDiff line change
@@ -1,16 +1,15 @@
1-
use std::{
2-
collections::VecDeque,
3-
fmt::{self, Formatter},
4-
pin::Pin,
5-
task::{ready, Context, Poll},
6-
};
1+
mod task_id_stream;
2+
mod task_stream;
3+
4+
use task_id_stream::TaskIdStream;
5+
use task_stream::TaskStream;
76

87
use crate::{
98
clients::{ClientConnExt, ClientError},
109
entity::{task::ProvisionableTask, Aggregator},
1110
handler::Error,
1211
};
13-
use futures_lite::{stream::Stream, Future, StreamExt};
12+
use futures_lite::StreamExt;
1413
use serde::{de::DeserializeOwned, Serialize};
1514
use trillium::{HeaderValue, KnownHeaderName, Method};
1615
use trillium_client::{Client, Conn};
@@ -148,137 +147,3 @@ impl AggregatorClient {
148147
TaskStream::new(self)
149148
}
150149
}
151-
152-
#[derive(Clone, Debug)]
153-
struct Page {
154-
task_ids: VecDeque<String>,
155-
pagination_token: Option<String>,
156-
}
157-
158-
impl From<TaskIds> for Page {
159-
fn from(
160-
TaskIds {
161-
task_ids,
162-
pagination_token,
163-
}: TaskIds,
164-
) -> Self {
165-
Page {
166-
task_ids: task_ids.into_iter().map(|t| t.to_string()).collect(),
167-
pagination_token,
168-
}
169-
}
170-
}
171-
172-
pub struct TaskIdStream<'a> {
173-
client: &'a AggregatorClient,
174-
page: Option<Page>,
175-
future: Option<Pin<Box<dyn Future<Output = Result<TaskIds, ClientError>> + Send + 'a>>>,
176-
}
177-
178-
impl<'a> TaskIdStream<'a> {
179-
fn new(client: &'a AggregatorClient) -> Self {
180-
Self {
181-
client,
182-
page: None,
183-
future: None,
184-
}
185-
}
186-
}
187-
188-
impl<'a> fmt::Debug for TaskIdStream<'a> {
189-
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
190-
f.debug_struct("TaskIdStream")
191-
.field("client", &self.client)
192-
.field("current_page", &self.page)
193-
.field("current_future", &"..")
194-
.finish()
195-
}
196-
}
197-
198-
impl Stream for TaskIdStream<'_> {
199-
type Item = Result<String, ClientError>;
200-
201-
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
202-
let Self {
203-
client,
204-
ref mut page,
205-
ref mut future,
206-
} = *self;
207-
208-
loop {
209-
if let Some(page) = page {
210-
if let Some(task_id) = page.task_ids.pop_front() {
211-
return Poll::Ready(Some(Ok(task_id)));
212-
}
213-
214-
if page.pagination_token.is_none() {
215-
return Poll::Ready(None);
216-
}
217-
}
218-
219-
if let Some(fut) = future {
220-
*page = Some(ready!(Pin::new(&mut *fut).poll(cx))?.into());
221-
*future = None;
222-
} else {
223-
let pagination_token = page.as_ref().and_then(|page| page.pagination_token.clone());
224-
225-
*future = Some(Box::pin(async move {
226-
client.get_task_id_page(pagination_token.as_deref()).await
227-
}));
228-
};
229-
}
230-
}
231-
}
232-
233-
pub struct TaskStream<'a> {
234-
client: &'a AggregatorClient,
235-
task_id_stream: TaskIdStream<'a>,
236-
task_future: Option<
237-
Pin<Box<dyn Future<Output = Option<Result<TaskResponse, ClientError>>> + Send + 'a>>,
238-
>,
239-
}
240-
241-
impl<'a> fmt::Debug for TaskStream<'a> {
242-
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
243-
f.debug_struct("TaskStream").field("future", &"..").finish()
244-
}
245-
}
246-
247-
impl<'a> TaskStream<'a> {
248-
fn new(client: &'a AggregatorClient) -> Self {
249-
Self {
250-
task_id_stream: client.task_id_stream(),
251-
client,
252-
task_future: None,
253-
}
254-
}
255-
}
256-
257-
impl Stream for TaskStream<'_> {
258-
type Item = Result<TaskResponse, ClientError>;
259-
260-
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
261-
let Self {
262-
client,
263-
ref mut task_id_stream,
264-
ref mut task_future,
265-
} = *self;
266-
267-
loop {
268-
if let Some(future) = task_future {
269-
let res = ready!(Pin::new(&mut *future).poll(cx));
270-
*task_future = None;
271-
return Poll::Ready(res);
272-
}
273-
274-
*task_future = match ready!(Pin::new(&mut *task_id_stream).poll_next(cx)) {
275-
Some(Ok(task_id)) => Some(Box::pin(async move {
276-
let task_id = task_id;
277-
Some(client.get_task(&task_id).await)
278-
})),
279-
None => return Poll::Ready(None),
280-
Some(Err(e)) => return Poll::Ready(Some(Err(e))),
281-
};
282-
}
283-
}
284-
}

‎src/clients/aggregator_client/task_id_stream.rs‎

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,93 @@
1+
use std::{
2+
collections::VecDeque,
3+
fmt::{self, Debug, Formatter},
4+
pin::Pin,
5+
task::{ready, Context, Poll},
6+
};
7+
8+
use super::{AggregatorClient, TaskIds};
9+
use crate::clients::ClientError;
10+
use futures_lite::{stream::Stream, Future};
11+
12+
#[derive(Clone, Debug)]
13+
struct Page {
14+
task_ids: VecDeque<String>,
15+
pagination_token: Option<String>,
16+
}
17+
18+
impl From<TaskIds> for Page {
19+
fn from(
20+
TaskIds {
21+
task_ids,
22+
pagination_token,
23+
}: TaskIds,
24+
) -> Self {
25+
Page {
26+
task_ids: task_ids.into_iter().map(|t| t.to_string()).collect(),
27+
pagination_token,
28+
}
29+
}
30+
}
31+
32+
type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
33+
34+
pub struct TaskIdStream<'a> {
35+
client: &'a AggregatorClient,
36+
page: Option<Page>,
37+
future: Option<BoxFuture<'a, Result<TaskIds, ClientError>>>,
38+
}
39+
40+
impl<'a> TaskIdStream<'a> {
41+
pub(super) fn new(client: &'a AggregatorClient) -> Self {
42+
Self {
43+
client,
44+
page: None,
45+
future: None,
46+
}
47+
}
48+
}
49+
50+
impl<'a> Debug for TaskIdStream<'a> {
51+
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
52+
f.debug_struct("TaskIdStream")
53+
.field("client", &self.client)
54+
.field("current_page", &self.page)
55+
.field("current_future", &"..")
56+
.finish()
57+
}
58+
}
59+
60+
impl Stream for TaskIdStream<'_> {
61+
type Item = Result<String, ClientError>;
62+
63+
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
64+
let Self {
65+
client,
66+
ref mut page,
67+
ref mut future,
68+
} = *self;
69+
70+
loop {
71+
if let Some(page) = page {
72+
if let Some(task_id) = page.task_ids.pop_front() {
73+
return Poll::Ready(Some(Ok(task_id)));
74+
}
75+
76+
if page.pagination_token.is_none() {
77+
return Poll::Ready(None);
78+
}
79+
}
80+
81+
if let Some(fut) = future {
82+
*page = Some(ready!(Pin::new(&mut *fut).poll(cx))?.into());
83+
*future = None;
84+
} else {
85+
let pagination_token = page.as_ref().and_then(|page| page.pagination_token.clone());
86+
87+
*future = Some(Box::pin(async move {
88+
client.get_task_id_page(pagination_token.as_deref()).await
89+
}));
90+
};
91+
}
92+
}
93+
}

‎src/clients/aggregator_client/task_stream.rs‎

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
use std::{
2+
fmt::{self, Debug, Formatter},
3+
pin::Pin,
4+
task::{ready, Context, Poll},
5+
};
6+
7+
use super::{task_id_stream::TaskIdStream, AggregatorClient, TaskResponse};
8+
use crate::clients::ClientError;
9+
use futures_lite::{stream::Stream, Future};
10+
11+
type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
12+
13+
pub struct TaskStream<'a> {
14+
client: &'a AggregatorClient,
15+
task_id_stream: TaskIdStream<'a>,
16+
task_future: Option<BoxFuture<'a, Option<Result<TaskResponse, ClientError>>>>,
17+
}
18+
19+
impl<'a> Debug for TaskStream<'a> {
20+
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
21+
f.debug_struct("TaskStream").field("future", &"..").finish()
22+
}
23+
}
24+
25+
impl<'a> TaskStream<'a> {
26+
pub(super) fn new(client: &'a AggregatorClient) -> Self {
27+
Self {
28+
task_id_stream: client.task_id_stream(),
29+
client,
30+
task_future: None,
31+
}
32+
}
33+
}
34+
35+
impl Stream for TaskStream<'_> {
36+
type Item = Result<TaskResponse, ClientError>;
37+
38+
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
39+
let Self {
40+
client,
41+
ref mut task_id_stream,
42+
ref mut task_future,
43+
} = *self;
44+
45+
loop {
46+
if let Some(future) = task_future {
47+
let res = ready!(Pin::new(&mut *future).poll(cx));
48+
*task_future = None;
49+
return Poll::Ready(res);
50+
}
51+
52+
*task_future = match ready!(Pin::new(&mut *task_id_stream).poll_next(cx)) {
53+
Some(Ok(task_id)) => Some(Box::pin(async move {
54+
let task_id = task_id;
55+
Some(client.get_task(&task_id).await)
56+
})),
57+
None => return Poll::Ready(None),
58+
Some(Err(e)) => return Poll::Ready(Some(Err(e))),
59+
};
60+
}
61+
}
62+
}

‎src/queue/job.rs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -81,13 +81,15 @@ pub struct SharedJobState {
8181
pub auth0_client: Auth0Client,
8282
pub postmark_client: PostmarkClient,
8383
pub http_client: Client,
84+
pub crypter: crate::Crypter,
8485
}
8586
impl From<&Config> for SharedJobState {
8687
fn from(config: &Config) -> Self {
8788
Self {
8889
auth0_client: Auth0Client::new(config),
8990
postmark_client: PostmarkClient::new(config),
9091
http_client: config.client.clone(),
92+
crypter: config.crypter.clone(),
9193
}
9294
}
9395
}

‎src/queue/job/v1/task_sync.rs‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ use crate::{
33
queue::job::{EnqueueJob, Job, JobError, SharedJobState, V1},
44
};
55
use futures_lite::StreamExt;
6-
use sea_orm::{ColumnTrait, ConnectionTrait, EntityTrait, ModelTrait, PaginatorTrait, QueryFilter};
6+
use sea_orm::{ColumnTrait, ConnectionTrait, EntityTrait, QueryFilter};
77
use serde::{Deserialize, Serialize};
88
use time::Duration;
99

@@ -24,7 +24,9 @@ impl TaskSync {
2424
.await?;
2525

2626
for aggregator in aggregators {
27-
let client = aggregator.client(job_state.http_client.clone());
27+
let client = aggregator
28+
.client(job_state.http_client.clone(), &job_state.crypter)
29+
.map_err(|e| JobError::ClientOther(e.to_string()))?;
2830
while let Some(task_from_aggregator) = client.task_stream().next().await.transpose()? {
2931
let task_id = task_from_aggregator.task_id.to_string();
3032
if let Some(_task_from_db) = Tasks::find_by_id(&task_id).one(db).await? {

0 commit comments

Comments
 (0)