commit 5182bd54a132c85170cca97c816724c1ca5305f2
parent 35ae8f48d6a90e84b47bde92e78d40fde2aee61c
Author: MTRNord <mtrnord1@gmail.com>
Date: Thu, 13 Apr 2023 22:06:43 +0200
fix: Fix multiple smaller bugs
Diffstat:
7 files changed, 82 insertions(+), 17 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
@@ -741,6 +741,28 @@ dependencies = [
]
[[package]]
+name = "dbus"
+version = "0.9.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1bb21987b9fb1613058ba3843121dd18b163b254d8a6e797e144cbac14d96d1b"
+dependencies = [
+ "futures-channel",
+ "futures-util",
+ "libc",
+ "libdbus-sys",
+ "winapi",
+]
+
+[[package]]
+name = "dbus-crossroads"
+version = "0.5.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3a4c83437187544ba5142427746835061b330446ca8902eabd70e4afb8f76de0"
+dependencies = [
+ "dbus",
+]
+
+[[package]]
name = "der"
version = "0.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1591,6 +1613,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "99227334921fae1a979cf0bfdfcc6b3e5ce376ef57e16fb6fb3ea2ed6095f80c"
[[package]]
+name = "libdbus-sys"
+version = "0.2.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9f8d7ae751e1cb825c840ae5e682f59b098cdfd213c350ac268b61449a5f58a0"
+dependencies = [
+ "pkg-config",
+]
+
+[[package]]
name = "libm"
version = "0.2.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1658,6 +1689,8 @@ dependencies = [
"async-channel",
"cfg-if",
"color-eyre",
+ "dbus",
+ "dbus-crossroads",
"futures",
"kdl",
"lazy_static",
@@ -2252,6 +2285,12 @@ dependencies = [
]
[[package]]
+name = "pkg-config"
+version = "0.3.26"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6ac9a59f73473f1b8d852421e59e64809f025994837ef743615c6d0c5b305160"
+
+[[package]]
name = "poly1305"
version = "0.7.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -3669,6 +3708,7 @@ dependencies = [
"serde",
"serde_json",
"tokio",
+ "tracing",
]
[[package]]
diff --git a/crates/matrix-indexer/src/config.rs b/crates/matrix-indexer/src/config.rs
@@ -221,9 +221,7 @@ pub fn write_access_token(access_token: String, device_id: String) -> color_eyre
matrix_nodes.push(access_token_node);
matrix_nodes.push(device_id_node);
- matrix_nodes.retain(|node| {
- node.name().value() != "username" && node.name().value() != "password"
- });
+ matrix_nodes.retain(|node| node.name().value() != "password");
matrix_nodes.sort_by(sort_by_name);
}
diff --git a/crates/matrix-indexer/src/indradb_utils.rs b/crates/matrix-indexer/src/indradb_utils.rs
@@ -55,15 +55,15 @@ impl BulkInserter {
Ok(())
}
- pub async fn flush(mut self) -> Result<()> {
+ pub async fn flush(&mut self) -> Result<()> {
if !self.buf.is_empty() {
- self.requests.send(self.buf).await?;
+ self.requests.send(self.buf.clone()).await?;
}
- self.requests.close();
- for worker in self.workers {
- worker.await??;
- }
- self.client.sync().await?;
+ //self.requests.close();
+ // for worker in &self.workers {
+ // worker.await??;
+ // }
+ self.sync().await?;
Ok(())
}
diff --git a/crates/matrix-indexer/src/main.rs b/crates/matrix-indexer/src/main.rs
@@ -30,6 +30,7 @@ use color_eyre::{eyre::bail, Result};
use config::{load, write_access_token};
use matrix::IndexerBot;
+use tracing::info;
mod config;
mod indradb_utils;
@@ -37,6 +38,7 @@ mod matrix;
#[tokio::main]
async fn main() -> Result<()> {
+ std::env::set_var("RUST_LOG", "matrix_sdk=info,matrix-indexer=debug");
tracing_subscriber::fmt::init();
color_eyre::install()?;
@@ -83,6 +85,7 @@ async fn main() -> Result<()> {
.await?
}
};
+ info!("Starting to process");
bot.start_processing().await?;
Ok(())
diff --git a/crates/matrix-indexer/src/matrix.rs b/crates/matrix-indexer/src/matrix.rs
@@ -165,8 +165,11 @@ impl IndexerBot {
pub async fn start_processing(&mut self) -> Result<()> {
let mut inserter = BulkInserter::new(self.indexer_client.clone());
+ info!("Got bulk inserter. Starting sync");
+
let mut sync_stream = Box::pin(self.client.sync_stream(SyncSettings::default()).await);
+ info!("Sync obtained. Starting to process sync stream");
while let Some(Ok(response)) = sync_stream.next().await {
for (ref room_id, room) in response.rooms.join {
for e in &room.timeline.events {
@@ -315,7 +318,7 @@ impl IndexerBot {
))
.await?;
}
- inserter.sync().await?;
+ inserter.flush().await?;
}
Ok(())
}
diff --git a/crates/utils/Cargo.toml b/crates/utils/Cargo.toml
@@ -12,3 +12,4 @@ tokio = { version = "1.26.0", features = ["time"] }
dirs = "5.0.0"
serde_json = "1.0.94"
serde = { version = "1.0.158", features = ["derive"] }
+tracing = "0.1.37"
diff --git a/crates/utils/src/lib.rs b/crates/utils/src/lib.rs
@@ -20,16 +20,17 @@
clippy::panic_in_result_fn,
clippy::clone_on_ref_ptr
)]
-#![allow(clippy::missing_panics_doc)]
+#![allow(clippy::missing_panics_doc, clippy::panic_in_result_fn)]
// I am lazy. Dont blame me!
#![allow(missing_docs)]
-use std::{collections::BTreeSet, fs::OpenOptions, path::PathBuf};
+use std::{collections::BTreeSet, fs::OpenOptions, io::Read, path::PathBuf};
pub use indradb;
pub use indradb_proto;
use serde::{Deserialize, Serialize};
use tokio::time::{sleep, Duration};
+use tracing::instrument;
pub async fn get_client(
endpoint: String,
@@ -85,15 +86,34 @@ pub fn get_identifier() -> Result<BTreeSet<String>, std::io::Error> {
Ok(identifier_config.identifiers)
}
+#[instrument]
pub fn add_identifiers(identifiers: &mut BTreeSet<String>) -> Result<(), std::io::Error> {
- let identifier_file = OpenOptions::new()
+ let config_path = get_identifier_config();
+ let prefix = config_path.parent().expect("No parent folder found");
+ std::fs::create_dir_all(prefix)?;
+
+ let mut identifier_file = OpenOptions::new()
.read(true)
.write(true)
+ .truncate(true)
.create(true)
- .open(get_identifier_config())?;
- let mut identifier_config: IdentifierConfig = serde_json::from_reader(&identifier_file)?;
+ .open(config_path)?;
+ let mut s = String::new();
+ identifier_file
+ .read_to_string(&mut s)
+ .expect("Unable to read file");
+
+ let mut identifier_config: IdentifierConfig;
+ if s.is_empty() {
+ identifier_config = IdentifierConfig {
+ identifiers: BTreeSet::new(),
+ };
+ } else {
+ identifier_config = serde_json::from_str(&s)?;
+ }
identifier_config.identifiers.append(identifiers);
- serde_json::to_writer_pretty(identifier_file, &identifier_config)?;
+ serde_json::to_writer_pretty(&identifier_file, &identifier_config)?;
+ identifier_file.sync_all()?;
Ok(())
}
