write_to_graph/
write_to_graph.rs

1use std::env;
2
3use lazy_static::lazy_static;
4use regex::Regex;
5use terminus_store::*;
6use tokio;
7use tokio::io::{self, AsyncBufReadExt};
8
9enum Command {
10    Add(ValueTriple),
11    Remove(ValueTriple),
12}
13
14async fn parse_command(s: &str) -> io::Result<Command> {
15    lazy_static! {
16        static ref RE: Regex =
17            Regex::new(r"(\S*)\s*(\S*)\s*,\s*(\S*)\s*,\s*(\S*)\s*(\S*)\s*").unwrap();
18    }
19
20    if let Some(matches) = RE.captures(s) {
21        let command_name = &matches[1];
22        let subject = &matches[2];
23        let predicate = &matches[3];
24        let object_type_name = &matches[4];
25        let object = &matches[5];
26
27        let triple = match object_type_name {
28            "node" => ValueTriple::new_node(subject, predicate, object),
29            "value" => ValueTriple::new_string_value(subject, predicate, object),
30            _ => {
31                return Err(io::Error::new(
32                    io::ErrorKind::InvalidData,
33                    format!("invalid object type {}", object_type_name),
34                ))
35            }
36        };
37
38        match command_name {
39            "add" => Ok(Command::Add(triple)),
40            "remove" => Ok(Command::Remove(triple)),
41            _ => Err(io::Error::new(
42                io::ErrorKind::InvalidData,
43                format!("invalid command {}", command_name),
44            )),
45        }
46    } else {
47        Err(io::Error::new(
48            io::ErrorKind::InvalidData,
49            format!("could not match line {}", s),
50        ))
51    }
52}
53
54async fn process_commands(store_path: &str, graph: &str) -> io::Result<()> {
55    let store = open_directory_store(store_path);
56    let graph = store
57        .open(graph)
58        .await?
59        .expect(&format!("expected graph {} to exist", graph));
60
61    // There are two types of builders. One creates a new base layer,
62    // which has no parent. The other creates a child layer, which has
63    // another layer as its parent.
64    let builder = match graph.head().await? {
65        Some(layer) => layer.open_write().await?,
66        None => store.create_base_layer().await?,
67    };
68    let mut stdin = io::BufReader::new(io::stdin()).lines();
69
70    while let Some(line) = stdin.next_line().await? {
71        let segment = line.trim();
72        if segment.len() == 0 {
73            continue;
74        }
75
76        let command = parse_command(segment).await?;
77
78        // add all the input data into the builder.
79        // The builder keeps an in-memory list of added and removed
80        // triples. If the same triple is added and removed on the
81        // same builder, it is a no-op. This is even the case when it
82        // is then later re-added on the same builder.
83        //
84        // Since no io is happening, adding triples to the builder is
85        // not a future.
86        match command {
87            Command::Add(triple) => builder.add_value_triple(triple)?,
88            Command::Remove(triple) => builder.remove_value_triple(triple)?,
89        }
90    }
91
92    // When commit is called, the builder writes its data to
93    // persistent storage.
94    let layer = builder.commit().await?;
95
96    // While a layer exists now, it's not yet attached to anything,
97    // and is therefore unusable unless you know the exact identifier
98    // of the layer itself. To make this the graph data, we have to
99    // set the grap head to this layer.
100    graph.set_head(&layer).await?;
101
102    println!(
103        "Added: {}, removed: {}",
104        layer.triple_layer_addition_count().await?,
105        layer.triple_layer_removal_count().await?
106    );
107
108    Ok(())
109}
110
111#[tokio::main]
112async fn main() {
113    let args: Vec<String> = env::args().collect();
114    if args.len() != 3 {
115        println!(
116            "usage: {} <path> <graph_name>
117Commands should come from standard input, and should be of the following format:
118  add subject, predicate, node object
119  add subject, predicate, value object
120  remove subject, predicate, node object
121  remove subject, predicate, value object",
122            args[0]
123        );
124    } else {
125        process_commands(&args[1], &args[2]).await.unwrap();
126    }
127}
write_to_graph/write_to_graph.rs

write_to_graph/
write_to_graph.rs
