Mirror/brk
1
0
Fork 0
mirror of https://github.com/bitcoinresearchkit/brk.git synced 2026-06-09 22:43:33 -07:00
Code Issues Packages Projects Releases Wiki Activity

server: rework api side

Browse Source
This commit is contained in:
nym21
2024-12-22 00:42:11 +01:00
parent 8fabbde13b
commit 9dd87a48a6
40 changed files with 633 additions and 746 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+3
View File
@@ -74,6 +74,9 @@ Please open an issue if you want to add another instance
- Git
- Unix based operating system (Mac OS or Linux)
- Ubuntu users need to install `open-ssl` via `sudo apt install libssl-dev pkg-config`
- Mac OS:
- Disable Spotlight or exclude the `--kibodir` folder from it
- Don't use Time Machine or exclude the `--kibodir` folder (especially needed for local snapshots)
### Build
src/main.rs
+6 -9
View File
@@ -8,7 +8,8 @@ mod server;
mod structs;
mod utils;
use parser::{Databases, Datasets};
use parser::Datasets;
use server::api::structs::Routes;
use structs::{Config, Exit};
use utils::init_log;
@@ -31,31 +32,27 @@ fn main() -> color_eyre::Result<()> {
let rpc = Client::from(&config);
let databases = Databases::import(&config);
let datasets = Datasets::import(&config)?;
let paths_to_type = datasets.get_paths_to_type(&config);
let routes = Routes::build(&Datasets::import(&config)?, &config);
tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.unwrap()
.block_on(async {
let config_clone = config.clone();
let run_parser = config.parser();
let run_server = config.server();
let config_clone = config.clone();
let handle = tokio::spawn(async move {
if run_server {
server::main(paths_to_type, &config_clone).await.unwrap();
server::main(routes, config_clone).await.unwrap();
} else {
info!("Skipping server");
}
});
if run_parser {
parser::main(&config, &rpc, &exit, databases, datasets)?;
parser::main(&config, &rpc, &exit)?;
} else {
info!("Skipping parser");
}
src/parser/actions/export.rs
+5 -5
View File
@@ -37,14 +37,14 @@ pub fn export(
exit.block();
let text = if defragment {
"export and defragmentation"
"Exporting and defragmenting..."
} else {
"export"
"Exporting..."
};
info!("Starting {text}");
info!("{text}");
time(&format!("Finished {text}"), || -> color_eyre::Result<()> {
datasets.export(config)?;
time("Finished export", || -> color_eyre::Result<()> {
datasets.export(config, height)?;
if let Some(databases) = databases {
databases.export(height, date, defragment)?;
src/parser/actions/iter_blocks.rs
+3 -3
View File
@@ -173,7 +173,7 @@ pub fn iter_blocks(
.as_ref()
.map_or(true, |date| date.is_first_of_month());
if (is_check_point && instant.elapsed().as_secs() >= 2)
if (is_check_point && instant.elapsed().as_secs() >= 1)
|| height.is_close_to_end(approx_block_count)
{
break 'days;
@@ -192,7 +192,7 @@ pub fn iter_blocks(
if first_unsafe_heights.computed <= last_height {
info!("Computing datasets...");
time("Computing datasets", || {
time("Computed datasets", || {
let dates = processed_dates.into_iter().collect_vec();
let heights = processed_heights.into_iter().collect_vec();
@@ -226,7 +226,7 @@ pub fn iter_blocks(
})?;
if config.record_ram_usage() {
time("Exporing allocation files", || {
time("Exporting allocation files", || {
generate_allocation_files(datasets, databases, &states, last_height)
})?;
}
src/parser/actions/min_height.rs
-7
View File
@@ -118,13 +118,6 @@ pub fn find_first_inserted_unsafe_height(
|| min_initial_inserted_last_address_date.is_none()
|| min_initial_inserted_last_address_height.is_none();
// if include_addresses {
// dbg!(include_addresses);
// panic!("");
// }
if true {panic!()}
states.reset(config, include_addresses);
databases.reset(include_addresses);
src/parser/databases/mod.rs
+4 -5
View File
@@ -131,8 +131,8 @@ impl Databases {
pub fn check_if_usable(
&self,
min_initial_last_address_height: Option<Height>,
min_initial_last_address_date: Option<Date>,
last_address_height: Option<Height>,
last_address_date: Option<Date>,
) -> bool {
let are_tx_databases_in_sync = self
.txout_index_to_amount
@@ -169,8 +169,7 @@ impl Databases {
return false;
}
// let are_address_datasets_farer_or_in_sync_with_address_databases =
min_initial_last_address_height >= self.address_to_address_index.metadata.last_height
&& min_initial_last_address_date >= self.address_to_address_index.metadata.last_date
last_address_height >= self.address_to_address_index.metadata.last_height
&& last_address_date >= self.address_to_address_index.metadata.last_date
}
}
src/parser/databases/txid_to_tx_data.rs
+1 -1
View File
@@ -84,7 +84,7 @@ impl TxidToTxData {
}
#[inline(always)]
pub fn _open_db(&mut self, db_index: u16) -> &mut Database {
fn _open_db(&mut self, db_index: u16) -> &mut Database {
let path = self.path.to_owned();
self.map.entry(db_index).or_insert_with(|| {
let path = path.join(db_index.to_string());
src/parser/datasets/_traits/min_initial_state.rs
+4
View File
@@ -29,6 +29,10 @@ impl MinInitialStates {
computed: MinInitialState::compute_from_datasets(datasets, Mode::Computed, config),
}
}
pub fn min_last_height(&self) -> Option<Height> {
self.computed.last_height.min(self.inserted.last_height)
}
}
#[derive(Default, Debug, Clone, Copy, Allocative)]
src/parser/datasets/mod.rs
+25 -57
View File
@@ -1,4 +1,4 @@
use std::{collections::BTreeMap, ops::RangeInclusive, path::PathBuf};
use std::{collections::BTreeMap, ops::RangeInclusive};
use allocative::Allocative;
@@ -139,14 +139,18 @@ impl Datasets {
utxo,
};
s.min_initial_states
.consume(MinInitialStates::compute_from_datasets(&s, config));
s.set_initial_states(config);
info!("Imported datasets");
Ok(s)
}
fn set_initial_states(&mut self, config: &Config) {
self.min_initial_states
.consume(MinInitialStates::compute_from_datasets(self, config));
}
pub fn insert(&mut self, insert_data: InsertData) {
if insert_data.compute_addresses {
self.address.insert(&insert_data);
@@ -228,21 +232,6 @@ impl Datasets {
&mut self.price.market_cap,
);
// No compute needed for now
// if self.block_metadata.should_compute(height, date) {
// self.block_metadata.compute(&compute_data);
// }
// No compute needed for now
// if self.date_metadata.should_compute(height, date) {
// self.date_metadata.compute(&compute_data);
// }
// No compute needed for now
// if self.coindays.should_compute(height, date) {
// self.coindays.compute(&compute_data);
// }
if self.transaction.should_compute(&compute_data) {
self.transaction.compute(
&compute_data,
@@ -268,23 +257,12 @@ impl Datasets {
}
}
pub fn get_paths_to_type(&self, config: &Config) -> BTreeMap<PathBuf, String> {
let mut path_to_type: BTreeMap<PathBuf, String> = self
.to_any_dataset_vec()
.into_iter()
.flat_map(|dataset| dataset.to_all_map_vec())
.flat_map(|map| map.get_paths_to_type())
.collect();
pub fn export(&mut self, config: &Config, height: Height) -> color_eyre::Result<()> {
let is_new = self
.min_initial_states
.min_last_height()
.map_or(true, |last| last <= height);
path_to_type.insert(
config.path_datasets_last_values().unwrap().to_owned(),
"Value".to_string(),
);
path_to_type
}
pub fn export(&mut self, config: &Config) -> color_eyre::Result<()> {
self.to_mut_any_dataset_vec()
.into_iter()
.for_each(|dataset| dataset.pre_export());
@@ -295,37 +273,27 @@ impl Datasets {
let mut path_to_last: BTreeMap<String, Value> = BTreeMap::default();
let path_dataset = config.path_datasets();
let path_dataset_ser = path_dataset.to_str().unwrap();
let path_price = config.path_price();
let path_price_ser = path_price.to_str().unwrap();
self.to_mut_any_dataset_vec()
.into_iter()
.for_each(|dataset| {
dataset.post_export();
dataset.to_all_map_vec().iter().for_each(|map| {
if let Some(last_path) = map.path_last() {
if let Some(last_value) = map.last_value() {
let mut last_path = last_path.clone();
last_path.pop();
let key = last_path
.to_str()
.unwrap()
.replace(path_dataset_ser, "")
.replace(path_price_ser, "")
.split("/")
.join("-")
.to_string();
path_to_last.insert(key, last_value);
if is_new {
dataset.to_all_map_vec().iter().for_each(|map| {
if map.path_last().is_some() {
if let Some(last_value) = map.last_value() {
path_to_last.insert(map.id(config), last_value);
}
}
}
});
});
}
});
Json::export(&config.path_datasets_last_values(), &path_to_last)?;
if is_new {
Json::export(&config.path_datasets_last_values(), &path_to_last)?;
}
self.set_initial_states(config);
Ok(())
}
src/parser/datasets/subs/recap.rs
+2 -2
View File
@@ -5,12 +5,12 @@ use allocative::Allocative;
use crate::{
structs::{
Date, DateMapChunkId, GenericMap, MapChunkId, MapKey, MapKind, MapPath, MapSerialized,
MapValue, SerializedBTreeMap,
MapValue, SerializedDateMap,
},
utils::{get_percentile, LossyFrom},
};
pub type DateRecapDataset<T> = RecapDataset<Date, T, DateMapChunkId, SerializedBTreeMap<Date, T>>;
pub type DateRecapDataset<T> = RecapDataset<Date, T, DateMapChunkId, SerializedDateMap<T>>;
#[derive(Allocative)]
pub struct RecapDataset<Key, Value, ChunkId, Serialized> {
src/parser/mod.rs
+4 -7
View File
@@ -16,18 +16,15 @@ pub use states::*;
use crate::structs::{Config, Exit};
pub fn main(
config: &Config,
rpc: &Client,
exit: &Exit,
mut databases: Databases,
mut datasets: Datasets,
) -> color_eyre::Result<()> {
pub fn main(config: &Config, rpc: &Client, exit: &Exit) -> color_eyre::Result<()> {
loop {
let block_count = rpc.get_blockchain_info().unwrap().blocks as usize;
info!("{block_count} blocks found.");
let mut databases = Databases::import(config);
let mut datasets = Datasets::import(config)?;
iter_blocks(
config,
rpc,
src/server/api/handlers/dataset.rs
+210 -160
View File
@@ -1,224 +1,274 @@
use std::{collections::BTreeMap, path::PathBuf};
use std::{fmt::Debug, path::PathBuf, time::Instant};
use axum::{
extract::{Path, Query, State},
http::HeaderMap,
response::{IntoResponse, Response},
};
use color_eyre::{eyre::eyre, owo_colors::OwoColorize};
use bincode::Decode;
use chrono::{DateTime, Utc};
use color_eyre::eyre::{eyre, ContextCompat};
use reqwest::StatusCode;
use serde::Deserialize;
use serde::{de::DeserializeOwned, Deserialize, Serialize};
use crate::{
io::{Json, COMPRESSED_BIN_EXTENSION, JSON_EXTENSION},
server::{
api::{
structs::{Chunk, Kind, Route},
structs::{ChunkMetadata, DatasetRange, DatasetRangeChunk, Extension, Kind, Route},
API_URL_PREFIX,
},
header_map::HeaderMapUtils,
log_result, AppState,
},
structs::{Date, DateMap, Height, HeightMap, MapChunkId, HEIGHT_MAP_CHUNK_SIZE, OHLC},
};
use super::{
extension::Extension,
response::{typed_value_to_response, value_to_response},
structs::{
Date, GenericMap, Height, MapChunkId, MapKey, MapSerialized, MapValue, SerializedDateMap,
SerializedVec, OHLC,
},
};
#[derive(Deserialize)]
pub struct Params {
chunk: Option<usize>,
all: Option<bool>,
pub struct DatasetParams {
pub chunk: Option<usize>,
pub all: Option<bool>,
pub kind: String,
}
pub async fn dataset_handler(
headers: HeaderMap,
path: Path<String>,
query: Query<Params>,
query: Query<DatasetParams>,
State(app_state): State<AppState>,
) -> Response {
match _dataset_handler(headers, &path, &query, app_state) {
Ok(response) => {
log_result(
response.status(),
&format!(
"{API_URL_PREFIX}/{}{}{}",
path.0,
query.chunk.map_or("".to_string(), |chunk| format!(
"{}{chunk}",
"?chunk=".bright_black()
)),
query.all.map_or("".to_string(), |all| format!(
"{}{all}",
"?all=".bright_black()
))
),
);
let instant = Instant::now();
let ser_path = format!(
"{API_URL_PREFIX}/{}?kind={}{}{}",
path.0,
query.kind,
query
.chunk
.map_or("".to_string(), |chunk| format!("&chunk={chunk}")),
query
.all
.map_or("".to_string(), |all| format!("&all={all}"))
);
match result_handler(headers, &path, &query, app_state) {
Ok(response) => {
log_result(response.status(), &ser_path, instant);
response
}
Err(error) => {
let code = StatusCode::INTERNAL_SERVER_ERROR;
log_result(code, &format!("{API_URL_PREFIX}/{}", path.0));
let mut response = (code, error.to_string()).into_response();
let mut response =
(StatusCode::INTERNAL_SERVER_ERROR, error.to_string()).into_response();
log_result(response.status(), &ser_path, instant);
response.headers_mut().insert_cors();
response
}
}
}
const DATE_PREFIX: &str = "date-to-";
const HEIGHT_PREFIX: &str = "height-to-";
fn _dataset_handler(
fn result_handler(
headers: HeaderMap,
Path(path): &Path<String>,
query: &Query<Params>,
AppState { routes }: AppState,
query: &Query<DatasetParams>,
AppState { routes, .. }: AppState,
) -> color_eyre::Result<Response> {
if query.chunk.is_some() && query.all.is_some() {
return Err(eyre!("chunk and all are exclusive"));
}
let (kind, id, route) = if path.starts_with(DATE_PREFIX) {
let id = convert_path_to_id(path.strip_prefix(DATE_PREFIX).unwrap());
let route = routes.date.get(&id);
(Kind::Date, id, route)
} else if path.starts_with(HEIGHT_PREFIX) {
let id = convert_path_to_id(path.strip_prefix(HEIGHT_PREFIX).unwrap());
let route = routes.height.get(&id);
(Kind::Height, id, route)
} else {
let id = convert_path_to_id(path);
let route = routes.last.get(&id);
(Kind::Last, id, route)
};
let path_buf = PathBuf::from(&path);
let id = path_buf.file_stem().unwrap().to_str().unwrap();
let ext = Extension::from(&path_buf);
let route = routes.get(id);
if route.is_none() {
return Err(eyre!("Path error"));
return Err(eyre!("Wrong path"));
}
let route = route.unwrap();
let mut route = route.unwrap().to_owned();
let mut chunk = None;
if query.all.map_or(true, |b| !b) {
match kind {
Kind::Date => {
let datasets = DateMap::<usize>::_read_dir(&route.file_path, &route.serialization);
process_datasets(&headers, kind, &mut chunk, &mut route, query, datasets)?;
}
Kind::Height => {
let datasets =
HeightMap::<usize>::_read_dir(&route.file_path, &route.serialization);
process_datasets(&headers, kind, &mut chunk, &mut route, query, datasets)?;
}
Kind::Last => {
if !route.values_type.ends_with("Value") {
route.file_path.set_extension(COMPRESSED_BIN_EXTENSION);
} else {
route.file_path.set_extension(JSON_EXTENSION);
}
}
}
}
let (date, response) = headers.check_if_modified_since(&route.file_path)?;
if let Some(response) = response {
return Ok(response);
}
let type_name = route.values_type.split("::").last().unwrap();
let extension = Extension::from(&std::path::PathBuf::from(&path));
let mut response = match type_name {
"u8" => typed_value_to_response::<u8>(kind, &route, chunk, id, extension)?,
"u16" => typed_value_to_response::<u16>(kind, &route, chunk, id, extension)?,
"u32" => typed_value_to_response::<u32>(kind, &route, chunk, id, extension)?,
"u64" => typed_value_to_response::<u64>(kind, &route, chunk, id, extension)?,
"usize" => typed_value_to_response::<usize>(kind, &route, chunk, id, extension)?,
"f32" => typed_value_to_response::<f32>(kind, &route, chunk, id, extension)?,
"f64" => typed_value_to_response::<f64>(kind, &route, chunk, id, extension)?,
"OHLC" => typed_value_to_response::<OHLC>(kind, &route, chunk, id, extension)?,
"Date" => typed_value_to_response::<Date>(kind, &route, chunk, id, extension)?,
"Height" => typed_value_to_response::<Height>(kind, &route, chunk, id, extension)?,
"Value" => {
value_to_response::<serde_json::Value>(Json::import(&route.file_path)?, extension)
}
let type_name = route.type_name.as_str();
Ok(match type_name {
"u8" => typed_handler::<u8>(headers, id, ext, query, route)?,
"u16" => typed_handler::<u16>(headers, id, ext, query, route)?,
"u32" => typed_handler::<u32>(headers, id, ext, query, route)?,
"u64" => typed_handler::<u64>(headers, id, ext, query, route)?,
"usize" => typed_handler::<usize>(headers, id, ext, query, route)?,
"f32" => typed_handler::<f32>(headers, id, ext, query, route)?,
"f64" => typed_handler::<f64>(headers, id, ext, query, route)?,
"OHLC" => typed_handler::<OHLC>(headers, id, ext, query, route)?,
"Date" => typed_handler::<Date>(headers, id, ext, query, route)?,
"Height" => typed_handler::<Height>(headers, id, ext, query, route)?,
// "Value" => {
// value_to_response::<serde_json::Value>(Json::import(&route.file_path)?, extension)
// }
_ => panic!("Incompatible type: {type_name}"),
};
})
}
fn typed_handler<T>(
headers: HeaderMap,
id: &str,
ext: Option<Extension>,
query: &Query<DatasetParams>,
route: &Route,
) -> color_eyre::Result<Response>
where
T: Serialize + Debug + DeserializeOwned + Decode + MapValue,
{
let kind = Kind::try_from(&query.kind)?;
if !route.list.contains(&kind) {
return Err(eyre!("{kind:?} not supported for this dataset"));
}
let range = DatasetRange::try_from(query)?;
let (mut response, date_modified) = match kind {
Kind::Date => map_to_response::<Date, T, _, SerializedDateMap<T>>(
id, headers, route, &ext, range, query,
),
Kind::Height => map_to_response::<Height, T, _, SerializedVec<T>>(
id, headers, route, &ext, range, query,
),
Kind::Last => {
let last_value: T = route.serialization.import(&route.path.join("last"))?;
return Ok(axum::response::Json(last_value).into_response());
}
}?;
let status_ok = response.status() == StatusCode::OK;
let headers = response.headers_mut();
headers.insert_last_modified(date);
headers.insert_cors();
if status_ok {
headers.insert_last_modified(date_modified);
}
match ext {
Some(extension) => {
headers.insert_content_disposition_attachment();
match extension {
Extension::CSV => headers.insert_content_type_text_csv(),
Extension::JSON => headers.insert_content_type_application_json(),
}
}
_ => headers.insert_content_type_application_json(),
}
Ok(response)
}
fn convert_path_to_id(s: &str) -> String {
Extension::remove_extension(s).replace('-', "_")
}
fn process_datasets<ChunkId>(
headers: &HeaderMap,
kind: Kind,
chunk: &mut Option<Chunk>,
route: &mut Route,
query: &Query<Params>,
datasets: BTreeMap<ChunkId, PathBuf>,
) -> color_eyre::Result<()>
fn map_to_response<Key, Value, ChunkId, Serialized>(
id: &str,
headers: HeaderMap,
route: &Route,
ext: &Option<Extension>,
range: DatasetRange,
query: &Query<DatasetParams>,
) -> color_eyre::Result<(Response, DateTime<Utc>)>
where
Key: MapKey<ChunkId>,
Value: MapValue,
ChunkId: MapChunkId,
Serialized: MapSerialized<Key, Value, ChunkId>,
{
let (last_chunk_id, _) = datasets.last_key_value().unwrap_or_else(|| {
dbg!(&datasets, &route);
panic!()
});
let folder_path = route.path.join(Key::map_name());
let serialization = &route.serialization;
let chunk_id = query
.chunk
.map(|id| ChunkId::from_usize(id))
.unwrap_or(*last_chunk_id);
let date_modified;
let path = datasets.get(&chunk_id);
let datasets =
GenericMap::<Key, Value, ChunkId, Serialized>::_read_dir(&folder_path, serialization);
if path.is_none() {
return Err(eyre!("Couldn't find chunk"));
}
let mut chunk_meta = None;
let path = path.unwrap();
route.file_path = path.clone();
let dataset = if let DatasetRange::Chunk(range_chunk) = range {
let chunk_id = match range_chunk {
DatasetRangeChunk::Last => {
*datasets
.last_key_value()
.context("Last tuple of dataset directory")?
.0
}
DatasetRangeChunk::Chunk(chunk) => ChunkId::from_usize(chunk),
};
let offset = match kind {
Kind::Date => 1,
Kind::Height => HEIGHT_MAP_CHUNK_SIZE as usize,
_ => panic!(),
let chunk_path = datasets.get(&chunk_id);
if chunk_path.is_none() {
return Err(eyre!("Couldn't find chunk"));
}
let chunk_path = chunk_path.unwrap();
let (date, response) = headers.check_if_modified_since(chunk_path)?;
date_modified = date;
if let Some(response) = response {
return Ok((response, date_modified));
}
let to_url = |chunk: Option<ChunkId>| {
chunk.and_then(|chunk| {
datasets.contains_key(&chunk).then(|| {
let scheme = headers.get_scheme();
let host = headers.get_host();
format!(
"{scheme}://{host}/api/{id}?kind={}&chunk={}",
query.kind,
chunk.to_usize()
)
})
})
};
chunk_meta.replace(ChunkMetadata {
id: chunk_id.to_usize(),
next: to_url(chunk_id.next()),
previous: to_url(chunk_id.previous()),
});
serialization.import::<Serialized>(chunk_path)?
} else {
let newest_file = datasets
.values()
.max_by(|a, b| {
a.metadata()
.unwrap()
.modified()
.unwrap()
.cmp(&b.metadata().unwrap().modified().unwrap())
})
.context("Expect to find newest file")?;
let (date, response) = headers.check_if_modified_since(newest_file)?;
date_modified = date;
if let Some(response) = response {
return Ok((response, date_modified));
}
Serialized::import_all(&folder_path, serialization)
};
let offsetted_to_url = |offseted| {
datasets.get(&ChunkId::from_usize(offseted)).map(|_| {
let scheme = headers.get_scheme();
let host = headers.get_host();
format!("{scheme}://{host}/api/{}?chunk={offseted}", route.url_path)
let response = if *ext == Some(Extension::CSV) {
dataset.to_csv(id).into_response()
} else if let Some(chunk) = chunk_meta {
axum::response::Json(SerializedMapChunk {
chunk,
map: dataset.map(),
version: dataset.version(),
})
.into_response()
} else {
axum::response::Json(dataset).into_response()
};
let chunk_id = chunk_id.to_usize();
chunk.replace(Chunk {
id: chunk_id,
next: chunk_id.checked_add(offset).and_then(offsetted_to_url),
previous: chunk_id.checked_sub(offset).and_then(offsetted_to_url),
});
Ok(())
Ok((response, date_modified))
}
#[derive(Serialize)]
struct SerializedMapChunk<T>
where
T: Serialize,
{
version: u32,
chunk: ChunkMetadata,
map: T,
}
src/server/api/handlers/fallback.rs
-19
View File
@@ -1,19 +0,0 @@
use axum::{extract::State, http::HeaderMap, response::Response};
use reqwest::header::HOST;
use crate::server::AppState;
use super::response::{generic_to_reponse, update_reponse_headers};
pub async fn fallback(headers: HeaderMap, State(app_state): State<AppState>) -> Response {
update_reponse_headers(
generic_to_reponse(
app_state
.routes
.to_full_paths(headers[HOST].to_str().unwrap().to_string()),
None,
),
60,
None,
)
}
src/server/api/handlers/last_values.rs
+13
View File
@@ -0,0 +1,13 @@
use axum::{
extract::State,
response::{IntoResponse, Response},
};
use serde_json::Value;
use crate::{io::Json, server::AppState};
pub async fn last_values_handler(State(app_state): State<AppState>) -> Response {
let values = Json::import::<Value>(&app_state.config.path_datasets_last_values()).unwrap();
let values = axum::Json(values);
values.into_response()
}
src/server/api/handlers/mod.rs
+2 -5
View File
@@ -1,8 +1,5 @@
mod dataset;
mod extension;
mod fallback;
mod response;
mod last_values;
pub use dataset::*;
pub use fallback::*;
pub use last_values::*;
src/server/api/handlers/response.rs
-167
View File
@@ -1,167 +0,0 @@
use std::fmt::Debug;
use axum::response::{IntoResponse, Json, Response};
use bincode::Decode;
use serde::de::DeserializeOwned;
use serde::Serialize;
use crate::{
server::{
api::structs::{Chunk, Kind, Route},
header_map::HeaderMapUtils,
},
structs::{Date, MapValue, SerializedBTreeMap, SerializedVec},
};
use super::extension::Extension;
#[derive(Serialize)]
struct WrappedDataset<'a, T>
where
T: Serialize,
{
source: &'a str,
chunk: Chunk,
dataset: T,
}
#[derive(Serialize)]
struct WrappedValue<T>
where
T: Serialize,
{
value: T,
}
pub fn typed_value_to_response<T>(
kind: Kind,
route: &Route,
chunk: Option<Chunk>,
id: String,
extension: Option<Extension>,
) -> color_eyre::Result<Response>
where
T: Serialize + Debug + DeserializeOwned + Decode + MapValue,
{
Ok(match kind {
Kind::Date => {
let dataset = if chunk.is_some() {
route
.serialization
.import::<SerializedBTreeMap<Date, T>>(&route.file_path)?
} else {
SerializedBTreeMap::<Date, T>::import_all(&route.file_path, &route.serialization)
};
if extension == Some(Extension::CSV) {
let mut csv = format!("date,{}\n", id);
dataset.map.iter().for_each(|(k, v)| {
csv += &format!("{},{:?}\n", k, v);
});
string_to_response(csv, extension)
} else {
dataset_to_response(dataset, chunk, extension)
}
}
Kind::Height => {
let dataset = if chunk.is_some() {
route
.serialization
.import::<SerializedVec<T>>(&route.file_path)?
} else {
SerializedVec::<T>::import_all(&route.file_path, &route.serialization)
};
if extension == Some(Extension::CSV) {
let mut csv = format!("height,{}\n", id);
let starting_height = chunk.map_or(0, |chunk| chunk.id);
dataset.map.iter().enumerate().for_each(|(k, v)| {
csv += &format!("{},{:?}\n", starting_height + k, v);
});
string_to_response(csv, extension)
} else {
dataset_to_response(dataset, chunk, extension)
}
}
Kind::Last => {
let value = route.serialization.import::<T>(&route.file_path)?;
if extension == Some(Extension::JSON) {
value_to_response(WrappedValue { value }, extension)
} else {
value_to_response(value, extension)
}
}
})
}
pub fn string_to_response(s: String, extension: Option<Extension>) -> Response {
update_reponse_headers(s.into_response(), 5, extension)
}
pub fn value_to_response<T>(value: T, extension: Option<Extension>) -> Response
where
T: Serialize,
{
update_reponse_headers(generic_to_reponse(value, None), 1, extension)
}
fn dataset_to_response<T>(
dataset: T,
chunk: Option<Chunk>,
extension: Option<Extension>,
) -> Response
where
T: Serialize,
{
update_reponse_headers(generic_to_reponse(dataset, chunk), 5, extension)
}
pub fn generic_to_reponse<T>(generic: T, chunk: Option<Chunk>) -> Response
where
T: Serialize,
{
if let Some(chunk) = chunk {
Json(WrappedDataset {
source: "https://kibo.money",
chunk,
dataset: generic,
})
.into_response()
} else {
Json(generic).into_response()
}
}
pub fn update_reponse_headers(
mut response: Response,
cache_time: u64,
extension: Option<Extension>,
) -> Response {
let headers = response.headers_mut();
let max_age = cache_time;
let stale_while_revalidate = max_age;
headers.insert_cors();
headers.insert_cache_control_revalidate(max_age, stale_while_revalidate);
match extension {
Some(extension) => {
headers.insert_content_disposition_attachment();
match extension {
Extension::CSV => headers.insert_content_type_text_csv(),
Extension::JSON => headers.insert_content_type_application_json(),
}
}
_ => headers.insert_content_type_application_json(),
}
response
}
src/server/api/mod.rs
+3 -4
View File
@@ -1,5 +1,5 @@
use axum::{routing::get, Router};
use handlers::{dataset_handler, fallback};
use handlers::{dataset_handler, last_values_handler};
use super::AppState;
@@ -14,8 +14,7 @@ pub trait ApiRoutes {
impl ApiRoutes for Router<AppState> {
fn add_api_routes(self) -> Self {
self.route(&format!("{API_URL_PREFIX}/*path"), get(dataset_handler))
.route(&format!("{API_URL_PREFIX}/"), get(fallback))
.route(API_URL_PREFIX, get(fallback))
self.route(&format!("{API_URL_PREFIX}/last"), get(last_values_handler))
.route(&format!("{API_URL_PREFIX}/*path"), get(dataset_handler))
}
}
src/server/api/structs/chunk.rs → src/server/api/structs/chunk_metadata.rs
+1 -1
View File
@@ -1,7 +1,7 @@
use serde::{Deserialize, Serialize};
#[derive(Debug, Serialize, Deserialize)]
pub struct Chunk {
pub struct ChunkMetadata {
pub id: usize,
pub previous: Option<String>,
pub next: Option<String>,
src/server/api/handlers/extension.rs → src/server/api/structs/extension.rs
-9
View File
@@ -31,13 +31,4 @@ impl Extension {
Extension::JSON => "json",
}
}
pub fn to_dot_str(&self) -> String {
format!(".{}", self.to_str())
}
pub fn remove_extension(s: &str) -> String {
s.replace(&Self::CSV.to_dot_str(), "")
.replace(&Self::JSON.to_dot_str(), "")
}
}
src/server/api/structs/kind.rs
+54 -1
View File
@@ -1,6 +1,59 @@
#[derive(PartialEq, Eq, Clone, Copy)]
use std::collections::BTreeSet;
use color_eyre::eyre::{eyre, ContextCompat};
use serde::Deserialize;
use crate::structs::{AnyMap, Date, Height, MapKey};
#[derive(Debug, Clone, Copy, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
pub enum Kind {
Date,
Height,
Last,
}
impl TryFrom<&String> for Kind {
type Error = color_eyre::Report;
fn try_from(str: &String) -> Result<Self, Self::Error> {
Ok(
match str
.to_lowercase()
.chars()
.next()
.context("Expect kind to have first letter")?
{
'd' => Self::Date,
'h' => Self::Height,
'l' => Self::Last,
_ => return Err(eyre!("Bad kind")),
},
)
}
}
impl From<&(dyn AnyMap + Send + Sync)> for BTreeSet<Kind> {
fn from(map: &(dyn AnyMap + Send + Sync)) -> Self {
let mut s = Self::new();
if map.key_name() == Date::map_name() {
s.insert(Kind::Date);
}
if map.key_name() == Height::map_name() {
s.insert(Kind::Height);
}
if map.last_value().is_some() {
s.insert(Kind::Last);
}
s
}
}
// impl std::fmt::Display for Kind {
// fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
// match *self {
// Self::Date => write!(f, "date"),
// Self::Height => write!(f, "height"),
// Self::Last => write!(f, "last"),
// }
// }
// }
src/server/api/structs/mod.rs
+8 -4
View File
@@ -1,9 +1,13 @@
mod chunk;
mod chunk_metadata;
mod extension;
mod kind;
mod paths;
mod range;
mod route;
mod routes;
pub use chunk::*;
pub use chunk_metadata::*;
pub use extension::*;
pub use kind::*;
pub use paths::*;
pub use range::*;
pub use route::*;
pub use routes::*;
src/server/api/structs/paths.rs
-9
View File
@@ -1,9 +0,0 @@
use std::collections::BTreeMap;
use derive_deref::{Deref, DerefMut};
use serde::Serialize;
use crate::server::Grouped;
#[derive(Clone, Default, Deref, DerefMut, Debug, Serialize)]
pub struct Paths(pub Grouped<BTreeMap<String, String>>);
src/server/api/structs/range.rs
+32
View File
@@ -0,0 +1,32 @@
use axum::extract::Query;
use color_eyre::eyre::eyre;
use crate::server::api::handlers::DatasetParams;
pub enum DatasetRange {
All,
Chunk(DatasetRangeChunk),
}
impl TryFrom<&Query<DatasetParams>> for DatasetRange {
type Error = color_eyre::Report;
fn try_from(query: &Query<DatasetParams>) -> Result<Self, Self::Error> {
if let Some(chunk) = query.chunk {
if query.all.is_some() {
Err(eyre!("chunk and all are exclusive"))
} else {
Ok(Self::Chunk(DatasetRangeChunk::Chunk(chunk)))
}
} else if query.all.is_some() {
Ok(Self::All)
} else {
Ok(Self::Chunk(DatasetRangeChunk::Last))
}
}
}
pub enum DatasetRangeChunk {
Chunk(usize),
Last,
}
src/server/api/structs/route.rs
+33
View File
@@ -0,0 +1,33 @@
use std::{collections::BTreeSet, path::PathBuf};
use crate::{io::Serialization, structs::AnyMap};
use super::Kind;
#[derive(Debug, Clone)]
pub struct Route {
pub type_name: String,
pub list: BTreeSet<Kind>,
pub path: PathBuf,
pub serialization: Serialization,
}
impl Route {
pub fn update(&mut self, map: &(dyn AnyMap + Send + Sync)) {
self.list.append(&mut BTreeSet::from(map));
if self.serialization != map.serialization() {
panic!("route.upate() different serialization")
}
}
}
impl From<&(dyn AnyMap + Send + Sync)> for Route {
fn from(map: &(dyn AnyMap + Send + Sync)) -> Self {
Self {
list: BTreeSet::from(map),
path: map.path_parent().to_owned(),
type_name: map.type_name().split("::").last().unwrap().to_owned(),
serialization: map.serialization(),
}
}
}
src/server/api/structs/routes.rs
+27 -146
View File
@@ -1,135 +1,49 @@
use std::{
collections::{BTreeMap, HashMap},
fs,
path::PathBuf,
};
use std::collections::BTreeMap;
use derive_deref::{Deref, DerefMut};
use itertools::Itertools;
use crate::{
io::Serialization,
server::{api::API_URL_PREFIX, Grouped},
parser::{AnyDatasets, Datasets},
structs::Config,
};
use super::Paths;
use super::Route;
#[derive(Clone, Debug)]
pub struct Route {
pub url_path: String,
pub file_path: PathBuf,
pub values_type: String,
pub serialization: Serialization,
}
#[derive(Clone, Default, Deref, DerefMut)]
pub struct Routes(pub Grouped<HashMap<String, Route>>);
#[derive(Debug, Clone, Default, Deref, DerefMut)]
pub struct Routes(BTreeMap<String, Route>);
const WEBSITE_TYPES_PATH: &str = "../website/scripts/types";
impl Routes {
pub fn build(paths_to_type: BTreeMap<PathBuf, String>, config: &Config) -> Self {
let mut routes = Routes::default();
paths_to_type.into_iter().for_each(|(path, value)| {
let try_from_path = if path.is_file() {
path.clone()
} else {
fs::read_dir(&path)
.unwrap_or_else(|_| {
dbg!(&path);
panic!();
})
.map(|e| e.unwrap().path())
.find(|e| e.is_file())
.unwrap()
};
let serialization =
Serialization::try_from(&try_from_path).unwrap_or(Serialization::Binary);
let file_path_ser = path.to_str().unwrap().to_owned();
let split_key = file_path_ser.replace(
&format!("{}/", config.path_datasets().to_str().unwrap()),
"",
);
let split_key =
split_key.replace(&format!("{}/", config.path_kibodir().to_str().unwrap()), "");
let mut split_key = split_key.split('/').collect_vec();
let last = split_key.pop().unwrap().to_owned();
let last = last.split('.').next().unwrap();
// Use case for: "../datasets/last": "Value",
if split_key.is_empty() {
split_key.push("last");
}
let map_key = split_key.iter().join("_");
let url_path = split_key.iter().join("-");
let values_type = value.to_owned();
match last {
"date" => {
routes.date.insert(
map_key,
Route {
url_path: format!("date-to-{url_path}"),
file_path: path,
values_type,
serialization,
},
);
}
"height" => {
routes.height.insert(
map_key,
Route {
url_path: format!("height-to-{url_path}"),
file_path: path,
values_type,
serialization,
},
);
}
"last" => {
routes.last.insert(
map_key,
Route {
url_path,
file_path: path,
values_type,
serialization,
},
);
}
_ => {
dbg!(&path, value, &last, &split_key);
panic!("")
}
}
});
routes
pub fn build(datasets: &Datasets, config: &Config) -> Self {
datasets
.to_any_dataset_vec()
.into_iter()
.flat_map(|dataset| dataset.to_all_map_vec())
.fold(Self::default(), |mut routes, map| {
routes
.entry(map.id(config))
.or_insert_with(|| Route::from(map))
.update(map);
routes
})
}
pub fn generate_dts_file(&self) {
let map_to_type = |name: &str, map: &HashMap<String, Route>| -> String {
let paths = map
.values()
.map(|route| format!("\"{}\"", route.url_path))
.join(" | ");
// let map_to_type = |name: &str, map: &HashMap<String, Route>| -> String {
// let paths = map
// .values()
// .map(|route| format!("\"{}\"", route.url_path))
// .join(" | ");
format!("export type {}Path = {};\n", name, paths)
};
// format!("export type {}Path = {};\n", name, paths)
// };
let date_type = map_to_type("Date", &self.date);
// let date_type = map_to_type("Date", &self.date);
let height_type = map_to_type("Height", &self.height);
// let height_type = map_to_type("Height", &self.height);
let last_type = map_to_type("Last", &self.last);
// let last_type = map_to_type("Last", &self.last);
// fs::write(
// format!("{WEBSITE_TYPES_PATH}/paths.d.ts"),
@@ -137,37 +51,4 @@ impl Routes {
// )
// .unwrap();
}
pub fn to_full_paths(&self, host: String) -> Paths {
let url = {
let scheme = if host.contains("0.0.0.0") || host.contains("localhost") {
"http"
} else {
"https"
};
format!("{scheme}://{host}")
};
let transform = |map: &HashMap<String, Route>| -> BTreeMap<String, String> {
map.iter()
.map(|(key, route)| {
(
key.to_owned(),
format!("{url}{API_URL_PREFIX}/{}", route.url_path.to_owned()),
)
})
.collect()
};
let date_paths = transform(&self.date);
let height_paths = transform(&self.height);
let last_paths = transform(&self.last);
Paths(Grouped {
date: date_paths,
height: height_paths,
last: last_paths,
})
}
}
src/server/header_map.rs
+1
View File
@@ -31,6 +31,7 @@ pub trait HeaderMapUtils {
) -> color_eyre::Result<(DateTime<Utc>, Option<Response<Body>>)>;
fn insert_cache_control_immutable(&mut self);
#[allow(unused)]
fn insert_cache_control_revalidate(&mut self, max_age: u64, stale_while_revalidate: u64);
fn insert_last_modified(&mut self, date: DateTime<Utc>);
src/server/mod.rs
+12 -21
View File
@@ -1,43 +1,32 @@
use std::{collections::BTreeMap, path::PathBuf, sync::Arc};
use std::{sync::Arc, time::Instant};
use api::{structs::Routes, ApiRoutes};
use axum::{routing::get, serve, Router};
use color_eyre::owo_colors::OwoColorize;
use log::{error, info};
use reqwest::StatusCode;
use serde::Serialize;
use tokio::net::TcpListener;
use tower_http::compression::CompressionLayer;
use website::WebsiteRoutes;
use crate::structs::Config;
mod api;
pub mod api;
mod header_map;
mod website;
#[derive(Clone, Debug, Default, Serialize)]
pub struct Grouped<T> {
pub date: T,
pub height: T,
pub last: T,
}
#[derive(Clone)]
pub struct AppState {
routes: Arc<Routes>,
config: Config,
}
pub async fn main(
paths_to_type: BTreeMap<PathBuf, String>,
config: &Config,
) -> color_eyre::Result<()> {
let routes = Routes::build(paths_to_type, config);
pub async fn main(routes: Routes, config: Config) -> color_eyre::Result<()> {
routes.generate_dts_file();
let state = AppState {
routes: Arc::new(routes),
config,
};
let compression_layer = CompressionLayer::new()
@@ -72,11 +61,13 @@ pub async fn main(
Ok(())
}
pub fn log_result(code: StatusCode, path: &str) {
pub fn log_result(code: StatusCode, path: &str, instant: Instant) {
let time = format!("{}µs", instant.elapsed().as_micros());
let time = time.bright_black();
match code {
StatusCode::INTERNAL_SERVER_ERROR => error!("{} {}", code.as_u16().red(), path),
StatusCode::NOT_MODIFIED => info!("{} {}", code.as_u16().bright_black(), path),
StatusCode::OK => info!("{} {}", code.as_u16().green(), path),
_ => error!("{} {}", code.as_u16().red(), path),
StatusCode::INTERNAL_SERVER_ERROR => error!("{} {} {}", code.as_u16().red(), path, time),
StatusCode::NOT_MODIFIED => info!("{} {} {}", code.as_u16().bright_black(), path, time),
StatusCode::OK => info!("{} {} {}", code.as_u16().green(), path, time),
_ => error!("{} {} {}", code.as_u16().red(), path, time),
}
}
src/server/website/handlers/file.rs
+4 -2
View File
@@ -1,6 +1,7 @@
use std::{
fs::{self},
path::{Path, PathBuf},
time::Instant,
};
use axum::{
@@ -27,6 +28,8 @@ pub async fn index_handler(headers: HeaderMap) -> Response {
}
fn any_handler(headers: HeaderMap, path: Option<extract::Path<String>>) -> Response {
let instant = Instant::now();
let response = if let Some(path) = path.as_ref() {
let path = path.0.replace("..", "").replace("\\", "");
@@ -56,6 +59,7 @@ fn any_handler(headers: HeaderMap, path: Option<extract::Path<String>>) -> Respo
log_result(
response.status(),
&format!("/{}", path.map_or("".to_owned(), |p| p.0)),
instant,
);
response
@@ -124,8 +128,6 @@ fn _path_to_response(headers: &HeaderMap, path: &Path) -> color_eyre::Result<Res
})
{
headers.insert_cache_control_immutable();
} else {
headers.insert_cache_control_revalidate(1, 1);
}
}
src/structs/any_map.rs
+9 -21
View File
@@ -1,35 +1,23 @@
use std::path::{Path, PathBuf};
use std::path::Path;
use serde_json::Value;
use super::{MapKind, MapPath};
use crate::io::Serialization;
use super::{Config, MapKind, MapPath};
pub trait AnyMap {
fn path(&self) -> &Path;
fn path_parent(&self) -> &Path;
fn path_last(&self) -> &Option<MapPath>;
fn last_value(&self) -> Option<Value>;
fn t_name(&self) -> &str;
fn get_paths_to_type(&self) -> Vec<(PathBuf, String)> {
let t_name = self.t_name().to_string();
if let Some(path_last) = self.path_last() {
vec![
(self.path().to_owned(), t_name.clone()),
(path_last.unwrap().to_owned(), t_name),
]
} else {
vec![(self.path().to_owned(), t_name)]
}
}
fn serialization(&self) -> Serialization;
fn type_name(&self) -> &str;
fn key_name(&self) -> &str;
fn pre_export(&mut self);
fn export(&self) -> color_eyre::Result<()>;
fn post_export(&mut self);
fn delete_files(&self);
fn kind(&self) -> MapKind;
fn id(&self, config: &Config) -> String;
}
src/structs/config.rs
+6 -2
View File
@@ -10,6 +10,8 @@ use color_eyre::eyre::eyre;
use log::info;
use serde::{Deserialize, Serialize};
use crate::io::JSON_EXTENSION;
use super::MapPath;
#[derive(Parser, Debug, Clone, Default, Serialize, Deserialize, PartialEq)]
@@ -74,6 +76,8 @@ pub struct Config {
}
impl Config {
pub const DATASET_DIR_NAME: &str = "datasets";
pub fn import() -> color_eyre::Result<Self> {
let path = Self::path_dot_kibo();
let _ = fs::create_dir_all(&path);
@@ -285,7 +289,7 @@ impl Config {
}
pub fn path_datasets_last_values(&self) -> MapPath {
self.path_datasets().join("last.json")
self.path_datasets().join(&format!("last.{JSON_EXTENSION}"))
}
pub fn path_price(&self) -> MapPath {
@@ -293,7 +297,7 @@ impl Config {
}
pub fn path_databases(&self) -> PathBuf {
self.path_kibodir().join("databases")
self.path_kibodir().join(Self::DATASET_DIR_NAME)
}
pub fn path_states(&self) -> PathBuf {
src/structs/date_map.rs
+2 -2
View File
@@ -1,10 +1,10 @@
use std::iter::Sum;
use super::{
AnyMap, Date, DateMapChunkId, GenericMap, Height, HeightMap, MapValue, SerializedBTreeMap,
AnyMap, Date, DateMapChunkId, GenericMap, Height, HeightMap, MapValue, SerializedDateMap,
};
pub type DateMap<Value> = GenericMap<Date, Value, DateMapChunkId, SerializedBTreeMap<Date, Value>>;
pub type DateMap<Value> = GenericMap<Date, Value, DateMapChunkId, SerializedDateMap<Value>>;
impl<Value> DateMap<Value>
where
src/structs/date_map_chunk_id.rs
+9 -1
View File
@@ -15,7 +15,7 @@ impl DateMapChunkId {
}
impl MapChunkId for DateMapChunkId {
fn to_name(&self) -> String {
fn to_string(&self) -> String {
self.0.to_string()
}
@@ -39,4 +39,12 @@ impl MapChunkId for DateMapChunkId {
fn from_usize(id: usize) -> Self {
Self(id as i32)
}
fn next(&self) -> Option<Self> {
self.0.checked_add(1).map(Self)
}
fn previous(&self) -> Option<Self> {
self.0.checked_sub(1).map(Self)
}
}
src/structs/generic_map.rs
+48 -15
View File
@@ -1,6 +1,6 @@
use std::{
collections::{BTreeMap, VecDeque},
fmt::Debug,
fmt::{Debug, Display},
fs,
iter::Sum,
mem,
@@ -19,7 +19,7 @@ use crate::{
utils::{get_percentile, LossyFrom},
};
use super::{AnyMap, MapPath, MapValue};
use super::{AnyMap, Config, MapPath, MapValue};
#[derive(Debug, Clone, Copy, Allocative, PartialEq, Eq)]
pub enum MapKind {
@@ -29,7 +29,7 @@ pub enum MapKind {
pub trait MapKey<ChunkId>
where
Self: Sized + PartialOrd + Ord + Clone + Copy + Debug,
Self: Sized + PartialOrd + Ord + Clone + Copy + Debug + Display,
ChunkId: MapChunkId,
{
fn to_chunk_id(&self) -> ChunkId;
@@ -61,16 +61,21 @@ where
fn get(&self, serialized_key: &Key) -> Option<&Value>;
fn last(&self) -> Option<&Value>;
fn extend(&mut self, map: BTreeMap<Key, Value>);
fn import_all(path: &Path, serialization: &Serialization) -> Self;
fn to_csv(self, id: &str) -> String;
fn map(&self) -> &impl Serialize;
}
pub trait MapChunkId
where
Self: Ord + Debug + Copy + Clone,
{
fn to_name(&self) -> String;
fn to_string(&self) -> String;
fn from_path(path: &Path) -> color_eyre::Result<Self>;
fn to_usize(self) -> usize;
fn from_usize(id: usize) -> Self;
fn previous(&self) -> Option<Self>;
fn next(&self) -> Option<Self>;
}
#[derive(Debug, Allocative)]
@@ -79,6 +84,7 @@ pub struct GenericMap<Key, Value, ChunkId, Serialized> {
kind: MapKind,
path_all: MapPath,
path_parent: MapPath,
path_last: Option<MapPath>,
chunks_in_memory: usize,
@@ -150,6 +156,7 @@ where
kind,
path_all,
path_parent: path.to_owned(),
path_last,
chunks_in_memory,
@@ -177,15 +184,7 @@ where
}
});
s.initial_last_key = s
.imported
.iter()
.last()
.and_then(|(last_chunk_id, serialized)| serialized.get_last_key(last_chunk_id));
s.initial_first_unsafe_key = s
.initial_last_key
.and_then(|last_key| last_key.to_first_unsafe());
s.set_initial_keys();
// if s.initial_first_unsafe_key.is_none() {
// log(&format!("Missing dataset: {path:?}/{}", Key::map_name()));
@@ -194,6 +193,18 @@ where
s
}
fn set_initial_keys(&mut self) {
self.initial_last_key = self
.imported
.iter()
.last()
.and_then(|(last_chunk_id, serialized)| serialized.get_last_key(last_chunk_id));
self.initial_first_unsafe_key = self
.initial_last_key
.and_then(|last_key| last_key.to_first_unsafe());
}
fn read_dir(&self) -> BTreeMap<ChunkId, PathBuf> {
Self::_read_dir(&self.path_all, &self.serialization)
}
@@ -308,10 +319,26 @@ where
Key: MapKey<ChunkId>,
Serialized: MapSerialized<Key, Value, ChunkId>,
{
fn id(&self, config: &Config) -> String {
let path_to_string = |p: &Path| p.to_str().unwrap().to_owned();
path_to_string(self.path_parent())
.replace(&path_to_string(&config.path_kibodir()), "")
.replace(&format!("/{}/", Config::DATASET_DIR_NAME), "")
.replace("/", "-")
}
fn serialization(&self) -> Serialization {
self.serialization
}
fn path(&self) -> &Path {
&self.path_all
}
fn path_parent(&self) -> &Path {
&self.path_parent
}
fn path_last(&self) -> &Option<MapPath> {
&self.path_last
}
@@ -323,10 +350,14 @@ where
.and_then(|v| serde_json::to_value(v).ok())
}
fn t_name(&self) -> &str {
fn type_name(&self) -> &str {
std::any::type_name::<Value>()
}
fn key_name(&self) -> &str {
Key::map_name()
}
fn pre_export(&mut self) {
self.to_insert.iter_mut().for_each(|(chunk_id, map)| {
if let Some((key, _)) = map.first_key_value() {
@@ -351,6 +382,8 @@ where
.or_insert(Serialized::new(self.version))
.extend(mem::take(map));
});
self.set_initial_keys();
}
fn export(&self) -> color_eyre::Result<()> {
@@ -367,7 +400,7 @@ where
panic!();
});
let path = self.path_all.join(&chunk_id.to_name());
let path = self.path_all.join(&chunk_id.to_string());
self.serialization.export(&path, serialized)?;
src/structs/height_map_chunk_id.rs
+11 -1
View File
@@ -19,7 +19,7 @@ impl HeightMapChunkId {
}
impl MapChunkId for HeightMapChunkId {
fn to_name(&self) -> String {
fn to_string(&self) -> String {
let start = ***self;
let end = start + HEIGHT_MAP_CHUNK_SIZE;
@@ -46,4 +46,14 @@ impl MapChunkId for HeightMapChunkId {
fn from_usize(id: usize) -> Self {
Self(Height::new(id as u32))
}
fn next(&self) -> Option<Self> {
self.checked_add(HEIGHT_MAP_CHUNK_SIZE)
.map(|h| Self(Height::new(h)))
}
fn previous(&self) -> Option<Self> {
self.checked_sub(HEIGHT_MAP_CHUNK_SIZE)
.map(|h| Self(Height::new(h)))
}
}
src/structs/map_value.rs
+2 -1
View File
@@ -1,4 +1,4 @@
use std::fmt::Debug;
use std::fmt::{Debug, Display};
use allocative::Allocative;
use bincode::{Decode, Encode};
@@ -18,6 +18,7 @@ pub trait MapValue:
+ Sync
+ Send
+ Allocative
+ Display
{
}
src/structs/mod.rs
-1
View File
@@ -72,7 +72,6 @@ pub use map_value::*;
pub use ohlc::*;
pub use partial_txout_data::*;
pub use price::*;
pub use rpc::*;
pub use sent_data::*;
pub use serialized_btreemap::*;
pub use serialized_vec::*;
src/structs/ohlc.rs
+12
View File
@@ -1,3 +1,5 @@
use std::fmt::{self};
use allocative::Allocative;
use bincode::{Decode, Encode};
use serde::{Deserialize, Serialize};
@@ -10,3 +12,13 @@ pub struct OHLC {
pub low: f32,
pub close: f32,
}
impl fmt::Display for OHLC {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
f,
"{{ open: {}, high: {}, low: {}, close: {} }}",
self.open, self.high, self.low, self.close
)
}
}
src/structs/serialized_btreemap.rs
+35 -32
View File
@@ -6,7 +6,9 @@ use serde::{de::DeserializeOwned, Deserialize, Serialize};
use crate::io::Serialization;
use super::{DateMap, MapChunkId, MapKey, MapSerialized, MapValue};
use super::{Date, DateMap, MapChunkId, MapKey, MapSerialized, MapValue};
pub type SerializedDateMap<T> = SerializedBTreeMap<Date, T>;
#[derive(Debug, Default, Serialize, Deserialize, Encode, Decode, Allocative)]
pub struct SerializedBTreeMap<Key, Value>
@@ -17,41 +19,11 @@ where
pub map: BTreeMap<Key, Value>,
}
impl<Key, Value> SerializedBTreeMap<Key, Value>
where
Key: Ord,
{
pub fn import_all<ChunkId>(path: &Path, serialization: &Serialization) -> Self
where
Self: Debug + Serialize + DeserializeOwned + Encode + Decode,
ChunkId: MapChunkId,
Key: MapKey<ChunkId>,
Value: MapValue,
{
let mut s = None;
DateMap::<usize>::_read_dir(path, serialization)
.iter()
.for_each(|(_, path)| {
let map = serialization.import::<Self>(path).unwrap();
if s.is_none() {
s.replace(map);
} else {
#[allow(clippy::unnecessary_unwrap)]
s.as_mut().unwrap().map.extend(map.map);
}
});
s.unwrap()
}
}
impl<Key, Value, ChunkId> MapSerialized<Key, Value, ChunkId> for SerializedBTreeMap<Key, Value>
where
Self: Debug + Serialize + DeserializeOwned + Encode + Decode,
ChunkId: MapChunkId,
Key: MapKey<ChunkId>,
Key: MapKey<ChunkId> + Serialize,
Value: MapValue,
{
fn new(version: u32) -> Self {
@@ -80,4 +52,35 @@ where
fn extend(&mut self, map: BTreeMap<Key, Value>) {
self.map.extend(map)
}
fn import_all(path: &Path, serialization: &Serialization) -> Self {
let mut s = None;
DateMap::<usize>::_read_dir(path, serialization)
.iter()
.for_each(|(_, path)| {
let map = serialization.import::<Self>(path).unwrap();
if s.is_none() {
s.replace(map);
} else {
#[allow(clippy::unnecessary_unwrap)]
s.as_mut().unwrap().map.extend(map.map);
}
});
s.unwrap()
}
fn to_csv(self, id: &str) -> String {
let mut csv = format!("{},{}\n", Key::map_name(), id);
self.map.iter().for_each(|(k, v)| {
csv += &format!("{},{}\n", k, v);
});
csv
}
fn map(&self) -> &impl Serialize {
&self.map
}
}
src/structs/serialized_vec.rs
+35 -25
View File
@@ -14,31 +14,6 @@ pub struct SerializedVec<Value> {
pub map: Vec<Value>,
}
impl<Value> SerializedVec<Value> {
pub fn import_all(path: &Path, serialization: &Serialization) -> Self
where
Self: Debug + Serialize + DeserializeOwned + Encode + Decode,
Value: MapValue,
{
let mut s = None;
HeightMap::<usize>::_read_dir(path, serialization)
.iter()
.for_each(|(_, path)| {
let mut map = serialization.import::<Self>(path).unwrap();
if s.is_none() {
s.replace(map);
} else {
#[allow(clippy::unnecessary_unwrap)]
s.as_mut().unwrap().map.append(&mut map.map);
}
});
s.unwrap()
}
}
impl<Key, Value, ChunkId> MapSerialized<Key, Value, ChunkId> for SerializedVec<Value>
where
Self: Debug + Serialize + DeserializeOwned + Encode + Decode,
@@ -87,4 +62,39 @@ where
}
});
}
fn import_all(path: &Path, serialization: &Serialization) -> Self
where
Self: Debug + Serialize + DeserializeOwned + Encode + Decode,
Value: MapValue,
{
let mut s = None;
HeightMap::<usize>::_read_dir(path, serialization)
.iter()
.for_each(|(_, path)| {
let mut map = serialization.import::<Self>(path).unwrap();
if s.is_none() {
s.replace(map);
} else {
#[allow(clippy::unnecessary_unwrap)]
s.as_mut().unwrap().map.append(&mut map.map);
}
});
s.unwrap()
}
fn to_csv(self, id: &str) -> String {
let mut csv = format!("{},{}\n", Key::map_name(), id);
self.map.iter().enumerate().for_each(|(k, v)| {
csv += &format!("{:?},{:?}\n", k, v);
});
csv
}
fn map(&self) -> &impl Serialize {
&self.map
}
}
src/structs/timestamp.rs
+7 -1
View File
@@ -1,4 +1,4 @@
use std::ops::Sub;
use std::{fmt, ops::Sub};
use allocative::Allocative;
use bincode::{Decode, Encode};
@@ -81,3 +81,9 @@ impl Sub for Timestamp {
Self::wrap(self.0 - rhs.0)
}
}
impl fmt::Display for Timestamp {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "{}", **self)
}
}