message: Only box-pin reconcilers when using the Controller builder
diff: @@ -11,13 +11,13 @@ use crate::{
  };
  use derivative::Derivative;
  use futures::{
  - channel,
  + channel, future,
  stream::{self, SelectAll},
  - SinkExt, Stream, StreamExt, TryFuture, TryFutureExt, TryStream, TryStreamExt,
  + FutureExt, SinkExt, Stream, StreamExt, TryFuture, TryFutureExt, TryStream, TryStreamExt,
  };
  use kube::api::{Api, ListParams, Meta};
  use serde::de::DeserializeOwned;
  -use snafu::{futures::TryStreamExt as SnafuTryStreamExt, Backtrace, OptionExt, ResultExt, Snafu};
  +use snafu::{futures::TryStreamExt as SnafuTryStreamExt, Backtrace, ResultExt, Snafu};
  use std::{sync::Arc, time::Duration};
  use stream::BoxStream;
  use tokio::time::Instant;
  @@ -144,7 +144,7 @@ pub fn applier<K, QueueStream, ReconcilerFut, T>(
  ) -> impl Stream<Item = Result<(ObjectRef<K>, ReconcilerAction), Error<ReconcilerFut::Error, QueueStream::Error>>>
  where
  K: Clone + Meta + 'static,
  - ReconcilerFut: TryFuture<Ok = ReconcilerAction>,
  + ReconcilerFut: TryFuture<Ok = ReconcilerAction> + Unpin,
  ReconcilerFut::Error: std::error::Error + 'static,
  QueueStream: TryStream<Ok = ObjectRef<K>>,
  QueueStream::Error: std::error::Error + 'static,
  @@ -167,11 +167,15 @@ where
  move |s| {
  Runner::new(scheduler(s), move |obj_ref| {
  let obj_ref = obj_ref.clone();
  - let obj = store.get(&obj_ref).context(ObjectNotFound {
  - obj_ref: obj_ref.clone(),
  - });
  - let reconciler_fut = obj.map(|obj| reconciler(obj, context.clone()));
  - Box::pin(async move { Ok((obj_ref, reconciler_fut?.into_future().await)) })
  + match store.get(&obj_ref) {
  + Some(obj) => reconciler(obj, context.clone())
  + .into_future()
  + // Reconciler errors are OK from the applier's PoV, we need to apply the error policy
  + // to them separately
  + .map(|res| Ok((obj_ref, res)))
  + .left_future(),
  + None => future::err(ObjectNotFound { obj_ref }.build()).right_future(),
  + }
  })
  .context(SchedulerDequeueFailed)
  .map(|res| res.and_then(|x| x))
  @@ -345,7 +349,7 @@ where
  /// with a configurable `Context`.
  pub fn run<ReconcilerFut, T>(
  self,
  - reconciler: impl FnMut(K, Context<T>) -> ReconcilerFut,
  + mut reconciler: impl FnMut(K, Context<T>) -> ReconcilerFut,
  error_policy: impl FnMut(&ReconcilerFut::Error, Context<T>) -> ReconcilerAction,
  context: Context<T>,
  ) -> impl Stream<Item = Result<(ObjectRef<K>, ReconcilerAction), Error<ReconcilerFut::Error, watcher::Error>>>
  @@ -354,7 +358,13 @@ where
  ReconcilerFut: TryFuture<Ok = ReconcilerAction>,
  ReconcilerFut::Error: std::error::Error + 'static,
  {
  - applier(reconciler, error_policy, context, self.reader, self.selector)
  + applier(
  + move |obj, ctx| Box::pin(reconciler(obj, ctx).into_future()),
  + error_policy,
  + context,
  + self.reader,
  + self.selector,
  + )
  }
  }
  