Pinning error using watcher or reflector (cannot be unpinned)

[floric]

Hey :)
I struggle with some Rust understanding regarding Pinning. The following code leads to an compilation error:

use futures::TryStreamExt;
use k8s_openapi::api::core::v1::ConfigMap;
use kube::{
    runtime::{
        reflector::{self, Store},
        watcher,
    },
    Api, Client,
};
use tracing::info;

pub async fn get_cm_reader() -> Result<Store<ConfigMap>, Box<dyn std::error::Error>> {
    let client = Client::try_default().await?;
    let api = Api::<ConfigMap>::default_namespaced(client);
    let (reader, writer) = reflector::store();
    let rf = reflector::reflector(writer, watcher(api, watcher::Config::default()));

    while let Some(cm_event) = rf.try_next().await? {
        info!("Received ConfigMap change: {:?}", cm_event);
    }

    Ok(reader)
}

leads to:

`{async block@reflector<ConfigMap, impl Stream<Item = Result<Event<ConfigMap>, Error>> + Send>::{closure#0}}` cannot be unpinned
the full name for the type has been written to 'xxx/explorer_lib-6a4db4070fe68568.long-type-13544658612713317214.txt'
consider using `--verbose` to print the full type name to the console
the full trait has been written to 'xxx/explorer_lib-6a4db4070fe68568.long-type-5654949343708741383.txt'
consider using the `pin!` macro
consider using `Box::pin` if you need to access the pinned value outside of the current scope
required for `AsyncStream<Result<Event<ConfigMap>, Error>, {async block@reflector<ConfigMap, ...>::{closure#0}}>` to implement `Unpin`

Why doesn't the try_next work there? I tried to search for examples or other issues/questions but couldn't find anything in this direction. :/

[clux]

Basically, it's because of Unpin requirements on the stream. To simplify it a bit, the futures used by the various stream implementations has a degree of self-referentiality and cannot be moved in async contexts / multi-threaded executors. This blog post is perhaps useful for the understanding.

To satisfy Unpin you can pin it to the stack with the stdlib's std::pin::pin macro, ala node_watcher, or, more safely (because this can consume a lot of stack size), you can put it on the heap via Box::pin, commonly invoked via StreamExt::boxed as seen in the crd_reflector ex.

[floric]

Thanks a lot for the helpful links. I will have a look into it.