commit 345ee1e632502e2956af92dff02d5c2955eb421c
parent 95d2e15e370d51c762a37eda4613acbef229ac4e
Author: MTRNord <mtrnord1@gmail.com>
Date: Mon, 22 May 2023 16:25:41 +0200
Add support for files downloaded using torrents
Diffstat:
| M | Cargo.lock | | | 29 | +++++++++++++++++++++++++++++ |
| M | Cargo.toml | | | 1 | + |
| M | src/main.rs | | | 207 | ++++++++++++++++++++++++++++++++++++++++++------------------------------------- |
| M | src/osm/osm_data.rs | | | 126 | +++++++++++++++++++++++++++++++++++++++++++++++++------------------------------ |
4 files changed, 219 insertions(+), 144 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
@@ -866,6 +866,7 @@ dependencies = [
"tokio",
"tracing",
"tracing-subscriber",
+ "walkdir",
]
[[package]]
@@ -1133,6 +1134,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f91339c0467de62360649f8d3e185ca8de4224ff281f66000de5eb2a77a79041"
[[package]]
+name = "same-file"
+version = "1.0.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "93fc1dc3aaa9bfed95e02e6eadabb4baf7e3078b0bd1b4d7b6b0b68378900502"
+dependencies = [
+ "winapi-util",
+]
+
+[[package]]
name = "scopeguard"
version = "1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1595,6 +1605,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426"
[[package]]
+name = "walkdir"
+version = "2.3.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "36df944cda56c7d8d8b7496af378e6b16de9284591917d307c9b4d313c44e698"
+dependencies = [
+ "same-file",
+ "winapi-util",
+]
+
+[[package]]
name = "want"
version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1741,6 +1761,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6"
[[package]]
+name = "winapi-util"
+version = "0.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "70ec6ce85bb158151cae5e5c87f95a8e97d2c0c4b001223f33a334e3ce5de178"
+dependencies = [
+ "winapi",
+]
+
+[[package]]
name = "winapi-x86_64-pc-windows-gnu"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
diff --git a/Cargo.toml b/Cargo.toml
@@ -20,3 +20,4 @@ time = { version = "0.3.21", features = ["formatting", "parsing"] }
tokio = { version = "1.28.1", features = ["full"] }
tracing = "0.1.37"
tracing-subscriber = "0.3.17"
+walkdir = "2.3.3"
diff --git a/src/main.rs b/src/main.rs
@@ -43,6 +43,11 @@ struct Cli {
/// This is to avoid causing a lot of load on the OSM servers
#[arg(long, default_value = "500")]
wait_time: u64,
+ /// If we should use torrent files from the cache instead for changesets
+ ///
+ /// These need to be downloaded manually and be put into the cache_folder/changesets/torrents folder
+ #[arg(long)]
+ use_torrents: bool,
}
#[tokio::main]
@@ -79,100 +84,105 @@ async fn main() -> Result<()> {
)?;
info!("Git repository initialized");
- // Main download loop
- let mut changeset_position_top = cli.start_changesets[0..3].parse::<u64>()?;
- let mut changeset_position_middle = cli.start_changesets[4..7].parse::<u64>()?;
- let mut changeset_position_bottom = cli.start_changesets[8..11].parse::<u64>()?;
- let mut changeset_position_middle_incremented = false;
- let mut changeset_position_top_incremented = false;
+ if !cli.use_torrents {
+ // Main download loop
+ let mut changeset_position_top = cli.start_changesets[0..3].parse::<u64>()?;
+ let mut changeset_position_middle = cli.start_changesets[4..7].parse::<u64>()?;
+ let mut changeset_position_bottom = cli.start_changesets[8..11].parse::<u64>()?;
+ let mut changeset_position_middle_incremented = false;
+ let mut changeset_position_top_incremented = false;
- loop {
- // Check for cache and use it if it exists
- let cache_file_path = format!(
- "{}/changesets/{:03}/{:03}/{:03}.osm.gz",
- cli.cache_path,
- changeset_position_top,
- changeset_position_middle,
- changeset_position_bottom
- );
-
- if std::path::Path::new(&cache_file_path).exists() {
- info!(
- "We already got the changeset file at {}. Skipping",
- cache_file_path
+ loop {
+ // Check for cache and use it if it exists
+ let cache_file_path = format!(
+ "{}/changesets/{:03}/{:03}/{:03}.osm.gz",
+ cli.cache_path,
+ changeset_position_top,
+ changeset_position_middle,
+ changeset_position_bottom
);
- // Increment the changeset position
- changeset_position_bottom += 1;
- changeset_position_middle_incremented = false;
- changeset_position_top_incremented = false;
- } else {
- {
- // First we download the changeset files
- let changeset_url = format!(
- "{}/{:03}/{:03}/{:03}.osm.gz",
- cli.changeset_server,
- changeset_position_top,
- changeset_position_middle,
- changeset_position_bottom
+ if std::path::Path::new(&cache_file_path).exists() {
+ info!(
+ "We already got the changeset file at {}. Skipping",
+ cache_file_path
);
- info!("Downloading changeset file from {}", changeset_url);
- let changeset_response: reqwest::Response =
- client.get(&changeset_url).send().await?;
- if changeset_response.status() == reqwest::StatusCode::NOT_FOUND {
- warn!("Changeset file not found at {}", changeset_url);
- // We've reached the end of the changesets for this bottom position.
- // If we incremented top and failed again, we're done.
- if changeset_position_top_incremented {
- info!("Finished or failed downloading changesets");
- info!(
- "Changeset position: {} {} {}",
- changeset_position_top,
- changeset_position_middle,
- changeset_position_bottom
- );
- warn!("Response body: {:?}", changeset_response.text().await?);
- // TODO: We want to have an endless loop here optionally so that we can keep trying to download changesets.
- break;
- }
- // We reset bottom to 0 and increment middle.
- // We also mark middle as incremented so that we increment top on the next failure.
- if !changeset_position_middle_incremented && changeset_position_bottom != 0 {
- changeset_position_bottom = 0;
- changeset_position_middle += 1;
- changeset_position_middle_incremented = true;
- changeset_position_top_incremented = false;
- } else {
- changeset_position_middle_incremented = false;
- changeset_position_top += 1;
- changeset_position_top_incremented = true;
- changeset_position_middle = 0;
- changeset_position_bottom = 0;
+
+ // Increment the changeset position
+ changeset_position_bottom += 1;
+ changeset_position_middle_incremented = false;
+ changeset_position_top_incremented = false;
+ } else {
+ {
+ // First we download the changeset files
+ let changeset_url = format!(
+ "{}/{:03}/{:03}/{:03}.osm.gz",
+ cli.changeset_server,
+ changeset_position_top,
+ changeset_position_middle,
+ changeset_position_bottom
+ );
+ info!("Downloading changeset file from {}", changeset_url);
+ let changeset_response: reqwest::Response =
+ client.get(&changeset_url).send().await?;
+ if changeset_response.status() == reqwest::StatusCode::NOT_FOUND {
+ warn!("Changeset file not found at {}", changeset_url);
+ // We've reached the end of the changesets for this bottom position.
+ // If we incremented top and failed again, we're done.
+ if changeset_position_top_incremented {
+ info!("Finished or failed downloading changesets");
+ info!(
+ "Changeset position: {} {} {}",
+ changeset_position_top,
+ changeset_position_middle,
+ changeset_position_bottom
+ );
+ warn!("Response body: {:?}", changeset_response.text().await?);
+ // TODO: We want to have an endless loop here optionally so that we can keep trying to download changesets.
+ break;
+ }
+ // We reset bottom to 0 and increment middle.
+ // We also mark middle as incremented so that we increment top on the next failure.
+ if !changeset_position_middle_incremented && changeset_position_bottom != 0
+ {
+ changeset_position_bottom = 0;
+ changeset_position_middle += 1;
+ changeset_position_middle_incremented = true;
+ changeset_position_top_incremented = false;
+ } else {
+ changeset_position_middle_incremented = false;
+ changeset_position_top += 1;
+ changeset_position_top_incremented = true;
+ changeset_position_middle = 0;
+ changeset_position_bottom = 0;
+ }
+ continue;
}
- continue;
- }
- let changeset_data = changeset_response.bytes().await?;
- info!("Caching changeset file to disk");
- std::fs::create_dir_all(std::path::Path::new(&cache_file_path).parent().unwrap())?;
- std::fs::write(&cache_file_path, &changeset_data)?;
- info!("Changeset file downloaded");
- };
+ let changeset_data = changeset_response.bytes().await?;
+ info!("Caching changeset file to disk");
+ std::fs::create_dir_all(
+ std::path::Path::new(&cache_file_path).parent().unwrap(),
+ )?;
+ std::fs::write(&cache_file_path, &changeset_data)?;
+ info!("Changeset file downloaded");
+ };
- // TODO: We need to dynamically do this based on the data instead. Otherwise we dont have ram larrge enough
- // let file = File::open(cache_file_path)?;
- // let changeset_data = unsafe { Mmap::map(&file)? };
+ // TODO: We need to dynamically do this based on the data instead. Otherwise we dont have ram larrge enough
+ // let file = File::open(cache_file_path)?;
+ // let changeset_data = unsafe { Mmap::map(&file)? };
- // let parsed_changeset = parse_changeset(&changeset_data)?;
- // info!("Changeset file parsed");
- // changesets.extend(parsed_changeset);
+ // let parsed_changeset = parse_changeset(&changeset_data)?;
+ // info!("Changeset file parsed");
+ // changesets.extend(parsed_changeset);
- // Increment the changeset position
- changeset_position_bottom += 1;
- changeset_position_middle_incremented = false;
- changeset_position_top_incremented = false;
+ // Increment the changeset position
+ changeset_position_bottom += 1;
+ changeset_position_middle_incremented = false;
+ changeset_position_top_incremented = false;
- // Wait a few seconds before downloading the next changeset file
- tokio::time::sleep(Duration::from_millis(cli.wait_time)).await;
+ // Wait a few seconds before downloading the next changeset file
+ tokio::time::sleep(Duration::from_millis(cli.wait_time)).await;
+ }
}
}
@@ -188,17 +198,20 @@ async fn main() -> Result<()> {
// Check for cache and use it if it exists
let cache_file_path = format!(
"{}/replication/{:03}/{:03}/{:03}.osm.gz",
- cli.cache_path,
- changeset_position_top,
- changeset_position_middle,
- changeset_position_bottom
+ cli.cache_path, data_position_top, data_position_middle, data_position_bottom
);
if std::path::Path::new(&cache_file_path).exists() {
info!("Using cached data file at {}", cache_file_path);
let file = File::open(&cache_file_path)?;
let data = unsafe { Mmap::map(&file)? };
- convert_objects_to_git(&repository, &author, &data, cli.cache_path.clone())?;
+ convert_objects_to_git(
+ &repository,
+ &author,
+ &data,
+ cli.cache_path.clone(),
+ cli.use_torrents,
+ )?;
info!("Data file parsed");
// Increment the data position
@@ -260,7 +273,13 @@ async fn main() -> Result<()> {
let file = File::open(cache_file_path)?;
let data = unsafe { Mmap::map(&file)? };
- convert_objects_to_git(&repository, &author, &data, cli.cache_path.clone())?;
+ convert_objects_to_git(
+ &repository,
+ &author,
+ &data,
+ cli.cache_path.clone(),
+ cli.use_torrents,
+ )?;
// Increment the data position
data_position_bottom += 1;
@@ -273,12 +292,6 @@ async fn main() -> Result<()> {
}
info!(
- "Downloaded changesets until {} {} {}",
- changeset_position_top,
- changeset_position_middle,
- changeset_position_bottom - 1
- );
- info!(
"Downloaded data until {} {} {}",
data_position_top,
data_position_middle,
diff --git a/src/osm/osm_data.rs b/src/osm/osm_data.rs
@@ -17,6 +17,7 @@ use std::{
};
use time::{format_description::well_known::Iso8601, OffsetDateTime};
use tracing::{debug, error, info, warn};
+use walkdir::WalkDir;
use crate::git::commit;
@@ -472,6 +473,7 @@ pub fn convert_objects_to_git(
committer: &Signature,
data: &[u8],
cache_folder: String,
+ use_torrents: bool,
) -> Result<()> {
// If the file is empty we skip it
if data.is_empty() {
@@ -848,8 +850,14 @@ pub fn convert_objects_to_git(
for changeset in changeset_list {
// Find the changeset within the files of the cache
- let changeset =
- find_changesets_in_cache(cache_folder.clone(), *changeset, None, None, None)?;
+ let changeset = find_changesets_in_cache(
+ cache_folder.clone(),
+ use_torrents,
+ *changeset,
+ None,
+ None,
+ None,
+ )?;
if let Some(changeset) = changeset {
// Get comment tag if it exists and trim it
@@ -947,6 +955,7 @@ pub fn convert_objects_to_git(
/// The changeset if found
fn find_changesets_in_cache(
cache_folder: String,
+ use_torrents: bool,
changeset_id: u64,
changeset_position_top: Option<u64>,
changeset_position_middle: Option<u64>,
@@ -956,57 +965,80 @@ fn find_changesets_in_cache(
//
// Each file has to be parsed using "parse_changeset(&changeset_data)?;"
- let changeset_folder = format!("{}/changesets", cache_folder);
- let mut changeset_position_top = changeset_position_top.unwrap_or(0);
- let mut changeset_position_middle = changeset_position_middle.unwrap_or(0);
- let mut changeset_position_bottom = changeset_position_bottom.unwrap_or(0);
- let changeset_file = format!(
- "{:03}/{:03}/{:03}.osm.gz",
- changeset_position_top, changeset_position_middle, changeset_position_bottom
- );
- let changeset_path = format!("{}/{}", changeset_folder, changeset_file);
-
- if !Path::new(&changeset_path).exists() {
- return Ok(None);
- }
+ let changeset_folder = if use_torrents {
+ format!("{}/changesets/torrents", cache_folder)
+ } else {
+ format!("{}/changesets", cache_folder)
+ };
+
+ if !use_torrents {
+ let mut changeset_position_top = changeset_position_top.unwrap_or(0);
+ let mut changeset_position_middle = changeset_position_middle.unwrap_or(0);
+ let mut changeset_position_bottom = changeset_position_bottom.unwrap_or(0);
+ let changeset_file = format!(
+ "{:03}/{:03}/{:03}.osm.gz",
+ changeset_position_top, changeset_position_middle, changeset_position_bottom
+ );
+ let changeset_path = format!("{}/{}", changeset_folder, changeset_file);
+
+ if !Path::new(&changeset_path).exists() {
+ return Ok(None);
+ }
- let mut changeset_file = File::open(changeset_path)?;
- let mut changeset_data = Vec::new();
- changeset_file.read_to_end(&mut changeset_data)?;
+ let mut changeset_file = File::open(changeset_path)?;
+ let mut changeset_data = Vec::new();
+ changeset_file.read_to_end(&mut changeset_data)?;
- let changesets = parse_changeset(&changeset_data)?;
+ let changesets = parse_changeset(&changeset_data)?;
- // Check if the file has the correct changeset id in vector otherwise we recurse to the next file
- if changesets.iter().any(|c| c.id == changeset_id) {
- return Ok(changesets.into_iter().find(|c| c.id == changeset_id));
- }
+ // Check if the file has the correct changeset id in vector otherwise we recurse to the next file
+ if changesets.iter().any(|c| c.id == changeset_id) {
+ return Ok(changesets.into_iter().find(|c| c.id == changeset_id));
+ }
- // We recurse to the next file since we found no changeset with the correct id
+ // We recurse to the next file since we found no changeset with the correct id
- if changeset_position_top == 999
- && changeset_position_middle == 999
- && changeset_position_bottom == 999
- {
- // Uhhhhhh?!
- return Ok(None);
- }
+ if changeset_position_top == 999
+ && changeset_position_middle == 999
+ && changeset_position_bottom == 999
+ {
+ // Uhhhhhh?!
+ return Ok(None);
+ }
- if changeset_position_middle == 999 && changeset_position_bottom == 999 {
- changeset_position_middle = 0;
- changeset_position_bottom = 0;
- changeset_position_top += 1;
- }
+ if changeset_position_middle == 999 && changeset_position_bottom == 999 {
+ changeset_position_middle = 0;
+ changeset_position_bottom = 0;
+ changeset_position_top += 1;
+ }
- if changeset_position_bottom == 999 {
- changeset_position_bottom = 0;
- changeset_position_middle += 1;
- }
+ if changeset_position_bottom == 999 {
+ changeset_position_bottom = 0;
+ changeset_position_middle += 1;
+ }
- find_changesets_in_cache(
- cache_folder,
- changeset_id,
- Some(changeset_position_top),
- Some(changeset_position_middle),
- Some(changeset_position_bottom),
- )
+ find_changesets_in_cache(
+ cache_folder,
+ use_torrents,
+ changeset_id,
+ Some(changeset_position_top),
+ Some(changeset_position_middle),
+ Some(changeset_position_bottom),
+ )
+ } else {
+ let mut changeset: Option<Changeset> = None;
+
+ for entry in WalkDir::new(changeset_folder) {
+ let mut changeset_file = File::open(entry?.path())?;
+ let mut changeset_data = Vec::new();
+ changeset_file.read_to_end(&mut changeset_data)?;
+
+ let changesets = parse_changeset(&changeset_data)?;
+ if changesets.iter().any(|c| c.id == changeset_id) {
+ changeset = changesets.into_iter().find(|c| c.id == changeset_id);
+ }
+ }
+
+ Ok(changeset)
+ }
}
