stream: provide StreamMap utility
#2185
Conversation
`StreamMap` is similar to `StreamExt::merge` in that it combines source streams into a single merged stream that yields values in the order that they arrive from the source streams. However, `StreamMap` has a lot more flexibility in usage patterns. `StreamMap` can: - Merge an arbitrary number of streams. - Track which source stream the value was received from. - Handle inserting and removing streams from the set of managed streams at any point during iteration. All source streams held by `StreamMap` are indexed using a key. This key is included with the value when a source stream yields a value. The key is also used to remove the stream from the `StreamMap` before the stream has completed streaming. Because the `StreamMap` API moves streams during runtime, both streams and keys must be `Unpin`. In order to insert a `!Unpin` stream into a `StreamMap`, use `pin!` to pin the stream to the stack or `Box::pin` to pin the stream in the heap.
carllerche
requested review from
hawkw,
jebrosen,
LucioFranco,
Matthias247 and
vorot93
There was a problem hiding this comment.
I think it would be good to have a realistic example of why it can be helpful to know which stream produced a value, and possibly a comparison to approaches where the stream identifies itself in the values it produces (e.g. with stream.map(|v| ("stream 1", v)).
|
@jebrosen I added a more involved example. Is that what you were thinking? |
| where | ||
| K: Hash + Eq, | ||
| { | ||
| let ret = self.remove(&k); |
There was a problem hiding this comment.
This makes insertion of n streams O(n^2).
| /// assert!(map.remove(&1).is_some()); | ||
| /// assert!(map.remove(&1).is_none()); | ||
| /// ``` | ||
| pub fn remove<Q: ?Sized>(&mut self, k: &Q) -> Option<V> |
There was a problem hiding this comment.
I think the docs should mention that this is an expensive operation.
There was a problem hiding this comment.
The struct docs include a note, is this sufficient?
|
@Darksonn re: complexity, the thing to keep in mind is, the current implementation needs to scan regardless of the backend structure due to how polling scans all entries. Switching to |
|
I didn't actually notice the note in the docs above. That said, one could have a I'm okay with it as long as there's a note, although I think it is somewhat unfortunate. Edit: It would still have to be a |
|
It's a bit more complicated than that. One has to be able to get the entry key while iterating as well. This means, to use a HashMap, the key would have to be cloned and stored twice, once in the HashMap and once in the vec. |
StreamMapis similar toStreamExt::mergein that it combines sourcestreams into a single merged stream that yields values in the order that
they arrive from the source streams. However,
StreamMaphas a lot moreflexibility in usage patterns.
StreamMapcan:at any point during iteration.
All source streams held by
StreamMapare indexed using a key. This keyis included with the value when a source stream yields a value. The key
is also used to remove the stream from the
StreamMapbefore the streamhas completed streaming.
Because the
StreamMapAPI moves streams during runtime, both streamsand keys must be
Unpin. In order to insert a!Unpinstream into aStreamMap, usepin!to pin the stream to the stack orBox::pintopin the stream in the heap.