use std::collections::*; use std::io; use std::path::PathBuf; use std::sync::Arc; use std::time; use futures::compat::Stream01CompatExt; use futures::prelude::*; use inotify::ffi::*; use matrix_sdk::identifiers::RoomId; use tokio::fs as tokiofs; use tokio::io::{AsyncRead, AsyncReadExt}; use tokio::sync::mpsc; use tokio_inotify; use crate::client::{self, MatrixClient}; use crate::config::{self, *}; #[derive(Debug)] pub enum Error { FileFormatMismatch, UnspecifiedClient(String), Io(io::Error), Matrix(matrix_sdk::Error), } impl From for Error { fn from(f: io::Error) -> Self { Self::Io(f) } } impl From for Error { fn from(f: matrix_sdk::Error) -> Self { Self::Matrix(f) } } #[derive(Clone)] struct SpoolAction { client: Arc, room: RoomId, delay_secs: u32, watch_path: PathBuf, } pub async fn start_spoolers( conf: Config, client_chans: HashMap>, ) -> Result<(), Error> { for sd in conf.spool_dirs { let chan = client_chans .get(&sd.sender_acct_label) .ok_or_else(|| Error::UnspecifiedClient(sd.sender_acct_label.clone()))?; let ain = tokio_inotify::AsyncINotify::init().expect("spool: inotify init"); let w = ain.add_watch(&sd.path, IN_CLOSE_WRITE | IN_MOVED_TO)?; println!("[spool] added watch: {:?}", sd.path); tokio::spawn(do_watch_dir(ain, sd, chan.clone())); } Ok(()) } async fn do_watch_dir( inot: tokio_inotify::AsyncINotify, sdc: config::SpoolDir, mut dest: mpsc::Sender, ) { let mut iter = inot.compat(); while let Some(ent) = iter.next().await { match ent { Ok(ent) => { // Just succ up the file and send it over. We'll do the // formatting later. let mut real_path = sdc.path.clone(); real_path.push(ent.name); let s = match file_as_string(&real_path).await { Ok(s) => s, Err(e) => { eprintln!( "[spool] warning, could not read file, ignoring: {:?}", real_path ); continue; } }; let msg = client::Message::new_delay(sdc.dest_room_id.clone(), s, sdc.send_delay_sec); let tout = time::Duration::from_secs(30); dest.send_timeout(msg, tout) .map_err(|_| ()) .await .expect("spool: relay channel send timeout"); if let Err(e) = tokiofs::remove_file(&real_path).await { eprintln!( "[spool] warning: could not remove sent file, ignoring: {:?}", real_path ); } } Err(e) => panic!("spool: error reading watch {:?}", e), } } } /* async fn process_file(p: &PathBuf, sa: &SpoolAction) -> Result<(), Error> { let ext = match p.extension().map(|e| e.to_str()).flatten() { Some(v) => v, None => { eprintln!("Found weird file {:?}, ignoring", p); return Ok(()); } }; let name = p .file_name() .map(|e| e.to_str()) .flatten() .unwrap_or("[non-UTF-8]"); // This makes me *mad*. let mut real_path = sa.watch_path.clone(); real_path.push(p); match ext { "txt" => { println!("Processing file for {} at {:?}", sa.room, p); let buf = match file_as_string(&real_path).await { Ok(v) => v, Err(Error::FileFormatMismatch) => { println!("File {} is not UTF-8, ignoring", name); return Ok(()); } Err(e) => return Err(e), }; let mut rng = rand::thread_rng(); let req = make_text_request(sa.room.clone(), buf.as_str(), &mut rng); match sa.client.as_ref().request(req).await { Ok(_) => { // Now delete it if it passed. } Err(e) => println!("Error processing {}: {:?}", name, e), } } _ => println!( "Found file {:?}, but it has unsupported extension \"{}\"", p, ext ), } Ok(()) }*/ async fn file_as_string(p: &PathBuf) -> Result { let mut buf = Vec::new(); let mut f = tokiofs::File::open(p).await?; f.read_to_end(&mut buf).await?; String::from_utf8(buf).map_err(|_| Error::FileFormatMismatch) }