raftify 0.1.82

Experimental High level Raft framework
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

use chrono::Utc;
use serde_json::{json, Value};
use std::{
    fs::{File, OpenOptions},
    io::{self, Read, Seek, Write as StdIoWrite},
    path::Path,
};

use crate::raft::{eraftpb::Entry, formatter::Bytes, formatter::CUSTOM_FORMATTER};

fn entry_type_to_str(entry_type: i32) -> &'static str {
    match entry_type {
        0 => "EntryNormal",
        1 => "EntryConfChange",
        2 => "EntryConfChangeV2",
        _ => unreachable!(),
    }
}

pub fn append_compacted_logs(dest_path: &Path, new_data: &[Entry]) -> io::Result<()> {
    if new_data.is_empty() {
        return Ok(());
    }

    if !dest_path.exists() {
        File::create(dest_path)?;
    }

    let mut file = OpenOptions::new().read(true).write(true).open(dest_path)?;
    let mut contents = String::new();
    file.read_to_string(&mut contents)?;

    let mut existing_data: Vec<Value> = if contents.is_empty() {
        Vec::new()
    } else {
        serde_json::from_str(&contents)?
    };

    let formatter = CUSTOM_FORMATTER.read().unwrap();

    let mut new_data_json = Vec::new();
    for entry in new_data {
        new_data_json.push(json!({
            "entry_type": entry_type_to_str(entry.entry_type),
            "term": entry.term,
            "index": entry.index,
            "data": formatter.format_entry_data(&Bytes::Prost(entry.data.clone())),
            "context": formatter.format_entry_context(&Bytes::Prost(entry.context.clone())),
            "sync_log": entry.sync_log,
        }));
    }

    let timestamp = Utc::now().to_rfc3339();

    existing_data.push(json!({ timestamp: new_data_json }));

    file.set_len(0)?;
    file.seek(io::SeekFrom::Start(0))?;
    write!(file, "{}", serde_json::to_string_pretty(&existing_data)?)?;

    Ok(())
}