Блог

У меня есть программа, похожая на следующую:

 struct MyEvent { /* some fields */ }
struct MyStruct { /* some fields */ }
struct MyStreamer { /* holds some state */ }

impl MyStreamer {
    pub fn stream_objects<'a, 'b: 'a>(
        amp;'a self,
        event_stream: Pin<Box<dyn Stream<Item = MyEvent>   Send   'b>>,
    ) -> Pin<Box<dyn Stream<Item = Arc<MyStruct>>   Send   'a>> { /* implementation */ }
}
 

Цель состоит в том, чтобы обрабатывать события и создавать поток MyStruct . Затем у меня есть два потребителя для потока MyStruct , и я изо всех сил пытаюсь его дублировать.

Я пытаюсь написать следующую функцию (также см. Мою попытку реализации):

 pub fn duplicate_stream<'a, 'b: 'a>(
    amp;'a self,
    struct_stream: Pin<Box<dyn Stream<Item = Arc<MyStruct>>   Send   'b>>,
) -> (
   Pin<Box<dyn Stream<Item = Arc<MyStruct>>   Send   'b>>,
   Pin<Box<dyn Stream<Item = Arc<MyStruct>>   Send   'b>>
) {
        let (s1, r1) = mpsc::unbounded::<Arc<MyStruct>>();
        let (s2, r2) = mpsc::unbounded::<Arc<MyStruct>>();

        let s = s1.fanout(s2);

        let _handle = tokio::spawn(async move { struct_stream.map(Ok).forward(s).await });

        (r1.boxed(), r2.boxed())
}
 

На данный момент мне сказали следующее:

     |
155 |         struct_stream: Pin<Box<dyn Stream<Item = Arc<MyStruct>>   Send   'b>>,
    |                               ----------------------------------------------- this data with lifetime `'b`...
...
165 |         let _handle = tokio::spawn(async move { struct_stream.map(Ok).forward(s).await });
    |                                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ ...is captured here...
    |
note: ...and is required to live as long as `'static` here
   --> *file name here*
    |
165 |         let _handle = tokio::spawn(async move { struct_stream.map(Ok).forward(s).await });
    |                       ^^^^^^^^^^^^
 

Я могу удалить время жизни, но затем static будет выведено, и я получаю ошибки в вызывающих абонентах.

Мне любопытно понять, какой наилучший способ — клонировать все элементы потока и получить два идентичных потока. Использование tokio::spawn mpsc каналов и, похоже, требует изменения большого времени жизни на static .

Я думаю, что это проблема XY

Компилятор правильный, и вам, вероятно, следует подумать о времени жизни, потому что разветвление в порядке.

tokio::spawn требуется 'static , и у вас есть 'b время жизни, указанное для вашего struct_stream . Может быть, обернуть struct_stream в Arc/Rc ?