rt: Arc<Runtime>,
db: DatabaseConnection,
- pub current_profile: Option<profile::Model>,
- pub broadcast: Option<tokio::sync::broadcast::Sender<BroadcastMessage>>
-}
+ pub current_profile: profile::Model,
+ pub broadcast: tokio::sync::broadcast::Sender<BroadcastMessage>,
-type CmdFuture = Pin<Box<dyn Future<Output = ()> + Send>>;
-pub trait Cmd: Send {
- fn run(self: Box<Self>, backend: Arc<Backend>) -> CmdFuture;
+ pub broadcast_notify: Option<Box<dyn Fn() + Send + Sync>>
}
-pub type DynCmd = Box<dyn Cmd>;
+type CmdFuture = Pin<Box<dyn Future<Output = ()> + Send>>;
pub struct BackendHandle {
- pub tx: tokio::sync::mpsc::UnboundedSender<DynCmd>,
+ pub backend: Arc<Backend>,
+ pub tx: tokio::sync::mpsc::UnboundedSender<CmdFuture>,
pub btx: tokio::sync::broadcast::Sender<BroadcastMessage>,
pub brx: tokio::sync::broadcast::Receiver<BroadcastMessage>
}
impl Clone for BackendHandle {
fn clone(&self) -> Self {
BackendHandle {
+ backend: self.backend.clone(),
tx: self.tx.clone(),
btx: self.btx.clone(),
brx: self.btx.subscribe()
impl Backend {
- pub fn new() -> Result<Self, DbErr> {
+ pub fn new(profile: Option<i32>, host: &str, user: &str) -> Result<Self, DbErr> {
let rt = Arc::new(
tokio::runtime::Builder::new_multi_thread()
.enable_all()
Database::connect(connection_str).await
})?;
+ rt.block_on(async {
+ Migrator::up(&db, None).await
+ })?;
+
+ let profile = Backend::load_profile(&rt, &db, profile, host, user)?;
+
+ let (tx, _) = broadcast::channel::<BroadcastMessage>(16);
+
Ok(Self {
rt,
db,
- current_profile: None,
- broadcast: None
+ current_profile: profile,
+ broadcast: tx,
+ broadcast_notify: None
})
}
- pub fn migrate(&self) -> Result<(), DbErr> {
- self.rt.block_on(async {
- Migrator::up(&self.db, None).await
- })?;
-
- Ok(())
- }
-
pub fn load_profile(
- &self,
+ rt: &Arc<Runtime>,
+ db: &DatabaseConnection,
profile_id: Option<i32>,
host: &str,
user: &str,
let host = host.to_string();
let user = user.to_string();
- let profile: profile::Model = self.rt.block_on(async {
+ let profile: profile::Model = rt.block_on(async {
// If a profile id was specified, we are trying to use that,
// but if it doesn't exist, an error is returned
if let Some(id) = profile_id {
if let Some(model) = Profile::find_by_id(id)
- .one(&self.db)
+ .one(db)
.await?
{
return Ok::<profile::Model, DbErr>(model);
if let Some(model) = Profile::find()
.filter(profile::Column::Host.eq(host.clone()))
.filter(profile::Column::User.eq(user.clone()))
- .one(&self.db)
+ .one(db)
.await?
{
return Ok::<profile::Model, DbErr>(model);
..Default::default()
};
- let inserted = active.insert(&self.db).await?;
+ let inserted = active.insert(db).await?;
// create some initial notebooks
let insert_notebooks = collection::ActiveModel {
..Default::default()
};
- insert_notebooks.insert(&self.db).await?;
+ insert_notebooks.insert(db).await?;
let insert_notes = collection::ActiveModel {
profile_id: Set(inserted.id),
..Default::default()
};
- insert_notes.insert(&self.db).await?;
+ insert_notes.insert(db).await?;
Ok::<profile::Model, DbErr>(inserted)
})?;
Ok(profile)
}
- pub fn load_collections(&self, profile_id: i32) -> Result<Vec<collection::Model>, DbErr> {
+ /// Synchronous function for loading the notebook structure, used at startup
+ /// before any window is opened
+ pub fn load_collections(&self) -> Result<Vec<collection::Model>, DbErr> {
self.rt.block_on(async {
Collection::find()
- .filter(collection::Column::ProfileId.eq(profile_id))
+ .filter(collection::Column::ProfileId.eq(self.current_profile.id))
.order_by_asc(collection::Column::Parent)
.all(&self.db).await
})
}
- pub fn start(mut self) -> BackendHandle {
- let (tx, mut rx) = mpsc::unbounded_channel::<DynCmd>();
- let (btx, brx) = broadcast::channel::<BroadcastMessage>(16);
- self.broadcast = Some(btx.clone());
+ /// Starts the backend thread
+ ///
+ /// This function returns a cloneable BackendHandle, that can be used to send
+ /// messages to the backend via the tx Sender.
+ ///
+ /// The handle also contains a broadcast sender/receiver (btx, brx).
+ pub fn start(self) -> BackendHandle {
+ let (tx, mut rx) = mpsc::unbounded_channel::<CmdFuture>();
+ let broadcast_tx = self.broadcast.clone();
+ let broadcast_rx = broadcast_tx.subscribe();
let backend = Arc::new(self);
+ let backend_clone = backend.clone();
let rt = backend.rt.clone();
std::thread::spawn(move || {
rt.block_on(async move {
while let Some(cmd) = rx.recv().await {
- let backend = backend.clone();
-
tokio::spawn(async move {
- cmd.run(backend).await;
+ cmd.await
});
}
});
});
BackendHandle {
+ backend: backend_clone,
tx: tx,
- btx: btx,
- brx: brx,
+ btx: broadcast_tx,
+ brx: broadcast_rx,
}
}
}
impl BackendHandle {
+ /// Reloads the notebook structure/collections. On success, it sends a NotebookStructureUpdate
+ /// to the broadcast channel.
pub fn reload_collections(&self) {
- let cmd: DynCmd = Box::new(ReloadCollectionsCmd);
- let _ = self.tx.send(cmd);
- }
-}
-
-
-struct ReloadCollectionsCmd;
-
-impl Cmd for ReloadCollectionsCmd {
- fn run(self: Box<Self>, backend: Arc<Backend>) -> CmdFuture {
- Box::pin(async move {
- println!("reloading collections...");
-
- let profile_id = backend.current_profile.as_ref().unwrap().id;
+ let backend = self.backend.clone();
+ let cmd = Box::pin(async move {
+ let profile_id = backend.current_profile.id;
let result = Collection::find()
.filter(collection::Column::ProfileId.eq(profile_id))
if let Ok(collection) = result {
let nodes = create_notebook_hierarchy(collection);
- if let Some(tx) = backend.broadcast.as_ref() {
- let _ = tx.send(BroadcastMessage::NotebookStructureUpdate(nodes));
- ui::broadcast_action("message");
+ let _ = backend.broadcast.send(BroadcastMessage::NotebookStructureUpdate(nodes));
+
+ if let Some(notify) = backend.broadcast_notify.as_ref() {
+ notify();
}
}
- })
+ });
+ let _ = self.tx.send(cmd);
}
-}
\ No newline at end of file
+}
return;
}
};
- let profile_id = backend.current_profile.as_ref().unwrap().id;
- let notebooks = match backend.load_collections(profile_id) {
+ let notebooks = match backend.load_collections() {
Ok(notebooks) => create_notebook_hierarchy(notebooks),
Err(e) => {
let msg = format!("Error loading collections: {:?}", e);
/// - Connecting to the database
/// - Migrating the database
/// - Getting the user profile / settings
+/// - Setup UI broadcasting
///
/// If no profile exists yet, a new profile is created
fn init_backend() -> Result<Backend, ErrMsg> {
- let mut backend = match Backend::new() {
+ let host = hostname();
+ let user = username();
+
+ let mut backend = match Backend::new(None, host.as_str(), user.as_str()) {
Ok(backend) => backend,
Err(e) => return Err( ErrMsg {
title: "Backend initialization failed".to_string(),
})
};
- match backend.migrate() {
- Ok(_) => (),
- Err(e) => return Err(ErrMsg {
- title: "Database migration failed".to_string(),
- message: e.to_string()
- })
- };
-
- let host = hostname();
- let user = username();
-
- let profile = match backend.load_profile(None, host.as_str(), user.as_str()) {
- Ok(profile) => profile,
- Err(e) => return Err(ErrMsg {
- title: "Cannot load user settings".to_string(),
- message: e.to_string()
- })
- };
- backend.current_profile = Some(profile);
+ backend.broadcast_notify = Some(Box::new(|| ui::broadcast_action("message") ));
Ok(backend)
}
https://uap-core.de/git / note.git / commitdiff