pub fn into_stream(
        self,
        token: stop_token::StopToken,
    ) -> impl Stream<Item = Result<Result<Event, ConfigError>, TimedOutError>>
           + Send
           + Sync
           + 'static
           + Unpin {
        let Self {
            setup_handle,
            flush_dir,
            ..
        } = self;
        let flush_id = Arc::new(AtomicU64::new(1));
        let flush_dir = Arc::new(flush_dir);
        let flush = Arc::new(Mutex::new(HashMap::<u64, oneshot::Sender<()>>::new()));

        // Wait until we have watched the flush directory, before we start processing
        // events

        // we need to map this to the correct error type
        let setup_stream = setup_handle.into_stream().map(|setup_result| {
            match setup_result.expect("globwatch setup thread panicked") {
                Ok(()) => Ok(WatcherCommand::Start),
                Err(e) => {
                    error!("failed to start server: {}", e);
                    Err(ConfigError::ServerFailedToStart)
                }
            }
        });

        Box::pin(
            (
                UnboundedReceiverStream::new(self.stream).map(Either::Left),
                setup_stream
                    .chain(UnboundedReceiverStream::new(self.config).map(Result::Ok))
                    .map(Either::Right),
            )
                .merge()
                // apply a filter_map, yielding only valid events and consuming config changes and
                // flushes
                .filter_map(move |f| {
                    let span = span!(tracing::Level::TRACE, "stream_processor");
                    let _ = span.enter();

                    // clone all the Arcs needed
                    let flush_id = flush_id.clone();
                    let flush_dir = flush_dir.clone();
                    let flush = flush.clone();

                    async move {
                        match f {
                            Either::Left(mut e) => {
                                // if we receive an event for a file in the flush dir, we need to
                                // remove it from the events list, and send a signal to the flush
                                // requester. flushes should not be considered as events.
                                for flush_id in e
                                    .paths
                                    .extract_if(.., |p| p.starts_with(flush_dir.as_path()))
                                    .filter_map(|p| {
                                        get_flush_id(
                                            p.strip_prefix(flush_dir.as_path())
                                                .expect("confirmed above"),
                                        )
                                    })
                                {
                                    trace!("flushing {:?}", flush);
                                    if let Some(tx) = flush
                                        .lock()
                                        .expect("only fails if holder panics")
                                        .remove(&flush_id)
                                    {
                                        // if this fails, it just means the requester has gone away
                                        // and we can ignore it
                                        tx.send(()).ok();
                                    }
                                }

                                // if we have any paths left on the event, yield it
                                if !e.paths.is_empty() {
                                    event!(parent: &span, Level::TRACE, "yielding {:?}", e);
                                    Some(Ok(e))
                                } else {
                                    None
                                }
                            }
                            Either::Right(Ok(WatcherCommand::Start)) => {
                                trace!("setting up watching flush directory has completed");
                                None
                            }
                            Either::Right(Ok(WatcherCommand::Flush(tx))) => {
                                // create file in flush dir
                                let flush_id = flush_id.fetch_add(1, Ordering::SeqCst);
                                let flush_file = flush_dir.join_component(&flush_id.to_string());
                                if let Err(e) = File::create(flush_file) {
                                    warn!("failed to create flush file: {}", e);
                                } else {
                                    flush
                                        .lock()
                                        .expect("only fails if holder panics")
                                        .insert(flush_id, tx);
                                }
                                None
                            }
                            Either::Right(Err(e)) => Some(Err(e)),
                        }
                    }
                })
                .timeout_at(token),
        )
    }
Convert the watcher into a stream of events, handling config changes and
flushing transparently.

This is implemented as a zipped stream which processes filesystem events
and config changes driven by the same stream. This allows us to ensure
that anything watching for filesystem is also propagating config changes
crates/turborepo-globwatch/src/lib.rs