5957916370

Committed 24 Aug 2023 12:27AM UTC coverage: 75.783% (-0.08%) from 75.86%

Build # 5957916370

Build Type

push

github

Committed by

web-flow

Commit Message

fix: introduce build states and fix inconsistencies (#1902)

* chore: return if state cant be sent

* chore: remove broadcasting state when stopped

* chore: send build message

* chore: fix build

Run Details

84 of 84 new or added lines in 7 files covered. (100.0%)

46996 of 62014 relevant lines covered (75.78%)

73508.46 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/dozer-cli/src/live/watcher.rs

use std::{sync::Arc, time::Duration};

use crate::shutdown::ShutdownReceiver;

use super::{state::LiveState, LiveError};

use crate::live::state::BroadcastType;
use dozer_types::log::info;
use notify::{RecursiveMode, Watcher};
use notify_debouncer_full::new_debouncer;
use tokio::runtime::Runtime;

pub async fn watch(
    runtime: &Arc<Runtime>,
    state: Arc<LiveState>,
    shutdown: ShutdownReceiver,
) -> Result<(), LiveError> {
    // setup debouncer
    let (tx, rx) = std::sync::mpsc::channel();

    let dir: std::path::PathBuf = std::env::current_dir()?;
    let mut debouncer = new_debouncer(Duration::from_millis(500), None, tx)?;

    let watcher = debouncer.watcher();

    watcher.watch(dir.as_path(), RecursiveMode::NonRecursive)?;

    let additional_paths = vec![dir.join("sql")];

    for path in additional_paths {
        let _ = watcher.watch(path.as_path(), RecursiveMode::NonRecursive);
    }

    debouncer
        .cache()
        .add_root(dir.as_path(), RecursiveMode::Recursive);

    let running = shutdown.get_running_flag().clone();
    loop {
        let event = rx.recv_timeout(Duration::from_millis(100));
        match event {
            Ok(result) => match result {
                Ok(_events) => {
                    build(runtime.clone(), state.clone()).await;
                }
                Err(errors) => errors.iter().for_each(|error| info!("{error:?}")),
            },
            Err(e) => {
                if !running.load(std::sync::atomic::Ordering::Relaxed) {
                    break;
                }
                if e == std::sync::mpsc::RecvTimeoutError::Disconnected {
                    break;
                }
            }
        }
    }

    Ok(())
}

async fn build(runtime: Arc<Runtime>, state: Arc<LiveState>) {
    state.broadcast(BroadcastType::Start).await;
    if let Err(res) = state.build(runtime).await {
        let message = res.to_string();
        state.broadcast(BroadcastType::Failed(message)).await;
    } else {
        state.broadcast(BroadcastType::Success).await;
    }
}

1	use std::{sync::Arc, time::Duration};
2
3	use crate::shutdown::ShutdownReceiver;
4
5	use super::{state::LiveState, LiveError};
6
7	use crate::live::state::BroadcastType;
8	use dozer_types::log::info;
9	use notify::{RecursiveMode, Watcher};
10	use notify_debouncer_full::new_debouncer;
11	use tokio::runtime::Runtime;
12		×
13	pub async fn watch(	×
14	runtime: &Arc<Runtime>,	×
15	state: Arc<LiveState>,	×
16	shutdown: ShutdownReceiver,	×
17	) -> Result<(), LiveError> {	×
18	// setup debouncer	×
19	let (tx, rx) = std::sync::mpsc::channel();	×
20		×
21	let dir: std::path::PathBuf = std::env::current_dir()?;	×
22	let mut debouncer = new_debouncer(Duration::from_millis(500), None, tx)?;	×
23		×
24	let watcher = debouncer.watcher();	×
25		×
26	watcher.watch(dir.as_path(), RecursiveMode::NonRecursive)?;	×
27		×
28	let additional_paths = vec![dir.join("sql")];	×
29		×
30	for path in additional_paths {	×
31	let _ = watcher.watch(path.as_path(), RecursiveMode::NonRecursive);	×
32	}	×
33		×
34	debouncer	×
35	.cache()	×
36	.add_root(dir.as_path(), RecursiveMode::Recursive);	×
37		×
38	let running = shutdown.get_running_flag().clone();	×
39	loop {	×
40	let event = rx.recv_timeout(Duration::from_millis(100));	×
41	match event {	×
42	Ok(result) => match result {	×
43	Ok(_events) => {	×
44	build(runtime.clone(), state.clone()).await;	×
45	}	×
46	Err(errors) => errors.iter().for_each(\|error\| info!("{error:?}")),	×
47	},	×
48	Err(e) => {	×
49	if !running.load(std::sync::atomic::Ordering::Relaxed) {	×
50	break;	×
51	}	×
52	if e == std::sync::mpsc::RecvTimeoutError::Disconnected {	×
53	break;	×
54	}	×
55	}	×
56	}	×
57	}
58		×
59	Ok(())	×
60	}	×
61		×
62	async fn build(runtime: Arc<Runtime>, state: Arc<LiveState>) {	×
63	state.broadcast(BroadcastType::Start).await;	×
64	if let Err(res) = state.build(runtime).await {	×
65	let message = res.to_string();	×
66	state.broadcast(BroadcastType::Failed(message)).await;	×
67	} else {
68	state.broadcast(BroadcastType::Success).await;	×
69	}
70	}	×

getdozer / dozer / 5957916370

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous