edward
/
mangadex-home-rs
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Fix some future not send lints

master
Edward Shen 2021-07-17 13:32:43 -04:00
parent e95afd3e32
commit afa2cf55fa
Signed by: edward
GPG Key ID: 19182661E818369F
4 changed files with 105 additions and 98 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/cache/mem.rs vendored
View File

@ -66,7 +66,7 @@ impl ToRedisArgs for CacheValue {
where
W: ?Sized + redis::RedisWrite,
{
out.write_arg(&bincode::serialize(self).expect("serialization to work"))
out.write_arg(&bincode::serialize(self).expect("serialization to work"));
}
}

2
src/cache/mod.rs vendored
View File

@ -41,7 +41,7 @@ impl ToRedisArgs for CacheKey {
where
W: ?Sized + redis::RedisWrite,
{
out.write_arg_fmt(self)
out.write_arg_fmt(self);
}
}

195
src/client.rs
View File

@ -84,8 +84,11 @@ impl CachingClient {
key: CacheKey,
cache: Data<dyn Cache>,
) -> FetchResult {
if let Some(recv) = self.locks.read().get(&url) {
let mut recv = recv.clone();
let maybe_receiver = {
let lock = self.locks.read();
lock.get(&url).map(Clone::clone)
};
if let Some(mut recv) = maybe_receiver {
loop {
if !matches!(*recv.borrow(), FetchResult::Processing) {
break;
@ -97,100 +100,17 @@ impl CachingClient {
return recv.borrow().clone();
}
let url_0 = url.clone();
let notify = Arc::new(Notify::new());
let notify2 = Arc::clone(&notify);
tokio::spawn(self.fetch_and_cache_impl(cache, url.clone(), key, Arc::clone(&notify)));
notify.notified().await;
tokio::spawn(async move {
let (tx, rx) = channel(FetchResult::Processing);
self.locks.write().insert(url.clone(), rx);
notify.notify_one();
let resp = self.inner.get(&url).send().await;
let resp = match resp {
Ok(mut resp) => {
let content_type = resp.headers().get(CONTENT_TYPE);
let is_image = content_type
.map(|v| String::from_utf8_lossy(v.as_ref()).contains("image/"))
.unwrap_or_default();
if resp.status() != StatusCode::OK || !is_image {
warn!("Got non-OK or non-image response code from upstream, proxying and not caching result.");
let mut headers = DEFAULT_HEADERS.clone();
if let Some(content_type) = content_type {
headers.insert(CONTENT_TYPE, content_type.clone());
}
FetchResult::Data(
resp.status(),
headers,
resp.bytes().await.unwrap_or_default(),
)
} else {
let (content_type, length, last_mod) = {
let headers = resp.headers_mut();
(
headers.remove(CONTENT_TYPE),
headers.remove(CONTENT_LENGTH),
headers.remove(LAST_MODIFIED),
)
};
let body = resp.bytes().await.unwrap();
debug!("Inserting into cache");
let metadata = ImageMetadata::new(
content_type.clone(),
length.clone(),
last_mod.clone(),
)
.unwrap();
match cache.put(key, body.clone(), metadata).await {
Ok(()) => {
debug!("Done putting into cache");
let mut headers = DEFAULT_HEADERS.clone();
if let Some(content_type) = content_type {
headers.insert(CONTENT_TYPE, content_type);
}
if let Some(content_length) = length {
headers.insert(CONTENT_LENGTH, content_length);
}
if let Some(last_modified) = last_mod {
headers.insert(LAST_MODIFIED, last_modified);
}
FetchResult::Data(StatusCode::OK, headers, body)
}
Err(e) => {
warn!("Failed to insert into cache: {}", e);
FetchResult::InternalServerError
}
}
}
}
Err(e) => {
error!("Failed to fetch image from server: {}", e);
FetchResult::ServiceUnavailable
}
};
// This shouldn't happen
tx.send(resp).unwrap();
self.locks.write().remove(&url);
});
notify2.notified().await;
let mut recv = self.locks.read().get(&url_0).unwrap().clone();
let mut recv = self
.locks
.read()
.get(&url)
.expect("receiver to exist since we just made one")
.clone();
loop {
if !matches!(*recv.borrow(), FetchResult::Processing) {
break;
@ -203,6 +123,95 @@ impl CachingClient {
resp
}
async fn fetch_and_cache_impl(
&self,
cache: Data<dyn Cache>,
url: String,
key: CacheKey,
notify: Arc<Notify>,
) {
let (tx, rx) = channel(FetchResult::Processing);
self.locks.write().insert(url.clone(), rx);
notify.notify_one();
let resp = self.inner.get(&url).send().await;
let resp = match resp {
Ok(mut resp) => {
let content_type = resp.headers().get(CONTENT_TYPE);
let is_image = content_type
.map(|v| String::from_utf8_lossy(v.as_ref()).contains("image/"))
.unwrap_or_default();
if resp.status() != StatusCode::OK || !is_image {
warn!("Got non-OK or non-image response code from upstream, proxying and not caching result.");
let mut headers = DEFAULT_HEADERS.clone();
if let Some(content_type) = content_type {
headers.insert(CONTENT_TYPE, content_type.clone());
}
FetchResult::Data(
resp.status(),
headers,
resp.bytes().await.unwrap_or_default(),
)
} else {
let (content_type, length, last_mod) = {
let headers = resp.headers_mut();
(
headers.remove(CONTENT_TYPE),
headers.remove(CONTENT_LENGTH),
headers.remove(LAST_MODIFIED),
)
};
let body = resp.bytes().await.unwrap();
debug!("Inserting into cache");
let metadata =
ImageMetadata::new(content_type.clone(), length.clone(), last_mod.clone())
.unwrap();
match cache.put(key, body.clone(), metadata).await {
Ok(()) => {
debug!("Done putting into cache");
let mut headers = DEFAULT_HEADERS.clone();
if let Some(content_type) = content_type {
headers.insert(CONTENT_TYPE, content_type);
}
if let Some(content_length) = length {
headers.insert(CONTENT_LENGTH, content_length);
}
if let Some(last_modified) = last_mod {
headers.insert(LAST_MODIFIED, last_modified);
}
FetchResult::Data(StatusCode::OK, headers, body)
}
Err(e) => {
warn!("Failed to insert into cache: {}", e);
FetchResult::InternalServerError
}
}
}
}
Err(e) => {
error!("Failed to fetch image from server: {}", e);
FetchResult::ServiceUnavailable
}
};
// This shouldn't happen
tx.send(resp).unwrap();
self.locks.write().remove(&url);
}
#[inline]
pub const fn inner(&self) -> &Client {
&self.inner

4
src/routes.rs
View File

@ -51,7 +51,6 @@ async fn index() -> impl Responder {
HttpResponse::Ok().body(include_str!("index.html"))
}
#[allow(clippy::future_not_send)]
#[get("/{token}/data/{chapter_hash}/{file_name}")]
async fn token_data(
state: Data<RwLockServerState>,
@ -68,7 +67,6 @@ async fn token_data(
fetch_image(state, cache, chapter_hash, file_name, false).await
}
#[allow(clippy::future_not_send)]
#[get("/{token}/data-saver/{chapter_hash}/{file_name}")]
async fn token_data_saver(
state: Data<RwLockServerState>,
@ -124,7 +122,7 @@ pub async fn default(state: Data<RwLockServerState>, req: HttpRequest) -> impl R
ServerResponse::HttpResponse(resp)
}
#[allow(clippy::future_not_send, clippy::unused_async)]
#[allow(clippy::unused_async)]
#[get("/prometheus")]
pub async fn metrics() -> impl Responder {
let metric_families = prometheus::gather();