[proxmox-backup.git] / src / api2 / admin / datastore / backup / environment.rs

use failure::*;
use std::sync::{Arc, Mutex};
use std::collections::HashMap;

use serde_json::Value;

use crate::api_schema::router::{RpcEnvironment, RpcEnvironmentType};
use crate::server::WorkerTask;
use crate::backup::*;
use crate::server::formatter::*;
use hyper::{Body, Response};

struct SharedBackupState {
    uid_counter: usize,
    dynamic_writers: HashMap<usize, (u64 /* offset */, DynamicIndexWriter)>,
}

/// `RpcEnvironmet` implementation for backup service
#[derive(Clone)]
pub struct BackupEnvironment {
    env_type: RpcEnvironmentType,
    result_attributes: HashMap<String, Value>,
    user: String,
    pub formatter: &'static OutputFormatter,
    pub worker: Arc<WorkerTask>,
    pub datastore: Arc<DataStore>,
    pub backup_dir: BackupDir,
    pub last_backup: Option<BackupInfo>,
    state: Arc<Mutex<SharedBackupState>>
}

impl BackupEnvironment {
    pub fn new(
        env_type: RpcEnvironmentType,
        user: String,
        worker: Arc<WorkerTask>,
        datastore: Arc<DataStore>,
        backup_dir: BackupDir,
    ) -> Self {

        let state = SharedBackupState {
            uid_counter: 0,
            dynamic_writers: HashMap::new(),
        };

        Self {
            result_attributes: HashMap::new(),
            env_type,
            user,
            worker,
            datastore,
            formatter: &JSON_FORMATTER,
            backup_dir,
            last_backup: None,
            state: Arc::new(Mutex::new(state)),
        }
    }

    /// Get an unique integer ID
    pub fn next_uid(&self) -> usize {
        let mut state = self.state.lock().unwrap();
        state.uid_counter += 1;
        state.uid_counter
    }

    /// Store the writer with an unique ID
    pub fn register_dynamic_writer(&self, writer: DynamicIndexWriter) -> usize {
       let mut state = self.state.lock().unwrap();
        state.uid_counter += 1;
        let uid = state.uid_counter;

        state.dynamic_writers.insert(uid, (0, writer));
        uid
    }

    /// Append chunk to dynamic writer
    pub fn dynamic_writer_append_chunk(&self, wid: usize, size: u64, digest: &[u8; 32]) -> Result<(), Error> {
        let mut state = self.state.lock().unwrap();

        let mut data = match state.dynamic_writers.get_mut(&wid) {
            Some(data) => data,
            None => bail!("dynamic writer '{}' not registered", wid),
        };

        data.0 += size;

        data.1.add_chunk(data.0, digest)?;

        Ok(())
    }

    pub fn log<S: AsRef<str>>(&self, msg: S) {
        self.worker.log(msg);
    }

    pub fn format_response(&self, result: Result<Value, Error>) -> Response<Body> {
        match result {
            Ok(data) => (self.formatter.format_data)(data, self),
            Err(err) => (self.formatter.format_error)(err),
        }
    }
}

impl RpcEnvironment for BackupEnvironment {

    fn set_result_attrib(&mut self, name: &str, value: Value) {
        self.result_attributes.insert(name.into(), value);
    }

    fn get_result_attrib(&self, name: &str) -> Option<&Value> {
        self.result_attributes.get(name)
    }

    fn env_type(&self) -> RpcEnvironmentType {
        self.env_type
    }

    fn set_user(&mut self, _user: Option<String>) {
        panic!("unable to change user");
    }

    fn get_user(&self) -> Option<String> {
        Some(self.user.clone())
    }
}

impl AsRef<BackupEnvironment> for RpcEnvironment {
    fn as_ref(&self) -> &BackupEnvironment {
        self.as_any().downcast_ref::<BackupEnvironment>().unwrap()
    }
}
impl AsRef<BackupEnvironment> for Box<RpcEnvironment> {
    fn as_ref(&self) -> &BackupEnvironment {
        self.as_any().downcast_ref::<BackupEnvironment>().unwrap()
    }
}
Commit	Line	Data
b4b63e52	1	use failure::*;
f9578f3c	2	use std::sync::{Arc, Mutex};
d95ced64 DM	3	use std::collections::HashMap;
	4
	5	use serde_json::Value;
	6
	7	use crate::api_schema::router::{RpcEnvironment, RpcEnvironmentType};
	8	use crate::server::WorkerTask;
21ee7912	9	use crate::backup::*;
b4b63e52 DM	10	use crate::server::formatter::*;
b4b63e52 DM	11	use hyper::{Body, Response};
d95ced64	12
f9578f3c DM	13	struct SharedBackupState {
	14	uid_counter: usize,
	15	dynamic_writers: HashMap<usize, (u64 /* offset */, DynamicIndexWriter)>,
	16	}
	17
