How does futures::stream::TryStreamExt::into_stream flatten errors into the stream vs map_err?
TryStreamExt::into_stream converts a TryStream (stream of Result<Item, Error>) into a regular Stream (stream of Item), emitting both successful items and errors as stream values, while map_err keeps the Result wrapper intact and only transforms the error type. The key distinction is that into_stream flattens the error into the stream's item type using Either<Item, Error>, allowing errors to be processed as regular stream elements rather than terminating the stream with an error, whereas map_err maintains the Result structure and merely changes the error type. This architectural difference means into_stream enables error handling as part of normal stream processing, while map_err is for transforming error types within the standard Result error propagation model.
Understanding TryStream
use futures::stream::{Stream, TryStream};
use std::pin::Pin;
// A TryStream is a stream of Result<Item, Error>
// It's used for fallible streams like I/O operations
// Regular Stream: produces items, can complete or fail
// Stream<Item = T, Error = E> -> produces Result<T, E>
// TryStream: a stream that yields Result<T, E>
// Item type is Result<T, E>, not T directly
// The stream can produce successful items or an error
// Example: reading lines from a file
// - Success: yields Ok(line) for each line
// - Error: yields Err(error) and typically stopsA TryStream wraps each item in a Result, allowing the stream to communicate failures.
map_err: Transforming Error Types
use futures::stream::{self, TryStreamExt};
async fn map_err_example() {
// map_err transforms the error type while keeping Result structure
let stream = stream::iter(vec![
Ok(1i32),
Ok(2),
Err("error"),
Ok(3),
]);
// Transform error type from &str to String
let mapped: impl TryStream<Ok = i32, Error = String> = stream
.map_err(|e| e.to_string());
// The stream still yields Result<i32, String>
// Structure is preserved: Ok(items) and Err(errors)
// Collecting requires handling the Result
let results: Result<Vec<i32>, String> = mapped
.try_collect()
.await;
// Result: Err("error") - stream terminates on first error
}map_err transforms errors but maintains the Result structure and error propagation semantics.
into_stream: Flattening Errors
use futures::stream::{Stream, TryStreamExt};
use futures::stream;
async fn into_stream_example() {
// into_stream converts TryStream to Stream
// Both Ok and Err become stream items
let try_stream = stream::iter(vec![
Ok(1i32),
Ok(2),
Err("error".to_string()),
Ok(3),
]);
// Convert to regular stream
let stream: impl Stream<Item = Result<i32, String>> = try_stream
.into_stream();
// Wait - this still yields Result!
// The difference is in how we process it
// With into_stream, we can use StreamExt methods
// that don't require TryStream (like map, filter, etc.)
use futures::StreamExt;
// Process as regular stream
let items: Vec<_> = stream.collect().await;
// items: [Ok(1), Ok(2), Err("error"), Ok(3)]
// All items (both Ok and Err) are collected
// Stream doesn't terminate on error
}into_stream converts to a regular Stream, enabling different processing approaches.
The Key Difference in Error Handling
use futures::stream::{self, TryStreamExt, StreamExt};
async fn error_handling_difference() {
let data = vec![
Ok(1i32),
Ok(2),
Err("failure"),
Ok(3),
];
// With map_err + try_collect:
// Stream terminates on first error
let result: Result<Vec<i32>, &str> = stream::iter(data.clone())
.map_err(|e| e) // Identity, but shows pattern
.try_collect()
.await;
// result: Err("failure")
// Items 1 and 2 are lost, item 3 never reached
// With into_stream + collect:
// Stream continues after error
let items: Vec<Result<i32, &str>> = stream::iter(data)
.into_stream()
.collect()
.await;
// items: [Ok(1), Ok(2), Err("failure"), Ok(3)]
// All items preserved
}map_err maintains Result termination semantics; into_stream allows processing past errors.
Processing Errors as Items with Either
use futures::stream::{self, TryStreamExt, StreamExt};
use either::Either;
async fn either_pattern() {
// A common pattern is using into_stream + map to get Either
let data = vec![
Ok(1i32),
Ok(2),
Err("error"),
Ok(3),
];
// Convert Result to Either
let stream = stream::iter(data)
.into_stream()
.map(|result| match result {
Ok(item) => Either::Left(item),
Err(e) => Either::Right(e),
});
// Now we have Stream<Item = Either<i32, String>>
// Process success and error branches differently
let (mut successes, mut errors) = (Vec::new(), Vec::new());
let results: Vec<_> = stream.collect().await;
for either in results {
match either {
Either::Left(item) => successes.push(item),
Either::Right(err) => errors.push(err),
}
}
// successes: [1, 2, 3]
// errors: ["error"]
}into_stream enables treating errors as first-class stream items.
When to Use map_err
use futures::stream::{self, TryStreamExt};
async fn when_to_use_map_err() {
// Use map_err when:
// 1. You want to transform error types
// 2. You're propagating errors up the call stack
// 3. You want to maintain TryStream semantics
let stream = stream::iter(vec![
Ok(1i32),
Err("io error"),
]);
// Transform error to a more descriptive type
let mapped = stream.map_err(|e| {
format!("Stream error: {}", e)
});
// Continue using TryStream operations
let result: Result<Vec<i32>, String> = mapped
.try_collect()
.await;
// Use map_err for:
// - Converting to a common error type
// - Adding context to errors
// - Wrapping errors in custom types
}Use map_err for error type transformation within TryStream processing.
When to Use into_stream
use futures::stream::{self, TryStreamExt, StreamExt};
async fn when_to_use_into_stream() {
// Use into_stream when:
// 1. You want to continue processing after errors
// 2. You need to collect both successes and errors
// 3. You want to use StreamExt methods not available on TryStream
let stream = stream::iter(vec![
Ok(1i32),
Err("error"),
Ok(2),
Err("another"),
Ok(3),
]);
// into_stream allows processing all items
let all_items: Vec<_> = stream
.into_stream()
.collect()
.await;
// Separate successes and errors
let (successes, errors): (Vec<_>, Vec<_>) = all_items
.into_iter()
.partition(Result::is_ok);
let successes: Vec<_> = successes
.into_iter()
.map(|r| r.unwrap())
.collect();
let errors: Vec<_> = errors
.into_iter()
.map(|r| r.unwrap_err())
.collect();
// successes: [1, 2, 3]
// errors: ["error", "another"]
}Use into_stream when you need to process all items regardless of errors.
Combining Both
use futures::stream::{self, TryStreamExt, StreamExt};
async fn combined_pattern() {
let data = vec![
Ok::<i32, &str>(1),
Err("error1"),
Ok(2),
Err("error2"),
Ok(3),
];
// Common pattern: transform error type, then flatten
let processed = stream::iter(data)
.map_err(|e| format!("Wrapped: {}", e)) // Transform error type
.into_stream() // Flatten to Stream
.filter_map(|result| async move {
// Filter out errors, log them, keep successes
match result {
Ok(item) => Some(item),
Err(e) => {
println!("Logging error: {}", e);
None
}
}
});
let items: Vec<i32> = processed.collect().await;
// items: [1, 2, 3]
// Errors were logged but didn't stop processing
}Combine map_err for type conversion with into_stream for flexible processing.
Practical Example: Reading Lines
use futures::stream::{self, TryStreamExt, StreamExt};
use tokio::io::AsyncBufReadExt;
async fn read_lines_example() {
// Reading lines from a file is a TryStream
// Each line could fail (e.g., invalid UTF-8)
// Hypothetical async lines reader
let lines = vec![
Ok("line 1"),
Ok("line 2"),
Err("invalid utf-8 sequence"),
Ok("line 4"),
];
let stream = stream::iter(lines);
// With TryStream methods:
// - try_filter, try_fold, try_collect all stop on error
let result: Result<Vec<&str>, &str> = stream.clone()
.try_collect()
.await;
// result: Err("invalid utf-8 sequence")
// Lines 1-2 are lost
// With into_stream:
// - Continue processing after error
// - Process valid lines, log errors
let all_lines: Vec<_> = stream
.into_stream()
.collect()
.await;
// all_lines: [Ok("line 1"), Ok("line 2"), Err(...), Ok("line 4")]
// Extract valid lines
let valid: Vec<_> = all_lines
.into_iter()
.filter_map(|r| r.ok())
.collect();
// valid: ["line 1", "line 2", "line 4"]
}into_stream allows continuing past parse errors in streaming data.
StreamExt vs TryStreamExt Methods
use futures::stream::{self, TryStreamExt, StreamExt};
async fn method_availability() {
// TryStreamExt methods (for TryStream):
// - try_next(): Get next Result<Item, Error>
// - try_filter(): Filter Ok items, error propagates
// - try_map(): Map Ok items, error propagates
// - try_fold(): Fold Ok items, error propagates
// - try_collect(): Collect Ok items into Result
// - try_for_each(): Process each Ok item, error propagates
// All terminate on first error
// StreamExt methods (for regular Stream):
// - next(): Get next Option<Item>
// - filter(): Filter items (Item is not Result)
// - map(): Map items (Item is not Result)
// - fold(): Fold items
// - collect(): Collect all items
// - for_each(): Process each item
// Continue processing after "errors" if you handle them in Item
// into_stream converts TryStream to Stream
// This lets you use StreamExt methods
// With Result<Item, Error> as the Item type
let try_stream = stream::iter(vec![Ok(1), Err("fail"), Ok(2)]);
// Can't use StreamExt::filter on TryStream directly
// let filtered = try_stream.filter(|x| async { x > 0 }); // Won't compile
// After into_stream, can use StreamExt methods
let filtered = try_stream
.into_stream()
.filter_map(|result| async move {
match result {
Ok(item) if item > 0 => Some(Ok(item)),
Ok(_) => None,
Err(e) => Some(Err(e)),
}
});
}into_stream enables StreamExt methods that don't terminate on errors.
Error Propagation Semantics
use futures::stream::{self, TryStreamExt, StreamExt};
async fn propagation_semantics() {
let data = vec![
Ok(1),
Ok(2),
Err("error"),
Ok(3),
];
// TryStream: errors propagate outward
// The "try" prefix indicates short-circuit behavior
// try_collect stops at first error
let result1: Result<Vec<i32>, &str> = stream::iter(data.clone())
.try_collect()
.await;
// result1: Err("error")
// into_stream: errors become items
// No short-circuit behavior
// collect gets all items
let items: Vec<Result<i32, &str>> = stream::iter(data)
.into_stream()
.collect()
.await;
// items: [Ok(1), Ok(2), Err("error"), Ok(3)]
// The choice depends on error handling strategy:
// - Fail-fast: Use TryStream methods
// - Process all: Use into_stream + Stream methods
}TryStream methods fail-fast; into_stream allows processing all results.
Type Signatures
use futures::stream::{Stream, TryStream};
use std::future::Future;
// TryStreamExt::map_err signature:
// fn map_err<E2, F>(self, f: F) -> MapErr<Self, F>
// where
// F: FnMut(Self::Error) -> E2,
// Output: TryStream<Ok = Self::Ok, Error = E2>
// The stream is still a TryStream, just with different Error type
// TryStreamExt::into_stream signature:
// fn into_stream(self) -> IntoStream<Self>
// Output: Stream<Item = Result<Self::Ok, Self::Error>>
// The stream is now a regular Stream, Item is Result
// Key difference:
// map_err: TryStream -> TryStream (different error type)
// into_stream: TryStream -> Stream (Result becomes Item)The type signatures show that map_err preserves TryStream while into_stream converts to Stream.
Use Case: Multiple Error Types
use futures::stream::{self, TryStreamExt, StreamExt};
#[derive(Debug)]
enum StreamError {
Parse(String),
Io(String),
}
async fn multiple_error_types() {
// With map_err, unify error types
let data = vec![
Ok::<i32, &str>(1),
Err("parse error"),
Ok(2),
];
// Convert &str to StreamError::Parse
let unified = stream::iter(data)
.map_err(|e| StreamError::Parse(e.to_string()));
// Now we have TryStream<Ok = i32, Error = StreamError>
let result: Result<Vec<i32>, StreamError> = unified
.try_collect()
.await;
// Alternative: flatten and handle as items
let items: Vec<_> = stream::iter(vec![
Ok::<i32, &str>(1),
Err("parse error"),
Ok(2),
])
.into_stream()
.collect()
.await;
// Process each Result individually
for result in items {
match result {
Ok(n) => println!("Success: {}", n),
Err(e) => println!("Error: {}", e),
}
}
}map_err unifies error types; into_stream allows per-item processing.
Synthesis
map_err characteristics:
// - Preserves TryStream type
// - Transforms error type: TryStream<Ok=T, Error=E1> -> TryStream<Ok=T, Error=E2>
// - Maintains error propagation semantics (short-circuit)
// - Use with try_collect, try_fold, etc.
// - Terminates on first error
// - Good for: error type conversion, error wrapping
let stream = try_stream.map_err(|e| format!("Error: {}", e));
// Type: TryStream<Ok = Item, Error = String>into_stream characteristics:
// - Converts TryStream to Stream
// - Result<Item, Error> becomes the Stream Item type
// - Enables StreamExt methods
// - Errors become stream items, not stream terminators
// - Processing continues after errors
// - Good for: collecting all results, processing errors as data
let stream = try_stream.into_stream();
// Type: Stream<Item = Result<Item, Error>>Error handling comparison:
// TryStream + map_err:
// - Fail-fast error handling
// - Error propagates to caller
// - try_collect returns Result<Vec<T>, E>
// - Stop on first error
// Stream + into_stream:
// - Process-all error handling
// - Errors are data, not control flow
// - collect returns Vec<Result<T, E>>
// - Continue after errors
// Choose based on error handling strategy:
// - Fail-fast: Use TryStream methods
// - Process-all: Use into_stream + handle ResultsKey insight: TryStreamExt::into_stream and map_err represent two fundamentally different approaches to error handling in streams. map_err maintains the TryStream abstraction where errors are exceptional conditions that propagate outward and terminate processingāthis matches Rust's ? operator semantics and is appropriate when errors should abort the operation. into_stream flattens the Result into the stream's item type, treating errors as regular data elementsāthis enables continued processing after errors and allows using StreamExt methods that don't have short-circuit semantics. The choice is architectural: map_err when you want error type transformation within a fail-fast model, into_stream when you want to handle errors as part of normal stream processing. A common pattern combines both: try_stream.map_err(transform).into_stream().filter_map(handle) to transform error types, then process all items including errors.