michael/panorama
1
0
Fork 0
Code Issues 10 Pull requests Projects 1 Releases Wiki Activity

Change to 2 spaces

Browse source
This commit is contained in:
Michael Zhang 2021-10-28 22:44:16 -05:00
parent 1ab69aadb9
commit ff18e98eff
24 changed files with 2051 additions and 2088 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
daemon/src/config/watcher.rs
View file

@ -5,8 +5,8 @@ use std::sync::mpsc as stdmpsc;
use anyhow::{Context, Result};
use futures::future::TryFutureExt;
use notify::{
recommended_watcher, Event as NotifyEvent, RecommendedWatcher,
RecursiveMode, Watcher,
recommended_watcher, Event as NotifyEvent, RecommendedWatcher, RecursiveMode,
Watcher,
};
use tokio::{sync::watch, task::JoinHandle};
use xdg::BaseDirectories;
@ -87,19 +87,14 @@ async fn start_notify_stream(
};
// TODO: any better way to do this?
let config_path_c =
config_path.canonicalize().context("cfg_path")?;
let config_path_c = config_path.canonicalize().context("cfg_path")?;
if config_path_c != path_expect {
debug!(
"did not match {:?} {:?}",
config_path_c, path_expect
);
debug!("did not match {:?} {:?}", config_path_c, path_expect);
continue;
}
debug!("reading config from {:?}", path_expect);
let config =
Config::from_file(path_expect).await.context("read")?;
let config = Config::from_file(path_expect).await.context("read")?;
// debug!("sending config {:?}", config);
config_tx.send(config)?;
}

3
daemon/src/lib.rs
View file

