radix-clis 1.4.0

A collection of CLIs for developing, building and testing Scrypto code, from the Radix DLT project.
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

use super::Error;
use flate2::read::GzDecoder;
use flume;
use flume::Sender;
use radix_common::prelude::*;
use radix_transactions::model::RawLedgerTransaction;
use rocksdb::{Direction, IteratorMode, Options, DB};
use std::fs::File;
use std::io::Read;
use std::path::PathBuf;
use std::thread;
use std::time::Duration;
use tar::Archive;

pub enum TxnReader {
    TransactionFile(Box<Archive<GzDecoder<File>>>),
    StateManagerDatabaseDir(PathBuf),
}

impl TxnReader {
    pub fn read(
        &mut self,
        from_version: u64,
        to_version: Option<u64>,
        tx: Sender<RawLedgerTransaction>,
    ) -> Result<(), Error> {
        match self {
            TxnReader::TransactionFile(archive) => {
                for entry in archive.entries().map_err(Error::IOError)? {
                    // read the entry
                    let mut entry = entry.map_err(Error::IOError)?;
                    let tx_version = entry
                        .header()
                        .path()
                        .ok()
                        .and_then(|path| path.to_str().map(ToOwned::to_owned))
                        .and_then(|s| u64::from_str(&s).ok())
                        .ok_or(Error::InvalidTransactionArchive)?;
                    let mut tx_payload = Vec::new();
                    entry.read_to_end(&mut tx_payload).map_err(Error::IOError)?;

                    if tx_version <= from_version {
                        continue;
                    }
                    if let Some(to_version) = to_version {
                        if tx_version > to_version {
                            break;
                        }
                    }

                    tx.send(RawLedgerTransaction::from_vec(tx_payload)).unwrap();
                }
            }
            TxnReader::StateManagerDatabaseDir(db_dir) => {
                let temp_dir = tempfile::tempdir().map_err(Error::IOError)?;

                let db = DB::open_cf_as_secondary(
                    &Options::default(),
                    db_dir.as_path(),
                    temp_dir.as_ref(),
                    vec![
                        "raw_ledger_transactions",
                        "committed_transaction_identifiers",
                    ],
                )
                .unwrap();

                let iter_start_state_version = from_version + 1;

                loop {
                    db.try_catch_up_with_primary()
                        .expect("DB catch up with primary failed");
                    let txn_iter = db.iterator_cf(
                        &db.cf_handle("raw_ledger_transactions").unwrap(),
                        IteratorMode::From(
                            &iter_start_state_version.to_be_bytes(),
                            Direction::Forward,
                        ),
                    );
                    for next_txn in txn_iter {
                        let next_txn = next_txn.unwrap();
                        tx.send(RawLedgerTransaction::from_vec(next_txn.1.to_vec()))
                            .unwrap();
                    }
                    thread::sleep(Duration::from_secs(1));
                }
            }
        }

        Ok(())
    }
}