fixup! tough: migrate to async

tough/src/lib.rs

@@ -56,7 +56,7 @@ use crate::schema::{
     DelegatedRole, Delegations, Role, RoleType, Root, Signed, Snapshot, Timestamp,
 };
 pub use crate::target_name::TargetName;
-use crate::transport::IntoVec;
+pub use crate::transport::IntoVec;
 pub use crate::transport::{
     DefaultTransport, FilesystemTransport, Transport, TransportError, TransportErrorKind,
 };
@@ -443,7 +443,8 @@ impl Repository {
     pub async fn read_target(
         &self,
         name: &TargetName,
-    ) -> Result<Option<impl Stream<Item = error::Result<Bytes>> + Send>> {
+    ) -> Result<Option<impl Stream<Item = error::Result<Bytes>> + IntoVec<error::Error> + Send>>
+    {
         // Check for repository metadata expiration.
         if self.expiration_enforcement == ExpirationEnforcement::Safe {
             ensure!(

tough/src/transport.rs

@@ -16,14 +16,16 @@ use url::Url;
 
 pub type TransportStream = Pin<Box<dyn Stream<Item = Result<Bytes, TransportError>> + Send>>;
 
+/// Fallible byte streams that collect into a `Vec<u8>`.
 #[async_trait]
-pub(crate) trait IntoVec {
-    async fn into_vec(self) -> Result<Vec<u8>, TransportError>;
+pub trait IntoVec<E> {
+    /// Try to collect into `Vec<u8>`.
+    async fn into_vec(self) -> Result<Vec<u8>, E>;
 }
 
 #[async_trait]
-impl IntoVec for TransportStream {
-    async fn into_vec(self) -> Result<Vec<u8>, TransportError> {
+impl<S: Stream<Item = Result<Bytes, E>> + Send, E: Send> IntoVec<E> for S {
+    async fn into_vec(self) -> Result<Vec<u8>, E> {
         self.try_fold(Vec::new(), |mut acc, bytes| {
             acc.extend(bytes.as_ref());
             std::future::ready(Ok(acc))

tough/tests/test_utils.rs

@@ -9,6 +9,7 @@ use futures::TryStreamExt;
 use futures_core::Stream;
 use std::io::Read;
 use std::path::{Path, PathBuf};
+use tough::IntoVec;
 use url::Url;
 
 /// Utilities for tests. Not every test module uses every function, so we suppress unused warnings.
@@ -30,17 +31,10 @@ pub fn dir_url<P: AsRef<Path>>(path: P) -> Url {
 }
 
 /// Gets the goods from a read and makes a Vec
-pub async fn read_to_end<S, T, E>(mut stream: S) -> Vec<u8>
+pub async fn read_to_end<E, S>(mut stream: S) -> Vec<u8>
 where
-    S: Stream<Item = Result<T, E>> + Send,
-    T: AsRef<[u8]>,
-    E: std::error::Error,
+    E: std::fmt::Debug,
+    S: IntoVec<E>,
 {
-    stream
-        .try_fold(Vec::new(), |mut acc, bytes| {
-            acc.extend(bytes.as_ref());
-            std::future::ready(Ok(acc))
-        })
-        .await
-        .unwrap()
+    stream.into_vec().await.unwrap()
 }

tuftool/tests/test_utils.rs

@@ -3,9 +3,8 @@
 
 use assert_cmd::Command;
 use chrono::{Duration, Utc};
-use futures::TryStreamExt;
-use futures_core::Stream;
 use std::path::{Path, PathBuf};
+use tough::IntoVec;
 use url::Url;
 
 /// Utilities for tests. Not every test module uses every function, so we suppress unused warnings.
@@ -26,19 +25,12 @@ pub fn dir_url<P: AsRef<Path>>(path: P) -> Url {
 
 /// Returns a vector of bytes from any stream of byte results
 #[allow(unused)]
-pub async fn read_to_end<S, T, E>(mut stream: S) -> Vec<u8>
+pub async fn read_to_end<E, S>(mut stream: S) -> Vec<u8>
 where
-    S: Stream<Item = Result<T, E>> + Send,
-    T: AsRef<[u8]>,
-    E: std::error::Error,
+    E: std::fmt::Debug,
+    S: IntoVec<E>,
 {
-    stream
-        .try_fold(Vec::new(), |mut acc, bytes| {
-            acc.extend(bytes.as_ref());
-            std::future::ready(Ok(acc))
-        })
-        .await
-        .unwrap()
+    stream.into_vec().await.unwrap()
 }
 
 /// Creates a repository with expired timestamp metadata.