@ -98,8 +98,7 @@ async fn run_with_config(config: Config, exit: ExitListener) -> Result<()> {
for (account_name, account) in config.mail_accounts {
let (exit_tx, exit_rx) = oneshot::channel();
tokio::spawn(async {
match run_single_mail_account(account_name, account, exit_rx).await
{
match run_single_mail_account(account_name, account, exit_rx).await {
Ok(_) => {}
Err(err) => panic!("failed: {:?}", err),
}

23
daemon/src/mail/mod.rs
View file

@ -118,16 +118,17 @@ pub async fn sync_main(
// TODO: remove this later
// continue;
// let's just select INBOX for now, maybe have a config for default
// mailbox later?
debug!("selecting the INBOX mailbox");
let select = authed.select("INBOX").await?;
debug!("select result: {:?}", select);
loop {
let message_uids = authed.uid_search().await?;
let message_uids =
message_uids.into_iter().take(30).collect::<Vec<_>>();
let message_uids = message_uids.into_iter().take(30).collect::<Vec<_>>();
// let _ = mail2ui_tx.send(MailEvent::MessageUids(
// acct_name.clone(),
// message_uids.clone(),
@ -157,10 +158,7 @@ pub async fn sync_main(
debug!("got an event: {:?}", evt);
match evt {
Response::MailboxData(MailboxData::Exists(uid)) => {
debug!(
"NEW MESSAGE WITH UID {:?}, droping everything",
uid
);
debug!("NEW MESSAGE WITH UID {:?}, droping everything", uid);
// send DONE to stop the idle
std::mem::drop(idle_stream);
// let handle = Notification::new()
@ -170,10 +168,8 @@ pub async fn sync_main(
// .timeout(Timeout::Milliseconds(6000))
// .show()?;
let message_uids = authed.uid_search().await?;
let message_uids = message_uids
.into_iter()
.take(20)
.collect::<Vec<_>>();
let message_uids =
message_uids.into_iter().take(20).collect::<Vec<_>>();
// let _ = mail2ui_tx.send(MailEvent::MessageUids(
// acct_name.clone(),
// message_uids.clone(),
@ -184,9 +180,7 @@ pub async fn sync_main(
.uid_fetch(&message_uids, &[], FetchItems::All)
.await
.unwrap();
while let Some((_uid, _attrs)) =
message_list.next().await
{
while let Some((_uid, _attrs)) = message_list.next().await {
// let evt = MailEvent::UpdateUid(acct_name.
// clone(), uid, attrs);
// debug!("sent {:?}", evt);
@ -200,8 +194,7 @@ pub async fn sync_main(
}
} else {
loop {
tokio::time::sleep(std::time::Duration::from_secs(20))
.await;
tokio::time::sleep(std::time::Duration::from_secs(20)).await;
debug!("heartbeat");
}
}

2
daemon/src/mail/store.rs
View file

@ -8,7 +8,7 @@ pub struct MailStore {
}
impl MailStore {
/// Creates a new connection to a SQLite database.
/// Creates a new store tied to a SQLite database.
pub async fn open(uri: impl AsRef<str>) -> Result<Self> {
let pool = SqlitePoolOptions::new().connect(uri.as_ref()).await?;

16
imap/src/client/client.rs
View file

@ -201,12 +201,8 @@ impl ClientAuthenticated {
..
}) => match code {
ResponseCode::Unseen(value) => select.unseen = Some(value),
ResponseCode::UidNext(value) => {
select.uid_next = Some(value)
}
ResponseCode::UidValidity(value) => {
select.uid_validity = Some(value)
}
ResponseCode::UidNext(value) => select.uid_next = Some(value),
ResponseCode::UidValidity(value) => select.uid_validity = Some(value),
_ => {}
},
_ => warn!("unknown response {:?}", resp),
@ -250,9 +246,7 @@ impl ClientAuthenticated {
let stream = self.execute(cmd).await?;
// let (done, data) = stream.wait().await?;
Ok(stream.filter_map(|resp| match resp {
Response::Fetch(n, attrs) => {
future::ready(Some((n, attrs))).boxed()
}
Response::Fetch(n, attrs) => future::ready(Some((n, attrs))).boxed(),
Response::Done(_) => future::ready(None).boxed(),
_ => future::pending().boxed(),
}))
@ -277,9 +271,7 @@ impl ClientAuthenticated {
let stream = self.execute(cmd).await?;
// let (done, data) = stream.wait().await?;
Ok(stream.filter_map(|resp| match resp {
Response::Fetch(n, attrs) => {
future::ready(Some((n, attrs))).boxed()
}
Response::Fetch(n, attrs) => future::ready(Some((n, attrs))).boxed(),
Response::Done(_) => future::ready(None).boxed(),
_ => future::pending().boxed(),
}))

23
imap/src/client/inner.rs
View file

@ -12,8 +12,7 @@ use futures::{
use panorama_proto_common::Bytes;
use tokio::{
io::{
split, AsyncRead, AsyncWrite, AsyncWriteExt, BufWriter, ReadHalf,
WriteHalf,
split, AsyncRead, AsyncWrite, AsyncWriteExt, BufWriter, ReadHalf, WriteHalf,
},
sync::{mpsc, oneshot, RwLock},
task::JoinHandle,
@ -93,8 +92,7 @@ where
// spawn the client->server loop
let (write_exit, exit_rx) = oneshot::channel();
let write_handle =
tokio::spawn(write_loop(write_half, exit_rx, write_rx));
let write_handle = tokio::spawn(write_loop(write_half, exit_rx, write_rx));
let tag_number = AtomicU32::new(0);
let capabilities = Arc::new(RwLock::new(None));
@ -112,10 +110,7 @@ where
})
}
pub async fn execute(
&mut self,
command: Command,
) -> Result<ResponseStream> {
pub async fn execute(&mut self, command: Command) -> Result<ResponseStream> {
let id = self.tag_number.fetch_add(1, Ordering::SeqCst);
let tag = Tag(Bytes::from(format!("{}{}", TAG_PREFIX, id)));
@ -130,10 +125,7 @@ where
Ok(stream)
}
pub async fn has_capability(
&mut self,
cap: impl AsRef<str>,
) -> Result<bool> {
pub async fn has_capability(&mut self, cap: impl AsRef<str>) -> Result<bool> {
let mut cap_slice = cap.as_ref().as_bytes().to_vec();
// since we're doing incremental parsing, we have to finish this off
@ -143,8 +135,8 @@ where
let cap_bytes = Bytes::from(cap_slice);
trace!("CAP_BYTES: {:?}", cap_bytes);
let (_, cap) = parse_capability(cap_bytes)
.context("could not parse capability")?;
let (_, cap) =
parse_capability(cap_bytes).context("could not parse capability")?;
let contains = {
let read = self.capabilities.read().await;
@ -193,7 +185,8 @@ where
.execute(Command::Starttls)
.await
.context("could not send starttls command")?;
resp.wait()
resp
.wait()
.await
.context("could not receive starttls response")?;
debug!("received OK from server");

4
imap/src/client/response_stream.rs
View file

@ -25,9 +25,7 @@ impl ResponseStream {
/// Waits for the entire stream to finish, returning the DONE status and the
/// stream
pub async fn wait(
mut self,
) -> Result<(Option<ResponseDone>, Vec<Response>)> {
pub async fn wait(mut self) -> Result<(Option<ResponseDone>, Vec<Response>)> {
let mut done = None;
let mut vec = Vec::new();
while let Some(resp) = self.inner.recv().await {

3
imap/src/client/tls.rs
View file

@ -5,8 +5,7 @@ use tokio::io::{AsyncRead, AsyncWrite};
use tokio_rustls::{
client::TlsStream,
rustls::{
ClientConfig as RustlsConfig, OwnedTrustAnchor, RootCertStore,
ServerName,
ClientConfig as RustlsConfig, OwnedTrustAnchor, RootCertStore, ServerName,
},
TlsConnector,
};

15
proto-common/src/convert_error.rs
View file

@ -47,29 +47,20 @@ pub fn convert_error<I: Deref<Target = [u8]> + Debug>(
if input.is_empty() {
match kind {
VerboseErrorKind::Char(c) => {
write!(
&mut result,
"{}: expected '{}', got empty input\n\n",
i, c
)
write!(&mut result, "{}: expected '{}', got empty input\n\n", i, c)
}
VerboseErrorKind::Context(s) => {
write!(&mut result, "{}: in {}, got empty input\n\n", i, s)
}
VerboseErrorKind::Nom(e) => {
write!(
&mut result,
"{}: in {:?}, got empty input\n\n",
i, e
)
write!(&mut result, "{}: in {:?}, got empty input\n\n", i, e)
}
}
} else {
let prefix = &input.as_bytes()[..offset];
// Count the number of newlines in the first `offset` bytes of input
let line_number =
prefix.iter().filter(|&&b| b == b'\n').count() + 1;
let line_number = prefix.iter().filter(|&&b| b == b'\n').count() + 1;
// Find the line that includes the subslice:
// Find the *last* newline before the substring starts

6
proto-common/src/parsers.rs
View file

@ -44,7 +44,8 @@ macro_rules! sep_list {
many0(preceded(panorama_proto_common::byte(b'\x20'), $t)),
)),
|opt| {
opt.map(|(hd, mut tl)| {
opt
.map(|(hd, mut tl)| {
tl.insert(0, hd);
tl
})
@ -54,7 +55,8 @@ macro_rules! sep_list {
};
(? $t:expr, $d:expr) => {
map(opt(pair($t, many0(preceded($d, $t)))), |opt| {
opt.map(|(hd, mut tl)| {
opt
.map(|(hd, mut tl)| {
tl.insert(0, hd);
tl
})

1
rustfmt.toml
View file

@ -1,3 +1,4 @@
fn_single_line = true
max_width = 80
wrap_comments = true
tab_spaces = 2