d95ced64 DM	18	/// `RpcEnvironmet` implementation for backup service
	19	#[derive(Clone)]
	20	pub struct BackupEnvironment {
	21	env_type: RpcEnvironmentType,
	22	result_attributes: HashMap<String, Value>,
	23	user: String,
b4b63e52	24	pub formatter: &'static OutputFormatter,
21ee7912 DM	25	pub worker: Arc<WorkerTask>,
21ee7912 DM	26	pub datastore: Arc<DataStore>,
f9578f3c	27	pub backup_dir: BackupDir,
b02a52e3	28	pub last_backup: Option<BackupInfo>,
f9578f3c	29	state: Arc<Mutex<SharedBackupState>>
d95ced64 DM	30	}
	31
	32	impl BackupEnvironment {
f9578f3c DM	33	pub fn new(
	34	env_type: RpcEnvironmentType,
	35	user: String,
	36	worker: Arc<WorkerTask>,
	37	datastore: Arc<DataStore>,
	38	backup_dir: BackupDir,
f9578f3c DM	39	) -> Self {
	40
	41	let state = SharedBackupState {
	42	uid_counter: 0,
	43	dynamic_writers: HashMap::new(),
	44	};
	45
d95ced64 DM	46	Self {
	47	result_attributes: HashMap::new(),
	48	env_type,
	49	user,
	50	worker,
21ee7912	51	datastore,
b4b63e52	52	formatter: &JSON_FORMATTER,
f9578f3c	53	backup_dir,
b02a52e3	54	last_backup: None,
f9578f3c	55	state: Arc::new(Mutex::new(state)),
d95ced64 DM	56	}
	57	}
	58
f9578f3c DM	59	/// Get an unique integer ID
	60	pub fn next_uid(&self) -> usize {
	61	let mut state = self.state.lock().unwrap();
	62	state.uid_counter += 1;
	63	state.uid_counter
	64	}
	65
	66	/// Store the writer with an unique ID
	67	pub fn register_dynamic_writer(&self, writer: DynamicIndexWriter) -> usize {
	68	let mut state = self.state.lock().unwrap();
	69	state.uid_counter += 1;
	70	let uid = state.uid_counter;
	71
	72	state.dynamic_writers.insert(uid, (0, writer));
	73	uid
	74	}
	75
	76	/// Append chunk to dynamic writer
	77	pub fn dynamic_writer_append_chunk(&self, wid: usize, size: u64, digest: &[u8; 32]) -> Result<(), Error> {
	78	let mut state = self.state.lock().unwrap();
	79
	80	let mut data = match state.dynamic_writers.get_mut(&wid) {
	81	Some(data) => data,
	82	None => bail!("dynamic writer '{}' not registered", wid),
	83	};
	84
	85	data.0 += size;
	86
	87	data.1.add_chunk(data.0, digest)?;
	88
	89	Ok(())
	90	}
	91
d95ced64 DM	92	pub fn log<S: AsRef<str>>(&self, msg: S) {
	93	self.worker.log(msg);
	94	}
b4b63e52 DM	95
	96	pub fn format_response(&self, result: Result<Value, Error>) -> Response<Body> {
	97	match result {
	98	Ok(data) => (self.formatter.format_data)(data, self),
	99	Err(err) => (self.formatter.format_error)(err),
	100	}
	101	}
d95ced64 DM	102	}
	103
	104	impl RpcEnvironment for BackupEnvironment {
	105
	106	fn set_result_attrib(&mut self, name: &str, value: Value) {
	107	self.result_attributes.insert(name.into(), value);
	108	}
	109
	110	fn get_result_attrib(&self, name: &str) -> Option<&Value> {
	111	self.result_attributes.get(name)
	112	}
	113
	114	fn env_type(&self) -> RpcEnvironmentType {
	115	self.env_type
	116	}
	117
	118	fn set_user(&mut self, _user: Option<String>) {
	119	panic!("unable to change user");
	120	}
	121
	122	fn get_user(&self) -> Option<String> {
	123	Some(self.user.clone())
	124	}
	125	}
21ee7912 DM	126
	127	impl AsRef<BackupEnvironment> for RpcEnvironment {
	128	fn as_ref(&self) -> &BackupEnvironment {
	129	self.as_any().downcast_ref::<BackupEnvironment>().unwrap()
	130	}
	131	}
b4b63e52 DM	132	impl AsRef<BackupEnvironment> for Box<RpcEnvironment> {
	133	fn as_ref(&self) -> &BackupEnvironment {
	134	self.as_any().downcast_ref::<BackupEnvironment>().unwrap()
	135	}
	136	}