|
|
@ -24,14 +24,13 @@
|
|
|
|
*
|
|
|
|
*
|
|
|
|
*/
|
|
|
|
*/
|
|
|
|
|
|
|
|
|
|
|
|
use crate::config::BGSave;
|
|
|
|
use crate::config::ConfigurationSet;
|
|
|
|
use crate::config::SnapshotConfig;
|
|
|
|
use crate::config::SnapshotConfig;
|
|
|
|
use crate::config::SnapshotPref;
|
|
|
|
use crate::config::SnapshotPref;
|
|
|
|
use crate::corestore::Corestore;
|
|
|
|
use crate::corestore::Corestore;
|
|
|
|
use crate::dbnet::{self, Terminator};
|
|
|
|
use crate::dbnet::{self, Terminator};
|
|
|
|
use crate::services;
|
|
|
|
use crate::services;
|
|
|
|
use crate::storage::sengine::SnapshotEngine;
|
|
|
|
use crate::storage::sengine::SnapshotEngine;
|
|
|
|
use crate::PortConfig;
|
|
|
|
|
|
|
|
use std::sync::Arc;
|
|
|
|
use std::sync::Arc;
|
|
|
|
use tokio::sync::broadcast;
|
|
|
|
use tokio::sync::broadcast;
|
|
|
|
|
|
|
|
|
|
|
@ -65,16 +64,19 @@ impl Future for UnixTerminationSignal {
|
|
|
|
|
|
|
|
|
|
|
|
/// Start the server waiting for incoming connections or a termsig
|
|
|
|
/// Start the server waiting for incoming connections or a termsig
|
|
|
|
pub async fn run(
|
|
|
|
pub async fn run(
|
|
|
|
ports: PortConfig,
|
|
|
|
ConfigurationSet {
|
|
|
|
bgsave_cfg: BGSave,
|
|
|
|
ports,
|
|
|
|
snapshot_cfg: SnapshotConfig,
|
|
|
|
bgsave,
|
|
|
|
_restore_filepath: Option<String>,
|
|
|
|
snapshot,
|
|
|
|
maxcon: usize,
|
|
|
|
maxcon,
|
|
|
|
|
|
|
|
..
|
|
|
|
|
|
|
|
}: ConfigurationSet,
|
|
|
|
|
|
|
|
restore_filepath: Option<String>,
|
|
|
|
) -> Result<Corestore, String> {
|
|
|
|
) -> Result<Corestore, String> {
|
|
|
|
// Intialize the broadcast channel
|
|
|
|
// Intialize the broadcast channel
|
|
|
|
let (signal, _) = broadcast::channel(1);
|
|
|
|
let (signal, _) = broadcast::channel(1);
|
|
|
|
let engine;
|
|
|
|
let engine;
|
|
|
|
match &snapshot_cfg {
|
|
|
|
match &snapshot {
|
|
|
|
SnapshotConfig::Enabled(SnapshotPref { atmost, .. }) => {
|
|
|
|
SnapshotConfig::Enabled(SnapshotPref { atmost, .. }) => {
|
|
|
|
engine = SnapshotEngine::new(*atmost);
|
|
|
|
engine = SnapshotEngine::new(*atmost);
|
|
|
|
engine
|
|
|
|
engine
|
|
|
@ -86,19 +88,22 @@ pub async fn run(
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
let engine = Arc::new(engine);
|
|
|
|
let engine = Arc::new(engine);
|
|
|
|
|
|
|
|
// restore data
|
|
|
|
|
|
|
|
services::restore_data(restore_filepath)
|
|
|
|
|
|
|
|
.map_err(|e| format!("Failed to restore data from backup with error: {}", e))?;
|
|
|
|
let db = Corestore::init_with_snapcfg(engine.clone())
|
|
|
|
let db = Corestore::init_with_snapcfg(engine.clone())
|
|
|
|
.map_err(|e| format!("Error while initializing database: {}", e))?;
|
|
|
|
.map_err(|e| format!("Error while initializing database: {}", e))?;
|
|
|
|
|
|
|
|
|
|
|
|
// initialize the background services
|
|
|
|
// initialize the background services
|
|
|
|
let bgsave_handle = tokio::spawn(services::bgsave::bgsave_scheduler(
|
|
|
|
let bgsave_handle = tokio::spawn(services::bgsave::bgsave_scheduler(
|
|
|
|
db.clone(),
|
|
|
|
db.clone(),
|
|
|
|
bgsave_cfg,
|
|
|
|
bgsave,
|
|
|
|
Terminator::new(signal.subscribe()),
|
|
|
|
Terminator::new(signal.subscribe()),
|
|
|
|
));
|
|
|
|
));
|
|
|
|
let snapshot_handle = tokio::spawn(services::snapshot::snapshot_service(
|
|
|
|
let snapshot_handle = tokio::spawn(services::snapshot::snapshot_service(
|
|
|
|
engine,
|
|
|
|
engine,
|
|
|
|
db.clone(),
|
|
|
|
db.clone(),
|
|
|
|
snapshot_cfg,
|
|
|
|
snapshot,
|
|
|
|
Terminator::new(signal.subscribe()),
|
|
|
|
Terminator::new(signal.subscribe()),
|
|
|
|
));
|
|
|
|
));
|
|
|
|
|
|
|
|
|
|
|
|