fartdown 0.1.0

An experimental semantic markup language and static site generator for crafting hypertext documents
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

use std::path::PathBuf;

use futures::stream::FuturesUnordered;
use futures::{future::BoxFuture, FutureExt};
use log::error;
use tokio::task::yield_now;
use tokio_stream::StreamExt;

use super::command::Command;
use super::graph_sorter::grouped_topological_sort;
use super::op::{OpId, Operation};
use super::state_context::State;
use super::task;
use super::utils::parents;
use super::{
    command::Config,
    types::{AppError, Event, EventRx, EventTx},
};

pub async fn handle(
    _event_tx: EventTx,
    event_rx: EventRx,
    config: &Config,
) -> Result<(), AppError> {
    let state = State::new();

    handle_state(event_rx, config, state).await
}

pub async fn handle_state(
    mut event_rx: EventRx,
    config: &Config,
    state: State,
) -> Result<(), AppError> {
    let mut tasks = FuturesUnordered::<BoxFuture<Result<bool, _>>>::new();
    let handle_error = |e| process_error(e, config);
    let done = |tasks: &FuturesUnordered<_>| tasks.is_empty() && !config.interactive;

    loop {
        // Allow other tasks to run
        yield_now().await;

        if let Ok(e) = event_rx.try_recv() {
            process_event(e, config, &mut tasks, &state).or_else(handle_error)?;
        } else if let Some(e) = tasks.next().await {
            process_result(e, config, &mut tasks, &state).or_else(handle_error)?;
        } else if tasks.is_empty() {
            process_graph(&mut tasks, &state);

            if done(&tasks) {
                break Ok(());
            }
        }
    }
}

fn process_event(
    event: Event,
    _config: &Config,
    tasks: &mut FuturesUnordered<BoxFuture<'static, Result<bool, AppError>>>,
    state: &State,
) -> Result<(), AppError> {
    match event {
        Event::Command(Ok(cmd)) => match cmd {
            Command::Load { ref paths } => {
                let paths = paths.iter().map(PathBuf::from).collect::<Vec<_>>();
                let paths_parents = parents(paths.clone().into_iter())?
                    .into_iter()
                    .collect::<Vec<_>>();
                let splits = None;

                tasks.push(task::index(paths_parents, state.locations.clone()).boxed());
                state
                    .insert_op_chain([Operation::Gather { cmd, paths, splits }, Operation::Finish]);
            }
            Command::Build { ref paths, ref splits } => {
                let paths = paths.iter().map(PathBuf::from).collect::<Vec<_>>();
                let paths_parents = parents(paths.clone().into_iter())?
                    .into_iter()
                    .collect::<Vec<_>>();
                let splits = Some(splits.clone());

                tasks.push(task::index(paths_parents, state.locations.clone()).boxed());
                state
                    .insert_op_chain([Operation::Gather { cmd, paths, splits }, Operation::Finish]);
            }
        },
        Event::Command(Err(_)) => todo!(),
        Event::CommandOk => todo!(),
        Event::TaskOk => {}
        Event::TaskError(e) => return Err(e),
    }
    Ok(())
}

fn process_result(
    res: Result<bool, AppError>,
    config: &Config,
    tasks: &mut FuturesUnordered<BoxFuture<'static, Result<bool, AppError>>>,
    state: &State,
) -> Result<(), AppError> {
    match res {
        Ok(true) => process_event(Event::CommandOk, config, tasks, state),
        Ok(false) => process_event(Event::TaskOk, config, tasks, state),
        Err(e) => process_event(Event::TaskError(e), config, tasks, state),
    }
}

fn process_error(error: AppError, config: &Config) -> Result<(), AppError> {
    error!("{}", error);
    match config.interactive {
        true => Ok(()),
        false => Err(error),
    }
}

fn process_graph(
    tasks: &mut FuturesUnordered<BoxFuture<'static, Result<bool, AppError>>>,
    state: &State,
) {
    let operations = state.operations.lock().expect("poisoned lock");
    let sorted = grouped_topological_sort(&*operations).unwrap();

    let mut next_tasks = sorted
        .into_iter()
        .find(|group| group.iter().any(|id| !state.is_op_processed(id)))
        .unwrap_or_default();
    next_tasks.retain(|id| !state.is_op_processed(id));

    for id in next_tasks {
        let vertex = operations.get(&id).unwrap();
        let op = vertex.clone();
        let dep = operations.get_first_node_dependency(&op).map(OpId::uri);
        let arts = state.artifacts.clone();
        let ops = state.operations.clone();

        match vertex {
            Operation::Gather { .. } => tasks.push(task::gather(op, ops).boxed()),
            Operation::Load { .. } => tasks.push(task::load(op, arts).boxed()),
            Operation::Parse { .. } => tasks.push(task::parse(op, dep.unwrap(), arts).boxed()),
            Operation::Finish { .. } => {}
        }
        state.mark_op_processed(id.clone());
    }
}