1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
|
use std::fmt;
use crate::sm_c8y_mapper::mapper::CumulocitySoftwareManagementMapper;
use crate::{
az_mapper::AzureMapper, c8y_mapper::CumulocityMapper, collectd_mapper::mapper::CollectdMapper,
component::TEdgeComponent, error::*,
};
use flockfile::check_another_instance_is_not_running;
use structopt::*;
use tedge_config::*;
use tedge_utils::paths::home_dir;
mod az_converter;
mod az_mapper;
mod c8y_converter;
mod c8y_fragments;
mod c8y_mapper;
mod collectd_mapper;
mod component;
mod converter;
mod error;
mod mapper;
mod operations;
mod size_threshold;
mod sm_c8y_mapper;
#[cfg(test)]
mod tests;
fn lookup_component(component_name: &MapperName) -> Box<dyn TEdgeComponent> {
match component_name {
MapperName::Az => Box::new(AzureMapper::new()),
MapperName::Collectd => Box::new(CollectdMapper::new()),
MapperName::C8y => Box::new(CumulocityMapper::new()),
MapperName::SmC8y => Box::new(CumulocitySoftwareManagementMapper::new()),
}
}
#[derive(Debug, StructOpt)]
#[structopt(
name = clap::crate_name!(),
version = clap::crate_version!(),
about = clap::crate_description!()
)]
pub struct MapperOpt {
#[structopt(subcommand)]
pub name: MapperName,
/// Turn-on the debug log level.
///
/// If off only reports ERROR, WARN, and INFO
/// If on also reports DEBUG and TRACE
#[structopt(long)]
pub debug: bool,
/// Start the mapper with clean session off, subscribe to the topics, so that no messages are lost
#[structopt(short, long)]
pub init: bool,
/// Start the agent with clean session on, drop the previous session and subscriptions
///
/// WARNING: All pending messages will be lost.
#[structopt(short, long)]
pub clear: bool,
}
#[derive(Debug, StructOpt)]
pub enum MapperName {
Az,
C8y,
Collectd,
SmC8y,
}
impl fmt::Display for MapperName {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
match self {
MapperName::Az => write!(f, "{}", "tedge-mapper-az"),
MapperName::C8y => write!(f, "{}", "tedge-mapper-c8y"),
MapperName::Collectd => write!(f, "{}", "tedge-mapper-collectd"),
MapperName::SmC8y => write!(f, "{}", "sm-c8y-mapper"),
}
}
}
#[tokio::main]
async fn main() -> anyhow::Result<()> {
let mapper = MapperOpt::from_args();
tedge_utils::logging::initialise_tracing_subscriber(mapper.debug);
let component = lookup_component(&mapper.name);
let config = tedge_config()?;
// Run only one instance of a mapper
let _flock = check_another_instance_is_not_running(&mapper.name.to_string())?;
if mapper.init {
let mut mapper = CumulocitySoftwareManagementMapper::new();
mapper.init_session().await
} else if mapper.clear {
let mut mapper = CumulocitySoftwareManagementMapper::new();
mapper.clear_session().await
} else {
component.start(config).await
}
}
fn tedge_config() -> anyhow::Result<TEdgeConfig> {
let config_repository = config_repository()?;
Ok(config_repository.load()?)
}
fn config_repository() -> Result<TEdgeConfigRepository, MapperError> {
let tedge_config_location = if tedge_users::UserManager::running_as_root()
|| tedge_users::UserManager::running_as("tedge-mapper")
{
tedge_config::TEdgeConfigLocation::from_default_system_location()
} else {
tedge_config::TEdgeConfigLocation::from_users_home_location(
home_dir().ok_or(MapperError::HomeDirNotFound)?,
)
};
let config_repository = tedge_config::TEdgeConfigRepository::new(tedge_config_location);
Ok(config_repository)
}
|
