Merge branch 'substack-non-buffering'

src/blob.rs

@@ -152,14 +152,14 @@ impl BlobHeader {
 
 /// A reader for PBF files that allows iterating over `Blob`s.
 #[derive(Clone, Debug)]
-pub struct BlobReader<R: Read> {
+pub struct BlobReader<R: Read + Send> {
     reader: R,
     /// Current reader offset in bytes from the start of the stream.
     offset: Option<ByteOffset>,
     last_blob_ok: bool,
 }
 
-impl<R: Read> BlobReader<R> {
+impl<R: Read + Send> BlobReader<R> {
     /// Creates a new `BlobReader`.
     ///
     /// # Example
@@ -258,7 +258,7 @@ impl BlobReader<BufReader<File>> {
     }
 }
 
-impl<R: Read> Iterator for BlobReader<R> {
+impl<R: Read + Send> Iterator for BlobReader<R> {
     type Item = Result<Blob>;
 
     fn next(&mut self) -> Option<Self::Item> {
@@ -294,7 +294,7 @@ impl<R: Read> Iterator for BlobReader<R> {
     }
 }
 
-impl<R: Read + Seek> BlobReader<R> {
+impl<R: Read + Seek + Send> BlobReader<R> {
     /// Creates a new `BlobReader` from the given reader that is seekable and will be initialized
     /// with a valid offset.
     ///

src/indexed.rs

@@ -38,12 +38,12 @@ struct BlobInfo {
 /// Allows filtering elements and iterating over their dependencies.
 /// It chooses an efficient method for navigating the PBF structure to achieve this in reasonable
 /// time and with reasonable memory.
-pub struct IndexedReader<R: Read + Seek> {
+pub struct IndexedReader<R: Read + Seek + Send> {
     reader: BlobReader<R>,
     index: Vec<BlobInfo>,
 }
 
-impl<R: Read + Seek> IndexedReader<R> {
+impl<R: Read + Seek + Send> IndexedReader<R> {
     /// Creates a new `IndexedReader`.
     ///
     /// # Example

src/reader.rs

@@ -10,11 +10,11 @@ use std::path::Path;
 
 /// A reader for PBF files that gives access to the stored elements: nodes, ways and relations.
 #[derive(Clone, Debug)]
-pub struct ElementReader<R: Read> {
+pub struct ElementReader<R: Read + Send> {
     blob_iter: BlobReader<R>,
 }
 
-impl<R: Read> ElementReader<R> {
+impl<R: Read + Send> ElementReader<R> {
     /// Creates a new `ElementReader`.
     ///
     /// # Example
@@ -68,11 +68,9 @@ impl<R: Read> ElementReader<R> {
     where
         F: for<'a> FnMut(Element<'a>),
     {
-        let blobs = self.blob_iter.collect::<Result<Vec<_>>>()?;
-
         //TODO do something useful with header blocks
-        for blob in &blobs {
-            match blob.decode() {
+        for blob in self.blob_iter {
+            match blob?.decode() {
                 Ok(BlobDecode::OsmHeader(_)) | Ok(BlobDecode::Unknown(_)) => {}
                 Ok(BlobDecode::OsmData(block)) => {
                     block.for_each_element(&mut f);
@@ -125,11 +123,9 @@ impl<R: Read> ElementReader<R> {
         ID: Fn() -> T + Sync + Send,
         T: Send,
     {
-        let blobs = self.blob_iter.collect::<Result<Vec<_>>>()?;
-
-        blobs
-            .into_par_iter()
-            .map(|blob| match blob.decode() {
+        self.blob_iter
+            .par_bridge()
+            .map(|blob| match blob?.decode() {
                 Ok(BlobDecode::OsmHeader(_)) | Ok(BlobDecode::Unknown(_)) => Ok(identity()),
                 Ok(BlobDecode::OsmData(block)) => Ok(block
                     .elements()