sander/rust-imap
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

send unsolicited responses to a channel instead of discarding them

Browse source 
notes

  * i tried to avoid the term "async", because that term is very
    overloaded and we're not using e.g. tokio/async-io here
  * i'm a little unhappy having to string the channel through the
    parser, because that seems rather a part of the client logic than
    parsing. on the other hand it's better than passing the whole
    client, so there's that at least.
This commit is contained in:
Johannes Schilling 2018-10-29 19:51:12 +01:00
parent 5c91f4c1bb
commit f98398aecf
3 changed files with 109 additions and 41 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

46
src/client.rs
View file

@ -4,6 +4,7 @@ use std::collections::HashSet;
use std::io::{self, Read, Write};
use std::net::{TcpStream, ToSocketAddrs};
use std::ops::{Deref, DerefMut};
use std::sync::mpsc;
use std::time::Duration;
use super::authenticator::Authenticator;
@ -44,6 +45,10 @@ fn validate_str(value: &str) -> Result<String> {
#[derive(Debug)]
pub struct Session<T: Read + Write> {
conn: Connection<T>,
/// Server responses that are not related to the current command. See also the note on
/// [unilateral server responses in RFC 3501](https://tools.ietf.org/html/rfc3501#section-7).
pub unsolicited_responses: mpsc::Receiver<UnsolicitedResponse>,
unsolicited_responses_tx: mpsc::Sender<UnsolicitedResponse>,
}
/// An (unauthenticated) handle to talk to an IMAP server. This is what you get when first
@ -396,7 +401,7 @@ impl<T: Read + Write> Client<T> {
);
} else {
ok_or_unauth_client_err!(self.read_response_onto(&mut line), self);
return Ok(Session { conn: self.conn });
return Ok(Session::new(self.conn));
}
}
}
@ -443,27 +448,34 @@ impl<T: Read + Write> Client<T> {
self
);
Ok(Session { conn: self.conn })
Ok(Session::new(self.conn))
}
}
impl<T: Read + Write> Session<T> {
// not public, just to avoid duplicating the channel creation code
fn new(conn: Connection<T>) -> Self {
let (tx, rx) = mpsc::channel();
Session { conn, unsolicited_responses: rx, unsolicited_responses_tx: tx }
}
/// Selects a mailbox
///
/// Note that the server *is* allowed to unilaterally send things to the client for messages in
/// a selected mailbox whose status has changed. See the note on [unilateral server responses
/// in RFC 3501](https://tools.ietf.org/html/rfc3501#section-7). This means that if you use
/// [`Connection::run_command_and_read_response`], you *may* see additional untagged `RECENT`,
/// `EXISTS`, `FETCH`, and `EXPUNGE` responses!
/// `EXISTS`, `FETCH`, and `EXPUNGE` responses. You can get them from the
/// `unsolicited_responses` channel of the [`Session`](struct.Session.html).
pub fn select(&mut self, mailbox_name: &str) -> Result<Mailbox> {
self.run_command_and_read_response(&format!("SELECT {}", validate_str(mailbox_name)?))
.and_then(|lines| parse_mailbox(&lines[..]))
.and_then(|lines| parse_mailbox(&lines[..], self.unsolicited_responses_tx.clone()))
}
/// Examine is identical to Select, but the selected mailbox is identified as read-only
pub fn examine(&mut self, mailbox_name: &str) -> Result<Mailbox> {
self.run_command_and_read_response(&format!("EXAMINE {}", validate_str(mailbox_name)?))
.and_then(|lines| parse_mailbox(&lines[..]))
.and_then(|lines| parse_mailbox(&lines[..], self.unsolicited_responses_tx.clone()))
}
/// Fetch retreives data associated with a set of messages in the mailbox.
@ -473,7 +485,7 @@ impl<T: Read + Write> Session<T> {
/// server responses in RFC 3501](https://tools.ietf.org/html/rfc3501#section-7).
pub fn fetch(&mut self, sequence_set: &str, query: &str) -> ZeroCopyResult<Vec<Fetch>> {
self.run_command_and_read_response(&format!("FETCH {} {}", sequence_set, query))
.and_then(parse_fetches)
.and_then(|lines| parse_fetches(lines, self.unsolicited_responses_tx.clone()))
}
/// Fetch retreives data associated with a set of messages by UID in the mailbox.
@ -483,7 +495,7 @@ impl<T: Read + Write> Session<T> {
/// server responses in RFC 3501](https://tools.ietf.org/html/rfc3501#section-7).
pub fn uid_fetch(&mut self, uid_set: &str, query: &str) -> ZeroCopyResult<Vec<Fetch>> {
self.run_command_and_read_response(&format!("UID FETCH {} {}", uid_set, query))
.and_then(parse_fetches)
.and_then(|lines| parse_fetches(lines, self.unsolicited_responses_tx.clone()))
}
/// Noop always succeeds, and it does nothing.
@ -530,7 +542,7 @@ impl<T: Read + Write> Session<T> {
/// Capability requests a listing of capabilities that the server supports.
pub fn capabilities(&mut self) -> ZeroCopyResult<Capabilities> {
self.run_command_and_read_response("CAPABILITY")
.and_then(parse_capabilities)
.and_then(|lines| parse_capabilities(lines, self.unsolicited_responses_tx.clone()))
}
/// Expunge permanently removes all messages that have the \Deleted flag set from the currently
@ -560,12 +572,12 @@ impl<T: Read + Write> Session<T> {
/// Store alters data associated with a message in the mailbox.
pub fn store(&mut self, sequence_set: &str, query: &str) -> ZeroCopyResult<Vec<Fetch>> {
self.run_command_and_read_response(&format!("STORE {} {}", sequence_set, query))
.and_then(parse_fetches)
.and_then(|lines| parse_fetches(lines, self.unsolicited_responses_tx.clone()))
}
pub fn uid_store(&mut self, uid_set: &str, query: &str) -> ZeroCopyResult<Vec<Fetch>> {
self.run_command_and_read_response(&format!("UID STORE {} {}", uid_set, query))
.and_then(parse_fetches)
.and_then(|lines| parse_fetches(lines, self.unsolicited_responses_tx.clone()))
}
/// Copy copies the specified message to the end of the specified destination mailbox.
@ -612,7 +624,7 @@ impl<T: Read + Write> Session<T> {
quote!(reference_name),
mailbox_search_pattern
))
.and_then(parse_names)
.and_then(|lines| parse_names(lines, self.unsolicited_responses_tx.clone()))
}
/// The LSUB command returns a subset of names from the set of names
@ -627,7 +639,7 @@ impl<T: Read + Write> Session<T> {
quote!(reference_name),
mailbox_search_pattern
))
.and_then(parse_names)
.and_then(|lines| parse_names(lines, self.unsolicited_responses_tx.clone()))
}
/// The STATUS command requests the status of the indicated mailbox.
@ -637,7 +649,7 @@ impl<T: Read + Write> Session<T> {
validate_str(mailbox_name)?,
status_data_items
))
.and_then(|lines| parse_mailbox(&lines[..]))
.and_then(|lines| parse_mailbox(&lines[..], self.unsolicited_responses_tx.clone()))
}
/// Returns a handle that can be used to block until the state of the currently selected
@ -664,14 +676,14 @@ impl<T: Read + Write> Session<T> {
/// the list of message sequence numbers of those messages.
pub fn search(&mut self, query: &str) -> Result<HashSet<u32>> {
self.run_command_and_read_response(&format!("SEARCH {}", query))
.and_then(parse_ids)
.and_then(|lines| parse_ids(lines, self.unsolicited_responses_tx.clone()))
}
/// Searches the mailbox for messages that match the given criteria and returns
/// the list of unique identifier numbers of those messages.
pub fn uid_search(&mut self, query: &str) -> Result<HashSet<u32>> {
self.run_command_and_read_response(&format!("UID SEARCH {}", query))
.and_then(parse_ids)
.and_then(|lines| parse_ids(lines, self.unsolicited_responses_tx.clone()))
}
// these are only here because they are public interface, the rest is in `Connection`
@ -836,9 +848,7 @@ mod tests {
macro_rules! mock_session {
($s:expr) => {
Session {
conn: Client::new($s).conn,
}
Session::new(Client::new($s).conn)
};
}

85
src/parse.rs
View file

@ -1,6 +1,7 @@
use imap_proto::{self, MailboxDatum, Response};
use regex::Regex;
use std::collections::HashSet;
use std::sync::mpsc;
use super::error::{Error, ParseError, Result};
use super::types::*;
@ -23,7 +24,7 @@ enum MapOrNot<T> {
Ignore,
}
unsafe fn parse_many<T, F>(lines: Vec<u8>, mut map: F) -> ZeroCopyResult<Vec<T>>
unsafe fn parse_many<T, F>(lines: Vec<u8>, mut map: F, unsolicited: mpsc::Sender<UnsolicitedResponse>) -> ZeroCopyResult<Vec<T>>
where
F: FnMut(Response<'static>) -> MapOrNot<T>,
{
@ -44,15 +45,28 @@ where
// check if this is simply a unilateral server response
// (see Section 7 of RFC 3501):
match resp {
Response::MailboxData(MailboxDatum::Recent { .. })
| Response::MailboxData(MailboxDatum::Exists { .. })
| Response::Fetch(..)
| Response::Expunge(..) => {
Response::MailboxData(MailboxDatum::Recent(n)) => {
unsolicited.send(UnsolicitedResponse::Recent(n))
.unwrap();
}
Response::MailboxData(MailboxDatum::Exists(n)) => {
unsolicited.send(UnsolicitedResponse::Exists(n))
.unwrap();
}
Response::Expunge(id) => {
unsolicited.send(UnsolicitedResponse::Expunge(id))
.unwrap();
}
Response::MailboxData(MailboxDatum::Status { mailbox, status }) => {
unsolicited.send(UnsolicitedResponse::Status(mailbox.into(), status))
.unwrap();
}
Response::Fetch(..) => {
continue;
}
resp => break Err(resp.into()),
}
}
},
MapOrNot::Ignore => continue,
}
}
@ -66,7 +80,7 @@ where
ZeroCopy::new(lines, f)
}
pub fn parse_names(lines: Vec<u8>) -> ZeroCopyResult<Vec<Name>> {
pub fn parse_names(lines: Vec<u8>, unsolicited: mpsc::Sender<UnsolicitedResponse>) -> ZeroCopyResult<Vec<Name>> {
use imap_proto::MailboxDatum;
let f = |resp| match resp {
// https://github.com/djc/imap-proto/issues/4
@ -87,10 +101,10 @@ pub fn parse_names(lines: Vec<u8>) -> ZeroCopyResult<Vec<Name>> {
resp => MapOrNot::Not(resp),
};
unsafe { parse_many(lines, f) }
unsafe { parse_many(lines, f, unsolicited) }
}
pub fn parse_fetches(lines: Vec<u8>) -> ZeroCopyResult<Vec<Fetch>> {
pub fn parse_fetches(lines: Vec<u8>, unsolicited: mpsc::Sender<UnsolicitedResponse>) -> ZeroCopyResult<Vec<Fetch>> {
let f = |resp| match resp {
Response::Fetch(num, attrs) => {
let mut fetch = Fetch {
@ -121,10 +135,10 @@ pub fn parse_fetches(lines: Vec<u8>) -> ZeroCopyResult<Vec<Fetch>> {
resp => MapOrNot::Not(resp),
};
unsafe { parse_many(lines, f) }
unsafe { parse_many(lines, f, unsolicited) }
}
pub fn parse_capabilities(lines: Vec<u8>) -> ZeroCopyResult<Capabilities> {
pub fn parse_capabilities(lines: Vec<u8>, unsolicited: mpsc::Sender<UnsolicitedResponse>) -> ZeroCopyResult<Capabilities> {
let f = |mut lines| {
let mut caps = HashSet::new();
loop {
@ -137,6 +151,10 @@ pub fn parse_capabilities(lines: Vec<u8>) -> ZeroCopyResult<Capabilities> {
break Ok(Capabilities(caps));
}
}
Ok((rest, Response::MailboxData(MailboxDatum::Status { mailbox, status }))) => {
lines = rest;
unsolicited.send(UnsolicitedResponse::Status(mailbox.into(), status)).unwrap();
}
Ok((_, resp)) => {
break Err(resp.into());
}
@ -150,7 +168,7 @@ pub fn parse_capabilities(lines: Vec<u8>) -> ZeroCopyResult<Capabilities> {
unsafe { ZeroCopy::new(lines, f) }
}
pub fn parse_mailbox(mut lines: &[u8]) -> Result<Mailbox> {
pub fn parse_mailbox(mut lines: &[u8], unsolicited: mpsc::Sender<UnsolicitedResponse>) -> Result<Mailbox> {
let mut mailbox = Mailbox::default();
loop {
@ -188,8 +206,8 @@ pub fn parse_mailbox(mut lines: &[u8]) -> Result<Mailbox> {
use imap_proto::MailboxDatum;
match m {
MailboxDatum::Status { .. } => {
// TODO: we probably want to expose statuses too
MailboxDatum::Status { mailbox, status } => {
unsolicited.send(UnsolicitedResponse::Status(mailbox.into(), status)).unwrap();
}
MailboxDatum::Exists(e) => {
mailbox.exists = e;
@ -219,7 +237,7 @@ pub fn parse_mailbox(mut lines: &[u8]) -> Result<Mailbox> {
}
}
pub fn parse_ids(lines: Vec<u8>) -> Result<HashSet<u32>> {
pub fn parse_ids(lines: Vec<u8>, unsolicited: mpsc::Sender<UnsolicitedResponse>) -> Result<HashSet<u32>> {
let mut lines = &lines[..];
let mut ids = HashSet::new();
loop {
@ -232,6 +250,10 @@ pub fn parse_ids(lines: Vec<u8>) -> Result<HashSet<u32>> {
break Ok(ids);
}
}
Ok((rest, Response::MailboxData(MailboxDatum::Status { mailbox, status }))) => {
lines = rest;
unsolicited.send(UnsolicitedResponse::Status(mailbox.into(), status)).unwrap();
}
Ok((_, resp)) => {
break Err(resp.into());
}
@ -250,7 +272,10 @@ mod tests {
fn parse_capability_test() {
let expected_capabilities = vec!["IMAP4rev1", "STARTTLS", "AUTH=GSSAPI", "LOGINDISABLED"];
let lines = b"* CAPABILITY IMAP4rev1 STARTTLS AUTH=GSSAPI LOGINDISABLED\r\n";
let capabilities = parse_capabilities(lines.to_vec()).unwrap();
let (send, recv) = mpsc::channel();
let capabilities = parse_capabilities(lines.to_vec(), send).unwrap();
// shouldn't be any unexpected responses parsed
assert!(recv.try_recv().is_err());
assert_eq!(capabilities.len(), 4);
for e in expected_capabilities {
assert!(capabilities.has(e));
@ -260,14 +285,18 @@ mod tests {
#[test]
#[should_panic]
fn parse_capability_invalid_test() {
let (send, recv) = mpsc::channel();
let lines = b"* JUNK IMAP4rev1 STARTTLS AUTH=GSSAPI LOGINDISABLED\r\n";
parse_capabilities(lines.to_vec()).unwrap();
parse_capabilities(lines.to_vec(), send).unwrap();
assert!(recv.try_recv().is_err());
}
#[test]
fn parse_names_test() {
let lines = b"* LIST (\\HasNoChildren) \".\" \"INBOX\"\r\n";
let names = parse_names(lines.to_vec()).unwrap();
let (send, recv) = mpsc::channel();
let names = parse_names(lines.to_vec(), send).unwrap();
assert!(recv.try_recv().is_err());
assert_eq!(names.len(), 1);
assert_eq!(names[0].attributes(), &["\\HasNoChildren"]);
assert_eq!(names[0].delimiter(), ".");
@ -277,7 +306,9 @@ mod tests {
#[test]
fn parse_fetches_empty() {
let lines = b"";
let fetches = parse_fetches(lines.to_vec()).unwrap();
let (send, recv) = mpsc::channel();
let fetches = parse_fetches(lines.to_vec(), send).unwrap();
assert!(recv.try_recv().is_err());
assert!(fetches.is_empty());
}
@ -286,7 +317,9 @@ mod tests {
let lines = b"\
* 24 FETCH (FLAGS (\\Seen) UID 4827943)\r\n\
* 25 FETCH (FLAGS (\\Seen))\r\n";
let fetches = parse_fetches(lines.to_vec()).unwrap();
let (send, recv) = mpsc::channel();
let fetches = parse_fetches(lines.to_vec(), send).unwrap();
assert!(recv.try_recv().is_err());
assert_eq!(fetches.len(), 2);
assert_eq!(fetches[0].message, 24);
assert_eq!(fetches[0].flags(), &["\\Seen"]);
@ -304,7 +337,9 @@ mod tests {
let lines = b"\
* 37 FETCH (UID 74)\r\n\
* 1 RECENT\r\n";
let fetches = parse_fetches(lines.to_vec()).unwrap();
let (send, recv) = mpsc::channel();
let fetches = parse_fetches(lines.to_vec(), send).unwrap();
assert_eq!(recv.try_recv(), Ok(UnsolicitedResponse::Recent(1)));
assert_eq!(fetches.len(), 1);
assert_eq!(fetches[0].message, 37);
assert_eq!(fetches[0].uid, Some(74));
@ -314,7 +349,9 @@ mod tests {
fn parse_ids_test() {
let lines = b"* SEARCH 1600 1698 1739 1781 1795 1885 1891 1892 1893 1898 1899 1901 1911 1926 1932 1933 1993 1994 2007 2032 2033 2041 2053 2062 2063 2065 2066 2072 2078 2079 2082 2084 2095 2100 2101 2102 2103 2104 2107 2116 2120 2135 2138 2154 2163 2168 2172 2189 2193 2198 2199 2205 2212 2213 2221 2227 2267 2275 2276 2295 2300 2328 2330 2332 2333 2334\r\n\
* SEARCH 2335 2336 2337 2338 2339 2341 2342 2347 2349 2350 2358 2359 2362 2369 2371 2372 2373 2374 2375 2376 2377 2378 2379 2380 2381 2382 2383 2384 2385 2386 2390 2392 2397 2400 2401 2403 2405 2409 2411 2414 2417 2419 2420 2424 2426 2428 2439 2454 2456 2467 2468 2469 2490 2515 2519 2520 2521\r\n";
let ids = parse_ids(lines.to_vec()).unwrap();
let (send, recv) = mpsc::channel();
let ids = parse_ids(lines.to_vec(), send).unwrap();
assert!(recv.try_recv().is_err());
let ids: HashSet<u32> = ids.iter().cloned().collect();
assert_eq!(
ids,
@ -335,7 +372,9 @@ mod tests {
);
let lines = b"* SEARCH\r\n";
let ids = parse_ids(lines.to_vec()).unwrap();
let (send, recv) = mpsc::channel();
let ids = parse_ids(lines.to_vec(), send).unwrap();
assert!(recv.try_recv().is_err());
let ids: HashSet<u32> = ids.iter().cloned().collect();
assert_eq!(ids, HashSet::<u32>::new());
}

19
src/types/mod.rs
View file

@ -10,6 +10,25 @@ pub use self::name::Name;
mod capabilities;
pub use self::capabilities::Capabilities;
/// re-exported from imap_proto;
pub use imap_proto::StatusAttribute;
/// Responses that the server sends that are not related to the current command.
/// [RFC 3501](https://tools.ietf.org/html/rfc3501#section-7) states that clients need to be able
/// to accept any response at any time. These are the ones we've encountered in the wild.
///
/// Note that `Recent`, `Exists` and `Expunge` responses refer to the currently `SELECT`ed folder,
/// so the user must take care when interpreting these.
#[derive(Debug, PartialEq, Eq)]
pub enum UnsolicitedResponse {
Status(String, Vec<StatusAttribute>),
Recent(u32),
Exists(u32),
Expunge(u32),
}
pub struct ZeroCopy<D> {
_owned: Box<[u8]>,
derived: D,