fn live_local()

in below/src/main.rs [1011:1111]

• 93 lines of code
• 8 McCabe index (conditional complexity)

fn live_local(
    logger: slog::Logger,
    errs: Receiver<Error>,
    interval: Duration,
    debug: bool,
    below_config: &BelowConfig,
) -> Result<()> {
    match bump_memlock_rlimit() {
        Err(e) => {
            warn!(
                logger,
                #"V",
                "Failed to initialize BPF: {}. Data collection will be degraded. \
                You can ignore this warning or try to run with sudo.",
                &e
            );
        }
        _ => {}
    };

    let (exit_buffer, bpf_errs) = start_exitstat(logger.clone(), debug);
    let mut bpf_err_warned = false;

    let mut collector = model::Collector::new(
        logger.clone(),
        model::CollectorOptions {
            cgroup_root: below_config.cgroup_root.clone(),
            exit_data: exit_buffer,
            ..Default::default()
        },
    );
    logutil::set_current_log_target(logutil::TargetLog::File);
    // Prepare advance obj for pause mode
    let mut adv = new_advance_local(
        logger.clone(),
        below_config.store_dir.clone(),
        SystemTime::now(),
    );
    adv.initialize();
    let mut view = view::View::new_with_advance(
        collector.collect_and_update_model()?,
        view::ViewMode::Live(Rc::new(RefCell::new(adv))),
    );

    let sink = view.cb_sink().clone();

    thread::Builder::new()
        .name("live_collector".to_owned())
        .spawn(move || {
            loop {
                if !bpf_err_warned {
                    bpf_err_warned = check_for_exitstat_errors(
                        &logger,
                        bpf_errs
                            .as_ref()
                            .expect("Failed to unwrap bpf_errs receiver"),
                    );
                }

                // Rely on timeout to guarantee interval between samples
                match errs.recv_timeout(interval) {
                    Ok(e) => {
                        error!(logger, "{:#}", e);
                        sink.send(Box::new(|c| c.quit()))
                            .expect("Failed to stop view");
                        return;
                    }
                    Err(RecvTimeoutError::Disconnected) => {
                        error!(logger, "error channel disconnected");
                        sink.send(Box::new(|c| c.quit()))
                            .expect("Failed to stop view");
                        return;
                    }
                    Err(RecvTimeoutError::Timeout) => {}
                };

                match collector.collect_and_update_model() {
                    Ok(model) => {
                        // Error only happens if the other side disconnected - just terminate the thread
                        let data_plane = Box::new(move |s: &mut Cursive| {
                            let view_state = s.user_data::<ViewState>().expect("user data not set");

                            // When paused, no need to update model
                            if !view_state.is_paused() {
                                view_state.update(model);
                            }
                        });
                        if sink.send(data_plane).is_err() {
                            return;
                        }
                    }
                    Err(e) => {
                        error!(logger, "{:#}", e);
                    }
                }
            }
        })
        .expect("Failed to spawn thread");

    view.run()
}