|
@@ -294,16 +294,21 @@ struct Peer {
|
|
|
}
|
|
|
|
|
|
#[derive(Debug, Deserialize)]
|
|
|
-struct Config {
|
|
|
- user: Peer,
|
|
|
+struct ConversationConfig {
|
|
|
group: String,
|
|
|
recipients: Vec<Peer>,
|
|
|
- socks: Option<String>,
|
|
|
bootstrap: f64,
|
|
|
retry: f64,
|
|
|
distributions: ConfigDistributions,
|
|
|
}
|
|
|
|
|
|
+#[derive(Debug, Deserialize)]
|
|
|
+struct Config {
|
|
|
+ user: Peer,
|
|
|
+ socks: Option<String>,
|
|
|
+ conversations: Vec<ConversationConfig>,
|
|
|
+}
|
|
|
+
|
|
|
#[tokio::main]
|
|
|
async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
|
|
let mut args = std::env::args();
|
|
@@ -317,77 +322,71 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
|
|
retry: f64,
|
|
|
}
|
|
|
|
|
|
- // a map from `user` to {a map from `recipient` to things the (user, recipient) reader/writer threads will need}
|
|
|
- let mut recipient_maps = HashMap::<String, HashMap<String, ForIoThreads>>::new();
|
|
|
-
|
|
|
let mut handles = vec![];
|
|
|
|
|
|
for config_file in args.flat_map(|a| glob::glob(a.as_str()).unwrap()) {
|
|
|
let toml_s = std::fs::read_to_string(config_file?)?;
|
|
|
let config: Config = toml::from_str(&toml_s)?;
|
|
|
|
|
|
- let (reader_to_state, state_from_reader) = mpsc::unbounded_channel();
|
|
|
-
|
|
|
- if !recipient_maps.contains_key(&config.user.address) {
|
|
|
- let map = HashMap::<String, ForIoThreads>::new();
|
|
|
- recipient_maps.insert(config.user.address.clone(), map);
|
|
|
- }
|
|
|
-
|
|
|
- let user_recipient_map = recipient_maps.get_mut(&config.user.address).unwrap();
|
|
|
-
|
|
|
- let mut conversation_recipient_map =
|
|
|
- HashMap::<String, StateToWriter<MessageHolder>>::with_capacity(config.recipients.len());
|
|
|
-
|
|
|
- for recipient in config.recipients.iter() {
|
|
|
- let state_to_writer = if !user_recipient_map.contains_key(&recipient.name) {
|
|
|
- let (state_to_writer, writer_from_state) = mpsc::unbounded_channel();
|
|
|
- let mut reader_to_states = HashMap::new();
|
|
|
- reader_to_states.insert(config.group.clone(), reader_to_state.clone());
|
|
|
- let str_params = SocksParams {
|
|
|
- socks: config.socks.clone(),
|
|
|
- target: recipient.address.clone(),
|
|
|
- user: config.user.name.clone(),
|
|
|
- recipient: recipient.name.clone(),
|
|
|
- };
|
|
|
- let for_io = ForIoThreads {
|
|
|
- state_to_writer: state_to_writer.clone(),
|
|
|
- writer_from_state,
|
|
|
- reader_to_states,
|
|
|
- str_params,
|
|
|
- retry: config.retry,
|
|
|
+ // map from `recipient` to things the (user, recipient) reader/writer threads will need
|
|
|
+ let mut recipient_map = HashMap::<String, ForIoThreads>::new();
|
|
|
+ for conversation in config.conversations.into_iter() {
|
|
|
+ let (reader_to_state, state_from_reader) = mpsc::unbounded_channel();
|
|
|
+
|
|
|
+ let mut conversation_recipient_map =
|
|
|
+ HashMap::<String, StateToWriter<MessageHolder>>::with_capacity(
|
|
|
+ conversation.recipients.len(),
|
|
|
+ );
|
|
|
+
|
|
|
+ for recipient in conversation.recipients.iter() {
|
|
|
+ let state_to_writer = if !recipient_map.contains_key(&recipient.name) {
|
|
|
+ let (state_to_writer, writer_from_state) = mpsc::unbounded_channel();
|
|
|
+ let mut reader_to_states = HashMap::new();
|
|
|
+ reader_to_states.insert(conversation.group.clone(), reader_to_state.clone());
|
|
|
+ let str_params = SocksParams {
|
|
|
+ socks: config.socks.clone(),
|
|
|
+ target: recipient.address.clone(),
|
|
|
+ user: config.user.name.clone(),
|
|
|
+ recipient: recipient.name.clone(),
|
|
|
+ };
|
|
|
+ let for_io = ForIoThreads {
|
|
|
+ state_to_writer: state_to_writer.clone(),
|
|
|
+ writer_from_state,
|
|
|
+ reader_to_states,
|
|
|
+ str_params,
|
|
|
+ retry: conversation.retry,
|
|
|
+ };
|
|
|
+ recipient_map.insert(recipient.name.clone(), for_io);
|
|
|
+ state_to_writer
|
|
|
+ } else {
|
|
|
+ let for_io = recipient_map.get_mut(&recipient.name).unwrap();
|
|
|
+ if !for_io.reader_to_states.contains_key(&conversation.group) {
|
|
|
+ for_io
|
|
|
+ .reader_to_states
|
|
|
+ .insert(conversation.group.clone(), reader_to_state.clone());
|
|
|
+ }
|
|
|
+ for_io.state_to_writer.clone()
|
|
|
};
|
|
|
- user_recipient_map.insert(recipient.name.clone(), for_io);
|
|
|
- state_to_writer
|
|
|
- } else {
|
|
|
- let for_io = user_recipient_map.get_mut(&recipient.name).unwrap();
|
|
|
- if !for_io.reader_to_states.contains_key(&config.group) {
|
|
|
- for_io
|
|
|
- .reader_to_states
|
|
|
- .insert(config.group.clone(), reader_to_state.clone());
|
|
|
- }
|
|
|
- for_io.state_to_writer.clone()
|
|
|
- };
|
|
|
- conversation_recipient_map.insert(
|
|
|
- recipient.name.clone(),
|
|
|
- StateToWriter {
|
|
|
- channel: state_to_writer,
|
|
|
- },
|
|
|
- );
|
|
|
- }
|
|
|
+ conversation_recipient_map.insert(
|
|
|
+ recipient.name.clone(),
|
|
|
+ StateToWriter {
|
|
|
+ channel: state_to_writer,
|
|
|
+ },
|
|
|
+ );
|
|
|
+ }
|
|
|
|
|
|
- let distributions: Distributions = config.distributions.try_into()?;
|
|
|
+ let distributions: Distributions = conversation.distributions.try_into()?;
|
|
|
|
|
|
- tokio::spawn(manage_conversation(
|
|
|
- config.user.name,
|
|
|
- config.group,
|
|
|
- distributions,
|
|
|
- config.bootstrap,
|
|
|
- state_from_reader,
|
|
|
- conversation_recipient_map,
|
|
|
- ));
|
|
|
- }
|
|
|
+ tokio::spawn(manage_conversation(
|
|
|
+ config.user.name.clone(),
|
|
|
+ conversation.group,
|
|
|
+ distributions,
|
|
|
+ conversation.bootstrap,
|
|
|
+ state_from_reader,
|
|
|
+ conversation_recipient_map,
|
|
|
+ ));
|
|
|
+ }
|
|
|
|
|
|
- for (address, mut recipient_map) in recipient_maps.drain() {
|
|
|
let mut name_to_io_threads: HashMap<String, (ReadSocketUpdaterIn, WriteSocketUpdaterIn)> =
|
|
|
HashMap::new();
|
|
|
|
|
@@ -415,7 +414,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
|
|
}
|
|
|
|
|
|
let handle: task::JoinHandle<Result<(), FatalError>> =
|
|
|
- tokio::spawn(listener(address.clone(), name_to_io_threads));
|
|
|
+ tokio::spawn(listener(config.user.address, name_to_io_threads));
|
|
|
handles.push(handle);
|
|
|
}
|
|
|
|