Skip to content

Commit 85a0aed

Browse files
pborzenkovpborzenkov
authored
Merge pull request #323 from superfly/corro-client
corro-client: don't depend on corro-agent
2 parents 72fc516 + ef8ea33 commit 85a0aed

File tree

8 files changed

+108
-87
lines changed

8 files changed

+108
-87
lines changed

‎Cargo.lock‎

Lines changed: 11 additions & 1 deletion
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎crates/corro-agent/Cargo.toml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ camino = { workspace = true }
1414
compact_str = { workspace = true }
1515
config = { workspace = true }
1616
corro-types = { path = "../corro-types" }
17+
corro-utils = { path = "../corro-utils" }
1718
eyre = { workspace = true }
1819
foca = { workspace = true }
1920
futures = { workspace = true }

‎crates/corro-agent/src/agent/run_root.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -94,7 +94,7 @@ async fn run(agent: Agent, opts: AgentOptions, pconf: PerfConfig) -> eyre::Resul
9494
util::initialise_foca(&agent).await;
9595

9696
// Load schema from paths
97-
let stmts = util::read_files_from_paths(&agent.config().db.schema_paths).await?;
97+
let stmts = corro_utils::read_files_from_paths(&agent.config().db.schema_paths).await?;
9898
if !stmts.is_empty() {
9999
if let Err(e) = execute_schema(&agent, stmts).await {
100100
error!("could not execute schema: {e}");

‎crates/corro-agent/src/agent/util.rs‎

Lines changed: 0 additions & 82 deletions
Original file line numberDiff line numberDiff line change
@@ -58,7 +58,6 @@ use std::{
5858
convert::Infallible,
5959
net::SocketAddr,
6060
ops::{Deref, RangeInclusive},
61-
path::Path,
6261
sync::{atomic::AtomicI64, Arc},
6362
time::{Duration, Instant},
6463
};
@@ -1446,87 +1445,6 @@ pub fn check_buffered_meta_to_clear(
14461445
conn.prepare_cached("SELECT EXISTS(SELECT 1 FROM __corro_seq_bookkeeping WHERE site_id = ? AND version >= ? AND version <= ?)")?.query_row(params![actor_id, versions.start(), versions.end()], |row| row.get(0))
14471446
}
14481447

1449-
pub async fn read_files_from_paths<P: AsRef<Path>>(
1450-
schema_paths: &[P],
1451-
) -> eyre::Result<Vec<String>> {
1452-
let mut contents = vec![];
1453-
1454-
for schema_path in schema_paths.iter() {
1455-
match tokio::fs::metadata(schema_path).await {
1456-
Ok(meta) => {
1457-
if meta.is_dir() {
1458-
match tokio::fs::read_dir(schema_path).await {
1459-
Ok(mut dir) => {
1460-
let mut entries = vec![];
1461-
1462-
while let Ok(Some(entry)) = dir.next_entry().await {
1463-
entries.push(entry);
1464-
}
1465-
1466-
let mut entries: Vec<_> = entries
1467-
.into_iter()
1468-
.filter_map(|entry| {
1469-
entry.path().extension().and_then(|ext| {
1470-
if ext == "sql" {
1471-
Some(entry)
1472-
} else {
1473-
None
1474-
}
1475-
})
1476-
})
1477-
.collect();
1478-
1479-
entries.sort_by_key(|entry| entry.path());
1480-
1481-
for entry in entries.iter() {
1482-
match tokio::fs::read_to_string(entry.path()).await {
1483-
Ok(s) => {
1484-
contents.push(s);
1485-
}
1486-
Err(e) => {
1487-
warn!(
1488-
"could not read schema file '{}', error: {e}",
1489-
entry.path().display()
1490-
);
1491-
}
1492-
}
1493-
}
1494-
}
1495-
Err(e) => {
1496-
warn!(
1497-
"could not read dir '{}', error: {e}",
1498-
schema_path.as_ref().display()
1499-
);
1500-
}
1501-
}
1502-
} else if meta.is_file() {
1503-
match tokio::fs::read_to_string(schema_path).await {
1504-
Ok(s) => {
1505-
contents.push(s);
1506-
// pushed.push(schema_path.clone());
1507-
}
1508-
Err(e) => {
1509-
warn!(
1510-
"could not read schema file '{}', error: {e}",
1511-
schema_path.as_ref().display()
1512-
);
1513-
}
1514-
}
1515-
}
1516-
}
1517-
1518-
Err(e) => {
1519-
warn!(
1520-
"could not read schema file meta '{}', error: {e}",
1521-
schema_path.as_ref().display()
1522-
);
1523-
}
1524-
}
1525-
}
1526-
1527-
Ok(contents)
1528-
}
1529-
15301448
pub fn log_at_pow_10(msg: &str, count: &mut u64) {
15311449
*count += 1;
15321450
if is_pow_10(*count) {

‎crates/corro-client/Cargo.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ license = "MIT"
1010
[dependencies]
1111
bytes = { workspace = true }
1212
corro-api-types = { version = "0.1.0-alpha.1", path = "../corro-api-types" }
13-
corro-agent = { path = "../corro-agent" }
13+
corro-utils = { path = "../corro-utils" }
1414
futures = { workspace = true }
1515
hickory-resolver = { workspace = true }
1616
http = { workspace = true }

‎crates/corro-client/src/lib.rs‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,5 @@
11
pub mod sub;
22

3-
use corro_agent::agent::util::read_files_from_paths;
43
use corro_api_types::{ChangeId, ExecResponse, ExecResult, SqliteValue, Statement};
54
use hickory_resolver::{
65
error::{ResolveError, ResolveErrorKind},
@@ -345,7 +344,7 @@ impl CorrosionApiClient {
345344
&self,
346345
schema_paths: &[P],
347346
) -> Result<Option<ExecResponse>, Error> {
348-
let statements: Vec<Statement> = read_files_from_paths(schema_paths)
347+
let statements: Vec<Statement> = corro_utils::read_files_from_paths(schema_paths)
349348
.await
350349
.map_err(|e| Error::ResponseError(e.to_string()))?
351350
.into_iter()

‎crates/corro-utils/Cargo.toml‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
[package]
2+
name = "corro-utils"
3+
version = "0.1.0"
4+
edition = "2021"
5+
6+
[dependencies]
7+
eyre = { workspace = true }
8+
tokio = { workspace = true }
9+
tracing = { workspace = true }

‎crates/corro-utils/src/lib.rs‎

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,84 @@
1+
use std::path::Path;
2+
3+
use tracing::warn;
4+
5+
pub async fn read_files_from_paths<P: AsRef<Path>>(
6+
schema_paths: &[P],
7+
) -> eyre::Result<Vec<String>> {
8+
let mut contents = vec![];
9+
10+
for schema_path in schema_paths.iter() {
11+
match tokio::fs::metadata(schema_path).await {
12+
Ok(meta) => {
13+
if meta.is_dir() {
14+
match tokio::fs::read_dir(schema_path).await {
15+
Ok(mut dir) => {
16+
let mut entries = vec![];
17+
18+
while let Ok(Some(entry)) = dir.next_entry().await {
19+
entries.push(entry);
20+
}
21+
22+
let mut entries: Vec<_> = entries
23+
.into_iter()
24+
.filter_map(|entry| {
25+
entry.path().extension().and_then(|ext| {
26+
if ext == "sql" {
27+
Some(entry)
28+
} else {
29+
None
30+
}
31+
})
32+
})
33+
.collect();
34+
35+
entries.sort_by_key(|entry| entry.path());
36+
37+
for entry in entries.iter() {
38+
match tokio::fs::read_to_string(entry.path()).await {
39+
Ok(s) => {
40+
contents.push(s);
41+
}
42+
Err(e) => {
43+
warn!(
44+
"could not read schema file '{}', error: {e}",
45+
entry.path().display()
46+
);
47+
}
48+
}
49+
}
50+
}
51+
Err(e) => {
52+
warn!(
53+
"could not read dir '{}', error: {e}",
54+
schema_path.as_ref().display()
55+
);
56+
}
57+
}
58+
} else if meta.is_file() {
59+
match tokio::fs::read_to_string(schema_path).await {
60+
Ok(s) => {
61+
contents.push(s);
62+
// pushed.push(schema_path.clone());
63+
}
64+
Err(e) => {
65+
warn!(
66+
"could not read schema file '{}', error: {e}",
67+
schema_path.as_ref().display()
68+
);
69+
}
70+
}
71+
}
72+
}
73+
74+
Err(e) => {
75+
warn!(
76+
"could not read schema file meta '{}', error: {e}",
77+
schema_path.as_ref().display()
78+
);
79+
}
80+
}
81+
}
82+
83+
Ok(contents)
84+
}

0 commit comments

Comments
 (0)