vortex_io/
kanal_ext.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright the Vortex contributors
3
4use async_stream::stream;
5use futures::Stream;
6use kanal::AsyncReceiver;
7
8pub trait KanalExt<T> {
9    fn into_stream(self) -> impl Stream<Item = T>;
10}
11
12impl<T> KanalExt<T> for AsyncReceiver<T> {
13    fn into_stream(self) -> impl Stream<Item = T> {
14        stream! {
15            // The Err case indicates the sender / channel has been closed so we terminate
16            // the stream.
17            while let Ok(next) = self.recv().await {
18                yield next
19            }
20        }
21    }
22}
vortex_io/kanal_ext.rs

vortex_io/
kanal_ext.rs
