Upload src/main.rs

src/main.rs

@@ -20,6 +20,8 @@ use cbc::Decryptor;
 use reqwest::{cookie::Jar, Url};
 use reqwest::header::ACCEPT;
 
+type BoxErr = Box<dyn std::error::Error + Send + Sync + 'static>;
+
 mod api;
 use api::{APIClient, APIError, Format};
 
@@ -228,9 +230,16 @@ async fn stream(State(state): State<Arc<RwLock<APIClient>>>, Query(q): Query<Str
 
     let mut upstream = upstream.bytes_stream();
 
-    let out = async_stream::try_stream! {
-        while let Some(chunk) = upstream.next().await {
-            let chunk = chunk.map_err(|_| std::io::Error::new(std::io::ErrorKind::Other, "upstream"))?;
+    let out = async_stream::stream! {
+        while let Some(chunk_res) = upstream.next().await {
+            let chunk = match chunk_res {
+                Ok(c) => c,
+                Err(e) => {
+                    yield Err::<Bytes, BoxErr>(Box::new(e));
+                    break;
+                }
+            };
+
             carry.extend_from_slice(&chunk);
 
             while carry.len() >= 2048 {
@@ -239,12 +248,12 @@ async fn stream(State(state): State<Arc<RwLock<APIClient>>>, Query(q): Query<Str
                     let _ = decrypt_stripe(&mut block, &key);
                 }
                 block_index += 1;
-                yield Bytes::from(block);
+                yield Ok::<Bytes, BoxErr>(Bytes::from(block));
             }
         }
 
         if !carry.is_empty() {
-            yield Bytes::from(carry);
+            yield Ok::<Bytes, BoxErr>(Bytes::from(std::mem::take(&mut carry)));
         }
     };