Struct KMerge

pub struct KMerge<'a, C, D>
where
    C: Compare<D>,
{ /* private fields */ }

A Stream that merges multiple streams in user specified order.

HeapEntry is stored in a max-bin-heap and the order is decided by Peeked and C. Peeked::No is uninitialized state, and will be placed at top of the max-bin-heap, so that all streams will be peeked before returning any item.

When all streams are initialized, i.e., peeked.is_peeked() == true, the order is decided by C(&D, &D).

To do a merge sort, the stream should output items in the same order as C decided, otherwise the result is undefined.

Example:

use futures::stream::iter;
use stream_more::comparators::Ascending;
let got = KMerge::by(|a,b| a < b)
            .merge(iter([1,3]))
            .merge(iter([2,4]))
            .collect::<Vec<u64>>().await;
assert_eq!(vec![1, 2, 3, 4], got);

Implementations

impl<'a, F, D> KMerge<'a, FnCmp<F>, D>

where F: Fn(&D, &D) -> bool, FnCmp<F>: Compare<D>,

pub fn by(first: F) -> Self

Return an empty Stream adaptor KMerge that flattens Streams by merging them according to the given closure first().

The closure first() is called with two elements a, b and should return true if a is ordered before b.

If all base Streams are sorted according to first(), the result is sorted.

Example

use futures::stream::iter;
let got = KMerge::by(|a,b| a < b)
            .merge(iter([1,3]))
            .merge(iter([2,4]))
            .collect::<Vec<u64>>().await;
assert_eq!(vec![1, 2, 3, 4], got);

impl<'a, D, C> KMerge<'a, C, D>

where C: Compare<D>,

pub fn by_cmp(cmp: C) -> Self

Merge streams by a comparator Compare, if Compare::compare(a,b) returns Ordering::Greater, a will be chosen prior to b.

Example

Sort merge two streams in ascending order:

use futures::stream::iter;
use stream_more::comparators::Ascending;
let m = KMerge::by_cmp(Ascending).merge(iter([1,3])).merge(iter([2,4]));
let got = m.collect::<Vec<u64>>().await;
assert_eq!(vec![1, 2, 3, 4], got);

pub fn merge(self, stream: impl Stream<Item = D> + Send + 'a) -> Self

Append another stream to the merging stream.

This method can be called any time after the stream is created.

impl<'a, D> KMerge<'a, Descending, D>

where Descending: Compare<D>,

pub fn max() -> Self

Merge streams by choosing the maximum item from the streams, behaving like a max-heap.

Example

use futures::stream::iter;
let got = KMerge::max()
            .merge(iter([3,1]))
            .merge(iter([4,2]))
            .merge(iter([5]))
            .collect::<Vec<u64>>().await;
assert_eq!(vec![5, 4, 3, 2, 1], got);

impl<'a, D> KMerge<'a, Ascending, D>

where Ascending: Compare<D>,

pub fn min() -> Self

Merge streams by choosing the minimal item from the streams, behaving like a min-heap.

Example

use futures::stream::iter;
let got = KMerge::min()
            .merge(iter([1,3]))
            .merge(iter([2,4]))
            .merge(iter([5]))
            .collect::<Vec<u64>>().await;
assert_eq!(vec![1, 2, 3, 4, 5], got);

Trait Implementations

impl<'a, D, C> Stream for KMerge<'a, C, D>

where D: Unpin, C: Compare<D> + Unpin,

fn poll_next( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Option<Self::Item>>

Algorithm to merge streams by item order:

1. The first round is to peek the first item of each stream.

   At first the peeked is No and will be placed at the top of the heap. Once a stream is peeked, it will be moved to the tail of heap because Peeked::No > Peeked::Yes(T). And continue the loop, examine the new heap head, until all streams are peeked.

   So all streams will be peeked before returning any item.

2. Return the peeked item, poll the heap-head stream for the next item, then replace the currently peeked of the head stream with the next, let BinaryHeap adjust the order(if next is Some) or just remove the stream(if next is None).

   Do this until all of them are exhausted.

type Item = D

Values yielded by the stream.

fn size_hint(&self) -> (usize, Option<usize>)

Returns the bounds on the remaining length of the stream.