Это частичные решения, которые я нашел, но которых по разным причинам не хватает.
Использование комбинаторов с изменчивостью внутренней части
Мне не нравится это решение, потому что я считаю, что изменчивость внутренней части должна не требуется для этой общей проблемы, но здесь требуется, потому что контролер заимствований не знает, как чередуются вызовы замыканий.
use futures::{stream, Stream, StreamExt};
use std::collections::VecDeque;
fn x(v: i32) -> impl Stream<Item = i32> {
stream::iter((0..v).rev())
}
use std::{cell::RefCell, rc::Rc};
fn y0() -> impl Stream<Item = i32> {
let to_visit = Rc::new(RefCell::new(VecDeque::from(vec![5])));
let to_visit_b = to_visit.clone();
stream::unfold(to_visit, |to_visit| async {
let i = to_visit.borrow_mut().pop_back()?;
Some((x(i), to_visit))
})
.flatten()
.inspect(move |&x| {
if x % 2 != 0 {
to_visit_b.borrow_mut().push_front(x);
}
})
}
#[tokio::main]
async fn main() {
y0().for_each(|v| async move {
println!("v: {}", v);
})
.await;
}
детская площадка
Пользовательская реализация Stream::poll_next
Мне не нравится это решение, потому что оно многословно и требует хитрого unsafe кода, который трудно рассуждать (я даже не уверен, что у меня правильно! )
use futures::{stream, Stream, StreamExt};
use std::collections::VecDeque;
fn x(v: i32) -> impl Stream<Item = i32> {
stream::iter((0..v).rev())
}
use std::{
pin::Pin,
task::{Context, Poll},
};
struct X<St, C, R, S>
where
C: Fn(&mut St) -> Option<S>,
R: Fn(&mut St, &mut S::Item),
S: Stream,
{
state: St,
create: C,
review: R,
current: Option<S>,
}
impl<St, C, R, S> Stream for X<St, C, R, S>
where
C: Fn(&mut St) -> Option<S>,
R: Fn(&mut St, &mut S::Item),
S: Stream,
{
type Item = S::Item;
fn poll_next(self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let (state, create, review, current) = unsafe {
let Self {
state,
create,
review,
current,
} = self.get_unchecked_mut();
(state, create, review, current)
};
loop {
if let Some(current) = current {
let v = unsafe { futures::ready!(Pin::new_unchecked(current).poll_next(ctx)) };
if let Some(mut v) = v {
review(state, &mut v);
return Poll::Ready(Some(v));
}
}
*current = create(state);
if current.is_none() {
return Poll::Ready(None);
}
}
}
}
fn y1() -> impl Stream<Item = i32> {
X {
state: VecDeque::from(vec![5]),
create: |to_visit| {
let i = to_visit.pop_back()?;
Some(x(i))
},
review: |to_visit, &mut x| {
if x % 2 != 0 {
to_visit.push_front(x);
}
},
current: None,
}
}
#[tokio::main]
async fn main() {
y1().for_each(|v| async move {
println!("v: {}", v);
})
.await;
}
игровая площадка
Использование каналов (не работает)
Это не работает, поскольку отправитель никогда не удаляется потому что получатель никогда не отбрасывается, потому что отправитель никогда не отбрасывается ...
Помимо неработоспособности, у этого есть ряд недостатков:
- Состояние должно неявно быть очередью ( который соответствует тому, что я хочу сделать, но не очень общее).
- Требуется, чтобы моя функция сама стала
async до pu sh начальным значением для посещения. - Я должен обработать ошибки, которые кажутся несущественными.
- Я должен клонировать
Sender внутри then замыкания.
use futures::{stream, Stream, StreamExt};
fn x(v: i32) -> impl Stream<Item = i32> {
stream::iter((0..v).rev())
}
use futures::channel::mpsc;
use futures::sink::SinkExt;
async fn y2() -> impl Stream<Item = i32> {
let (mut tx, rx) = mpsc::unbounded();
tx.send(5).await.unwrap();
rx.map(x).flatten().then(move |x| {
let mut tx = tx.clone();
async move {
if x % 2 != 0 {
tx.send(x).await.unwrap();
}
x
}
})
}
#[tokio::main]
async fn main() {
y2().await
.for_each(|v| async move {
println!("v: {}", v);
})
.await;
}
детская площадка