🐝 My patches for macos etc...
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 

1221 lines
48 KiB

/*
* meli - mailbox module.
*
* Copyright 2017 Manos Pitsidianakis
*
* This file is part of meli.
*
* meli is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* meli is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with meli. If not, see <http://www.gnu.org/licenses/>.
*/
/*!
* https://wiki2.dovecot.org/MailboxFormat/mbox
*/
use crate::backends::*;
use crate::conf::AccountSettings;
use crate::email::parser::BytesExt;
use crate::email::*;
use crate::error::{MeliError, Result};
use crate::get_path_hash;
use crate::shellexpand::ShellExpandTrait;
use nom::bytes::complete::tag;
use nom::character::complete::digit1;
use nom::combinator::map_res;
use nom::{self, error::ErrorKind, IResult};
extern crate notify;
use self::notify::{watcher, DebouncedEvent, RecursiveMode, Watcher};
use std::collections::hash_map::{DefaultHasher, HashMap};
use std::fs::File;
use std::hash::Hasher;
use std::io::{BufReader, Read};
use std::os::unix::io::AsRawFd;
use std::path::{Path, PathBuf};
use std::str::FromStr;
use std::sync::mpsc::channel;
use std::sync::{Arc, Mutex, RwLock};
type Offset = usize;
type Length = usize;
const F_OFD_SETLKW: libc::c_int = 38;
// Open file description locking
// # man fcntl
fn get_rw_lock_blocking(f: &File) {
let fd: libc::c_int = f.as_raw_fd();
let mut flock: libc::flock = libc::flock {
l_type: libc::F_WRLCK as libc::c_short,
l_whence: libc::SEEK_SET as libc::c_short,
l_start: 0,
l_len: 0, /* "Specifying 0 for l_len has the special meaning: lock all bytes starting at the location
specified by l_whence and l_start through to the end of file, no matter how large the file grows." */
l_pid: 0, /* "By contrast with traditional record locks, the l_pid field of that structure must be set to zero when using the commands described below." */
};
let ptr: *mut libc::flock = &mut flock;
let ret_val = unsafe { libc::fcntl(fd, F_OFD_SETLKW, ptr as *mut libc::c_void) };
debug!(&ret_val);
assert!(-1 != ret_val);
}
#[derive(Debug)]
struct MboxMailbox {
hash: MailboxHash,
name: String,
path: PathBuf,
fs_path: PathBuf,
content: Vec<u8>,
children: Vec<MailboxHash>,
parent: Option<MailboxHash>,
usage: Arc<RwLock<SpecialUsageMailbox>>,
is_subscribed: bool,
permissions: MailboxPermissions,
pub total: Arc<Mutex<usize>>,
pub unseen: Arc<Mutex<usize>>,
index: Arc<Mutex<HashMap<EnvelopeHash, (Offset, Length)>>>,
}
impl BackendMailbox for MboxMailbox {
fn hash(&self) -> MailboxHash {
self.hash
}
fn name(&self) -> &str {
self.name.as_str()
}
fn path(&self) -> &str {
/* We know it's valid UTF-8 because we supplied it */
self.path.to_str().unwrap()
}
fn change_name(&mut self, s: &str) {
self.name = s.to_string();
}
fn clone(&self) -> Mailbox {
Box::new(MboxMailbox {
hash: self.hash,
name: self.name.clone(),
path: self.path.clone(),
fs_path: self.fs_path.clone(),
content: self.content.clone(),
children: self.children.clone(),
usage: self.usage.clone(),
is_subscribed: self.is_subscribed,
parent: self.parent,
permissions: self.permissions,
unseen: self.unseen.clone(),
total: self.total.clone(),
index: self.index.clone(),
})
}
fn children(&self) -> &[MailboxHash] {
&self.children
}
fn parent(&self) -> Option<MailboxHash> {
self.parent
}
fn special_usage(&self) -> SpecialUsageMailbox {
*self.usage.read().unwrap()
}
fn permissions(&self) -> MailboxPermissions {
self.permissions
}
fn is_subscribed(&self) -> bool {
self.is_subscribed
}
fn set_is_subscribed(&mut self, new_val: bool) -> Result<()> {
self.is_subscribed = new_val;
Ok(())
}
fn set_special_usage(&mut self, new_val: SpecialUsageMailbox) -> Result<()> {
*self.usage.write()? = new_val;
Ok(())
}
fn count(&self) -> Result<(usize, usize)> {
Ok((*self.unseen.lock()?, *self.total.lock()?))
}
}
/// `BackendOp` implementor for Mbox
#[derive(Debug, Default)]
pub struct MboxOp {
hash: EnvelopeHash,
path: PathBuf,
offset: Offset,
length: Length,
slice: std::cell::RefCell<Option<Vec<u8>>>,
}
impl MboxOp {
pub fn new(hash: EnvelopeHash, path: &Path, offset: Offset, length: Length) -> Self {
MboxOp {
hash,
path: path.to_path_buf(),
slice: std::cell::RefCell::new(None),
offset,
length,
}
}
}
impl BackendOp for MboxOp {
fn as_bytes(&mut self) -> ResultFuture<Vec<u8>> {
if self.slice.get_mut().is_none() {
let file = std::fs::OpenOptions::new()
.read(true)
.write(false)
.open(&self.path)?;
get_rw_lock_blocking(&file);
let mut buf_reader = BufReader::new(file);
let mut contents = Vec::new();
buf_reader.read_to_end(&mut contents)?;
*self.slice.get_mut() = Some(contents);
}
let ret = Ok(self.slice.get_mut().as_ref().unwrap().as_slice()
[self.offset..self.offset + self.length]
.to_vec());
Ok(Box::pin(async move { ret }))
}
fn fetch_flags(&self) -> ResultFuture<Flag> {
let mut flags = Flag::empty();
if self.slice.borrow().is_none() {
let file = std::fs::OpenOptions::new()
.read(true)
.write(false)
.open(&self.path)?;
get_rw_lock_blocking(&file);
let mut buf_reader = BufReader::new(file);
let mut contents = Vec::new();
buf_reader.read_to_end(&mut contents)?;
*self.slice.borrow_mut() = Some(contents);
}
let slice_ref = self.slice.borrow();
let (_, headers) = parser::headers::headers_raw(slice_ref.as_ref().unwrap().as_slice())?;
if let Some(start) = headers.find(b"Status:") {
if let Some(end) = headers[start..].find(b"\n") {
let start = start + b"Status:".len();
let status = headers[start..start + end].trim();
if status.contains(&b'F') {
flags.set(Flag::FLAGGED, true);
}
if status.contains(&b'A') {
flags.set(Flag::REPLIED, true);
}
if status.contains(&b'R') {
flags.set(Flag::SEEN, true);
}
if status.contains(&b'D') {
flags.set(Flag::TRASHED, true);
}
if status.contains(&b'T') {
flags.set(Flag::DRAFT, true);
}
}
}
if let Some(start) = headers.find(b"X-Status:") {
let start = start + b"X-Status:".len();
if let Some(end) = headers[start..].find(b"\n") {
let status = headers[start..start + end].trim();
if status.contains(&b'F') {
flags.set(Flag::FLAGGED, true);
}
if status.contains(&b'A') {
flags.set(Flag::REPLIED, true);
}
if status.contains(&b'R') {
flags.set(Flag::SEEN, true);
}
if status.contains(&b'D') {
flags.set(Flag::TRASHED, true);
}
if status.contains(&b'T') {
flags.set(Flag::DRAFT, true);
}
}
}
Ok(Box::pin(async move { Ok(flags) }))
}
}
#[derive(Debug, Clone, Copy)]
pub enum MboxReader {
MboxO,
MboxRd,
MboxCl,
MboxCl2,
}
impl Default for MboxReader {
fn default() -> Self {
Self::MboxCl2
}
}
macro_rules! find_From__line {
($input:expr) => {{
//debug!("find_From__line invocation");
let input = $input;
let mut ptr = 0;
let mut found = None;
while ptr < input.len() {
// Find next From_ candidate line.
const TAG: &'static [u8] = b"\n\nFrom ";
if let Some(end) = input[ptr..].find(TAG) {
// This candidate is a valid From_ if it ends in a new line and the next line is
// a header.
if let Some(line_end) = input[ptr + end + TAG.len()..].find(b"\n") {
if crate::email::parser::headers::header(
&input[ptr + end + TAG.len() + line_end + 1..],
)
.is_ok()
{
found = Some(ptr + end);
break;
} else {
/* Ignore invalid From_ line. */
ptr += end + TAG.len() + line_end;
}
} else {
/* Ignore invalid From_ line. */
ptr += end + TAG.len();
}
} else {
found = Some(input.len());
break;
}
}
found
}};
}
impl MboxReader {
fn parse<'i>(&self, input: &'i [u8]) -> IResult<&'i [u8], Envelope> {
let orig_input = input;
let mut input = input;
match self {
Self::MboxO => {
let next_offset: Option<(usize, usize)> = find_From__line!(input)
.and_then(|end| input.find(b"\n").map(|start| (start + 1, end)));
if let Some((start, len)) = next_offset {
match Envelope::from_bytes(&input[start..len], None) {
Ok(mut env) => {
let mut flags = Flag::empty();
if env.other_headers().contains_key("Status") {
if env.other_headers()["Status"].contains('F') {
flags.set(Flag::FLAGGED, true);
}
if env.other_headers()["Status"].contains('A') {
flags.set(Flag::REPLIED, true);
}
if env.other_headers()["Status"].contains('R') {
flags.set(Flag::SEEN, true);
}
if env.other_headers()["Status"].contains('D') {
flags.set(Flag::TRASHED, true);
}
}
if env.other_headers().contains_key("X-Status") {
if env.other_headers()["X-Status"].contains('F') {
flags.set(Flag::FLAGGED, true);
}
if env.other_headers()["X-Status"].contains('A') {
flags.set(Flag::REPLIED, true);
}
if env.other_headers()["X-Status"].contains('R') {
flags.set(Flag::SEEN, true);
}
if env.other_headers()["X-Status"].contains('D') {
flags.set(Flag::TRASHED, true);
}
if env.other_headers()["X-Status"].contains('T') {
flags.set(Flag::DRAFT, true);
}
}
env.set_flags(flags);
if len == input.len() {
Ok((&[], env))
} else {
input = &input[len + 2..];
Ok((input, env))
}
}
Err(err) => {
debug!("Could not parse mail {:?}", err);
Err(nom::Err::Error((input, ErrorKind::Tag)))
}
}
} else {
let start: Offset = input.find(b"\n").map(|v| v + 1).unwrap_or(0);
match Envelope::from_bytes(&input[start..], None) {
Ok(mut env) => {
let mut flags = Flag::empty();
if env.other_headers().contains_key("Status") {
if env.other_headers()["Status"].contains('F') {
flags.set(Flag::FLAGGED, true);
}
if env.other_headers()["Status"].contains('A') {
flags.set(Flag::REPLIED, true);
}
if env.other_headers()["Status"].contains('R') {
flags.set(Flag::SEEN, true);
}
if env.other_headers()["Status"].contains('D') {
flags.set(Flag::TRASHED, true);
}
}
if env.other_headers().contains_key("X-Status") {
if env.other_headers()["X-Status"].contains('F') {
flags.set(Flag::FLAGGED, true);
}
if env.other_headers()["X-Status"].contains('A') {
flags.set(Flag::REPLIED, true);
}
if env.other_headers()["X-Status"].contains('R') {
flags.set(Flag::SEEN, true);
}
if env.other_headers()["X-Status"].contains('D') {
flags.set(Flag::TRASHED, true);
}
if env.other_headers()["X-Status"].contains('T') {
flags.set(Flag::DRAFT, true);
}
}
env.set_flags(flags);
Ok((&[], env))
}
Err(err) => {
debug!("Could not parse mail at {:?}", err);
Err(nom::Err::Error((input, ErrorKind::Tag)))
}
}
}
}
Self::MboxRd => {
let next_offset: Option<(usize, usize)> = find_From__line!(input)
.and_then(|end| input.find(b"\n").map(|start| (start + 1, end)));
if let Some((start, len)) = next_offset {
match Envelope::from_bytes(&input[start..len], None) {
Ok(mut env) => {
let mut flags = Flag::empty();
if env.other_headers().contains_key("Status") {
if env.other_headers()["Status"].contains('F') {
flags.set(Flag::FLAGGED, true);
}
if env.other_headers()["Status"].contains('A') {
flags.set(Flag::REPLIED, true);
}
if env.other_headers()["Status"].contains('R') {
flags.set(Flag::SEEN, true);
}
if env.other_headers()["Status"].contains('D') {
flags.set(Flag::TRASHED, true);
}
}
if env.other_headers().contains_key("X-Status") {
if env.other_headers()["X-Status"].contains('F') {
flags.set(Flag::FLAGGED, true);
}
if env.other_headers()["X-Status"].contains('A') {
flags.set(Flag::REPLIED, true);
}
if env.other_headers()["X-Status"].contains('R') {
flags.set(Flag::SEEN, true);
}
if env.other_headers()["X-Status"].contains('D') {
flags.set(Flag::TRASHED, true);
}
if env.other_headers()["X-Status"].contains('T') {
flags.set(Flag::DRAFT, true);
}
}
env.set_flags(flags);
if len == input.len() {
Ok((&[], env))
} else {
input = &input[len + 2..];
Ok((input, env))
}
}
Err(err) => {
debug!("Could not parse mail {:?}", err);
Err(nom::Err::Error((input, ErrorKind::Tag)))
}
}
} else {
let start: Offset = input.find(b"\n").map(|v| v + 1).unwrap_or(0);
match Envelope::from_bytes(&input[start..], None) {
Ok(mut env) => {
let mut flags = Flag::empty();
if env.other_headers().contains_key("Status") {
if env.other_headers()["Status"].contains('F') {
flags.set(Flag::FLAGGED, true);
}
if env.other_headers()["Status"].contains('A') {
flags.set(Flag::REPLIED, true);
}
if env.other_headers()["Status"].contains('R') {
flags.set(Flag::SEEN, true);
}
if env.other_headers()["Status"].contains('D') {
flags.set(Flag::TRASHED, true);
}
}
if env.other_headers().contains_key("X-Status") {
if env.other_headers()["X-Status"].contains('F') {
flags.set(Flag::FLAGGED, true);
}
if env.other_headers()["X-Status"].contains('A') {
flags.set(Flag::REPLIED, true);
}
if env.other_headers()["X-Status"].contains('R') {
flags.set(Flag::SEEN, true);
}
if env.other_headers()["X-Status"].contains('D') {
flags.set(Flag::TRASHED, true);
}
if env.other_headers()["X-Status"].contains('T') {
flags.set(Flag::DRAFT, true);
}
}
env.set_flags(flags);
Ok((&[], env))
}
Err(err) => {
debug!("Could not parse mail {:?}", err);
Err(nom::Err::Error((input, ErrorKind::Tag)))
}
}
}
}
Self::MboxCl | Self::MboxCl2 => {
let start: Offset = input.find(b"\n").map(|v| v + 1).unwrap_or(0);
input = &input[start..];
let headers_end: usize = input.find(b"\n\n").unwrap_or(input.len());
let content_length = if let Some(v) = input[..headers_end].find(b"Content-Length: ")
{
v
} else {
// Is not MboxCl{,2}
return Self::MboxRd.parse(orig_input);
};
let (_input, _) = if let Ok(s) = tag::<_, &[u8], (&[u8], nom::error::ErrorKind)>(
"Content-Length:",
)(&input[content_length..])
{
s
} else {
return Self::MboxRd.parse(orig_input);
};
let (_input, bytes) = if let Ok(s) =
map_res::<&[u8], _, _, (&[u8], nom::error::ErrorKind), _, _, _>(
digit1,
|s: &[u8]| String::from_utf8_lossy(s).parse::<usize>(),
)(_input.ltrim())
{
s
} else {
return Self::MboxRd.parse(orig_input);
};
match Envelope::from_bytes(&input[..headers_end + bytes], None) {
Ok(mut env) => {
let mut flags = Flag::empty();
if env.other_headers().contains_key("Status") {
if env.other_headers()["Status"].contains('F') {
flags.set(Flag::FLAGGED, true);
}
if env.other_headers()["Status"].contains('A') {
flags.set(Flag::REPLIED, true);
}
if env.other_headers()["Status"].contains('R') {
flags.set(Flag::SEEN, true);
}
if env.other_headers()["Status"].contains('D') {
flags.set(Flag::TRASHED, true);
}
}
if env.other_headers().contains_key("X-Status") {
if env.other_headers()["X-Status"].contains('F') {
flags.set(Flag::FLAGGED, true);
}
if env.other_headers()["X-Status"].contains('A') {
flags.set(Flag::REPLIED, true);
}
if env.other_headers()["X-Status"].contains('R') {
flags.set(Flag::SEEN, true);
}
if env.other_headers()["X-Status"].contains('D') {
flags.set(Flag::TRASHED, true);
}
if env.other_headers()["X-Status"].contains('T') {
flags.set(Flag::DRAFT, true);
}
}
env.set_flags(flags);
if headers_end + 2 + bytes >= input.len() {
Ok((&[], env))
} else {
input = &input[headers_end + 3 + bytes..];
Ok((input, env))
}
}
Err(_err) => Self::MboxRd.parse(orig_input),
}
}
}
}
}
pub fn mbox_parse(
index: Arc<Mutex<HashMap<EnvelopeHash, (Offset, Length)>>>,
input: &[u8],
file_offset: usize,
reader: Option<MboxReader>,
) -> IResult<&[u8], Vec<Envelope>> {
if input.is_empty() {
return Err(nom::Err::Error((input, ErrorKind::Tag)));
}
let mut offset = 0;
let mut index = index.lock().unwrap();
let mut envelopes = Vec::with_capacity(32);
let reader = reader.unwrap_or(MboxReader::MboxCl2);
while !input[offset + file_offset..].is_empty() {
let (next_input, env) = match reader.parse(&input[offset + file_offset..]) {
Ok(v) => v,
Err(e) => {
// Try to recover from this error by finding a new candidate From_ line
if let Some(next_offset) = find_From__line!(&input[offset + file_offset..]) {
offset += next_offset;
if offset != input.len() {
// If we are not at EOF, we will be at this point
// "\n\nFrom ..."
// ↑
// So, skip those two newlines.
offset += 2;
}
} else {
return Err(e);
}
continue;
}
};
let start: Offset = input[offset + file_offset..]
.find(b"\n")
.map(|v| v + 1)
.unwrap_or(0);
let len = input.len() - next_input.len() - offset - file_offset - start;
index.insert(env.hash(), (offset + file_offset + start, len));
offset += len + start;
envelopes.push(env);
}
Ok((&[], envelopes))
}
struct MessageIterator<'a> {
index: Arc<Mutex<HashMap<EnvelopeHash, (Offset, Length)>>>,
input: &'a [u8],
file_offset: usize,
offset: usize,
reader: Option<MboxReader>,
}
impl<'a> Iterator for MessageIterator<'a> {
type Item = Result<Envelope>;
fn next(&mut self) -> Option<Self::Item> {
if self.input.is_empty() {
return None;
}
let mut index = self.index.lock().unwrap();
let reader = self.reader.unwrap_or(MboxReader::MboxCl2);
while !self.input[self.offset + self.file_offset..].is_empty() {
let (next_input, env) =
match reader.parse(&self.input[self.offset + self.file_offset..]) {
Ok(v) => v,
Err(e) => {
// Try to recover from this error by finding a new candidate From_ line
if let Some(next_offset) =
find_From__line!(&self.input[self.offset + self.file_offset..])
{
self.offset += next_offset;
if self.offset != self.input.len() {
// If we are not at EOF, we will be at this point
// "\n\nFrom ..."
// ↑
// So, skip those two newlines.
self.offset += 2;
}
} else {
self.input = b"";
return Some(Err(e.into()));
}
continue;
}
};
let start: Offset = self.input[self.offset + self.file_offset..]
.find(b"\n")
.map(|v| v + 1)
.unwrap_or(0);
let len = self.input.len() - next_input.len() - self.offset - self.file_offset - start;
index.insert(env.hash(), (self.offset + self.file_offset + start, len));
self.offset += len + start;
return Some(Ok(env));
}
None
}
}
/// Mbox backend
#[derive(Debug)]
pub struct MboxType {
account_name: String,
path: PathBuf,
mailbox_index: Arc<Mutex<HashMap<EnvelopeHash, MailboxHash>>>,
mailboxes: Arc<Mutex<HashMap<MailboxHash, MboxMailbox>>>,
prefer_mbox_type: Option<MboxReader>,
event_consumer: BackendEventConsumer,
}
impl MailBackend for MboxType {
fn capabilities(&self) -> MailBackendCapabilities {
const CAPABILITIES: MailBackendCapabilities = MailBackendCapabilities {
is_async: false,
is_remote: false,
supports_search: false,
extensions: None,
supports_tags: false,
supports_submission: false,
};
CAPABILITIES
}
fn is_online(&self) -> ResultFuture<()> {
Ok(Box::pin(async { Ok(()) }))
}
fn fetch(
&mut self,
mailbox_hash: MailboxHash,
) -> Result<Pin<Box<dyn Stream<Item = Result<Vec<Envelope>>> + Send + 'static>>> {
struct FetchState {
mailbox_hash: MailboxHash,
mailbox_index: Arc<Mutex<HashMap<EnvelopeHash, MailboxHash>>>,
mailboxes: Arc<Mutex<HashMap<MailboxHash, MboxMailbox>>>,
prefer_mbox_type: Option<MboxReader>,
offset: usize,
file_offset: usize,
contents: Vec<u8>,
}
impl FetchState {
async fn fetch(&mut self) -> Result<Option<Vec<Envelope>>> {
let mailboxes_lck = self.mailboxes.lock().unwrap();
let index = mailboxes_lck[&self.mailbox_hash].index.clone();
drop(mailboxes_lck);
let mut message_iter = MessageIterator {
index,
input: &self.contents.as_slice(),
offset: self.offset,
file_offset: self.file_offset,
reader: self.prefer_mbox_type,
};
let mut payload = vec![];
let mut done = false;
'iter_for_loop: for _i in 0..150 {
match message_iter.next() {
Some(Ok(env)) => {
payload.push(env);
}
Some(Err(_err)) => {
debug!(&_err);
}
None => {
done = true;
break 'iter_for_loop;
}
}
}
self.offset = message_iter.offset;
self.file_offset = message_iter.file_offset;
{
let mut mailbox_index_lck = self.mailbox_index.lock().unwrap();
for env in &payload {
mailbox_index_lck.insert(env.hash(), self.mailbox_hash);
}
}
if done {
if payload.is_empty() {
Ok(None)
} else {
let mut mailbox_lock = self.mailboxes.lock().unwrap();
let contents = std::mem::replace(&mut self.contents, vec![]);
mailbox_lock
.entry(self.mailbox_hash)
.and_modify(|f| f.content = contents);
Ok(Some(payload))
}
} else {
Ok(Some(payload))
}
}
}
let mailboxes = self.mailboxes.clone();
let mailbox_path = mailboxes.lock().unwrap()[&mailbox_hash].fs_path.clone();
let file = std::fs::OpenOptions::new()
.read(true)
.write(false)
.open(&mailbox_path)?;
get_rw_lock_blocking(&file);
let mut buf_reader = BufReader::new(file);
let mut contents = Vec::new();
buf_reader.read_to_end(&mut contents)?;
let mut state = FetchState {
mailbox_hash,
mailboxes,
mailbox_index: self.mailbox_index.clone(),
prefer_mbox_type: self.prefer_mbox_type,
contents,
offset: 0,
file_offset: 0,
};
Ok(Box::pin(async_stream::try_stream! {
loop {
if let Some(res) = state.fetch().await.map_err(|err| {
debug!("fetch err {:?}", &err);
err})? {
yield res;
} else {
return;
}
}
}))
}
fn refresh(&mut self, _mailbox_hash: MailboxHash) -> ResultFuture<()> {
Err(MeliError::new("Unimplemented."))
}
fn watch(&self) -> ResultFuture<()> {
let sender = self.event_consumer.clone();
let (tx, rx) = channel();
let mut watcher = watcher(tx, std::time::Duration::from_secs(10))
.map_err(|e| e.to_string())
.map_err(MeliError::new)?;
for f in self.mailboxes.lock().unwrap().values() {
watcher
.watch(&f.fs_path, RecursiveMode::Recursive)
.map_err(|e| e.to_string())
.map_err(MeliError::new)?;
debug!("watching {:?}", f.fs_path.as_path());
}
let account_hash = {
let mut hasher = DefaultHasher::new();
hasher.write(self.account_name.as_bytes());
hasher.finish()
};
let mailboxes = self.mailboxes.clone();
let mailbox_index = self.mailbox_index.clone();
let prefer_mbox_type = self.prefer_mbox_type;
Ok(Box::pin(async move {
loop {
match rx.recv() {
/*
* Event types:
*
* pub enum RefreshEventKind {
* Update(EnvelopeHash, Envelope), // Old hash, new envelope
* Create(Envelope),
* Remove(EnvelopeHash),
* Rescan,
* }
*/
Ok(event) => match event {
/* Update */
DebouncedEvent::NoticeWrite(pathbuf) | DebouncedEvent::Write(pathbuf) => {
let mailbox_hash = get_path_hash!(&pathbuf);
let file = match std::fs::OpenOptions::new()
.read(true)
.write(false)
.open(&pathbuf)
{
Ok(f) => f,
Err(_) => {
continue;
}
};
get_rw_lock_blocking(&file);
let mut mailbox_lock = mailboxes.lock().unwrap();
let mut buf_reader = BufReader::new(file);
let mut contents = Vec::new();
if let Err(e) = buf_reader.read_to_end(&mut contents) {
debug!(e);
continue;
};
if contents.starts_with(mailbox_lock[&mailbox_hash].content.as_slice())
{
if let Ok((_, envelopes)) = mbox_parse(
mailbox_lock[&mailbox_hash].index.clone(),
&contents,
mailbox_lock[&mailbox_hash].content.len(),
prefer_mbox_type,
) {
let mut mailbox_index_lck = mailbox_index.lock().unwrap();
for env in envelopes {
mailbox_index_lck.insert(env.hash(), mailbox_hash);
(sender)(
account_hash,
BackendEvent::Refresh(RefreshEvent {
account_hash,
mailbox_hash,
kind: RefreshEventKind::Create(Box::new(env)),
}),
);
}
}
} else {
(sender)(
account_hash,
BackendEvent::Refresh(RefreshEvent {
account_hash,
mailbox_hash,
kind: RefreshEventKind::Rescan,
}),
);
}
mailbox_lock
.entry(mailbox_hash)
.and_modify(|f| f.content = contents);
}
/* Remove */
DebouncedEvent::NoticeRemove(pathbuf) | DebouncedEvent::Remove(pathbuf) => {
if mailboxes
.lock()
.unwrap()
.values()
.any(|f| f.fs_path == pathbuf)
{
let mailbox_hash = get_path_hash!(&pathbuf);
(sender)(
account_hash,
BackendEvent::Refresh(RefreshEvent {
account_hash,
mailbox_hash,
kind: RefreshEventKind::Failure(MeliError::new(format!(
"mbox mailbox {} was removed.",
pathbuf.display()
))),
}),
);
return Ok(());
}
}
DebouncedEvent::Rename(src, dest) => {
if mailboxes.lock().unwrap().values().any(|f| f.fs_path == src) {
let mailbox_hash = get_path_hash!(&src);
(sender)(
account_hash,
BackendEvent::Refresh(RefreshEvent {
account_hash,
mailbox_hash,
kind: RefreshEventKind::Failure(MeliError::new(format!(
"mbox mailbox {} was renamed to {}.",
src.display(),
dest.display()
))),
}),
);
return Ok(());
}
}
/* Trigger rescan of mailboxes */
DebouncedEvent::Rescan => {
for &mailbox_hash in mailboxes.lock().unwrap().keys() {
(sender)(
account_hash,
BackendEvent::Refresh(RefreshEvent {
account_hash,
mailbox_hash,
kind: RefreshEventKind::Rescan,
}),
);
}
return Ok(());
}
_ => {}
},
Err(e) => debug!("watch error: {:?}", e),
}
}
}))
}
fn mailboxes(&self) -> ResultFuture<HashMap<MailboxHash, Mailbox>> {
let ret = Ok(self
.mailboxes
.lock()
.unwrap()
.iter()
.map(|(h, f)| (*h, f.clone() as Mailbox))
.collect());
Ok(Box::pin(async { ret }))
}
fn operation(&self, env_hash: EnvelopeHash) -> Result<Box<dyn BackendOp>> {
let mailbox_hash = self.mailbox_index.lock().unwrap()[&env_hash];
let mailboxes_lck = self.mailboxes.lock().unwrap();
let (offset, length) = {
let index = mailboxes_lck[&mailbox_hash].index.lock().unwrap();
index[&env_hash]
};
let mailbox_path = mailboxes_lck[&mailbox_hash].fs_path.clone();
Ok(Box::new(MboxOp::new(
env_hash,
mailbox_path.as_path(),
offset,
length,
)))
}
fn copy_messages(
&mut self,
_env_hashes: EnvelopeHashBatch,
_source_mailbox_hash: MailboxHash,
_destination_mailbox_hash: MailboxHash,
_move_: bool,
) -> ResultFuture<()> {
Err(MeliError::new("Unimplemented."))
}
fn set_flags(
&mut self,
_env_hashes: EnvelopeHashBatch,
_mailbox_hash: MailboxHash,
_flags: SmallVec<[(std::result::Result<Flag, String>, bool); 8]>,
) -> ResultFuture<()> {
Err(MeliError::new("Unimplemented."))
}
fn delete_messages(
&mut self,
_env_hashes: EnvelopeHashBatch,
_mailbox_hash: MailboxHash,
) -> ResultFuture<()> {
Err(MeliError::new("Unimplemented."))
}
fn save(
&self,
_bytes: Vec<u8>,
_mailbox_hash: MailboxHash,
_flags: Option<Flag>,
) -> ResultFuture<()> {
Err(MeliError::new("Unimplemented."))
}
fn as_any(&self) -> &dyn Any {
self
}
fn as_any_mut(&mut self) -> &mut dyn Any {
self
}
}
macro_rules! get_conf_val {
($s:ident[$var:literal]) => {
$s.extra.get($var).ok_or_else(|| {
MeliError::new(format!(
"Configuration error ({}): mbox backend requires the field `{}` set",
$s.name.as_str(),
$var
))
})
};
($s:ident[$var:literal], $default:expr) => {
$s.extra
.get($var)
.map(|v| {
<_>::from_str(v).map_err(|e| {
MeliError::new(format!(
"Configuration error ({}): Invalid value for field `{}`: {}\n{}",
$s.name.as_str(),
$var,
v,
e
))
})
})
.unwrap_or_else(|| Ok($default))
};
}
impl MboxType {
pub fn new(
s: &AccountSettings,
_is_subscribed: Box<dyn Fn(&str) -> bool>,
event_consumer: BackendEventConsumer,
) -> Result<Box<dyn MailBackend>> {
let path = Path::new(s.root_mailbox.as_str()).expand();
if !path.exists() {
return Err(MeliError::new(format!(
"\"root_mailbox\" {} for account {} is not a valid path.",
s.root_mailbox.as_str(),
s.name()
)));
}
let prefer_mbox_type: String = get_conf_val!(s["prefer_mbox_type"], "auto".to_string())?;
let ret = MboxType {
account_name: s.name().to_string(),
event_consumer,
path,
prefer_mbox_type: match prefer_mbox_type.as_str() {
"auto" => None,
"mboxo" => Some(MboxReader::MboxO),
"mboxrd" => Some(MboxReader::MboxRd),
"mboxcl" => Some(MboxReader::MboxCl),
"mboxcl2" => Some(MboxReader::MboxCl2),
_ => {
return Err(MeliError::new(format!(
"{} invalid `prefer_mbox_type` value: `{}`",
s.name(),
prefer_mbox_type,
)))
}
},
mailbox_index: Default::default(),
mailboxes: Default::default(),
};
let name: String = ret
.path
.file_name()
.map(|f| f.to_string_lossy().into())
.unwrap_or_default();
let hash = get_path_hash!(&ret.path);
let read_only = if let Ok(metadata) = std::fs::metadata(&ret.path) {
metadata.permissions().readonly()
} else {
true
};
ret.mailboxes.lock().unwrap().insert(
hash,
MboxMailbox {
hash,
path: name.clone().into(),
fs_path: ret.path.clone(),
name,
content: Vec::new(),
children: Vec::new(),
parent: None,
usage: Arc::new(RwLock::new(SpecialUsageMailbox::Normal)),
is_subscribed: true,
permissions: MailboxPermissions {
create_messages: !read_only,
remove_messages: !read_only,
set_flags: !read_only,
create_child: !read_only,
rename_messages: !read_only,
delete_messages: !read_only,
delete_mailbox: !read_only,
change_permissions: false,
},
unseen: Arc::new(Mutex::new(0)),
total: Arc::new(Mutex::new(0)),
index: Default::default(),
},
);
/* Look for other mailboxes */
for (k, f) in s.mailboxes.iter() {
if let Some(path_str) = f.extra.get("path") {
let hash = get_path_hash!(path_str);
let pathbuf: PathBuf = path_str.into();
if !pathbuf.exists() || pathbuf.is_dir() {
return Err(MeliError::new(format!(
"mbox mailbox configuration entry \"{}\" path value {} is not a file.",
k, path_str
)));
}
let read_only = if let Ok(metadata) = std::fs::metadata(&pathbuf) {
metadata.permissions().readonly()
} else {
true
};
ret.mailboxes.lock().unwrap().insert(
hash,
MboxMailbox {
hash,
name: k.to_string(),
fs_path: pathbuf,
path: k.into(),
content: Vec::new(),
children: Vec::new(),
parent: None,
usage: Arc::new(RwLock::new(f.usage.unwrap_or_default())),
is_subscribed: f.subscribe.is_true(),
permissions: MailboxPermissions {
create_messages: !read_only,
remove_messages: !read_only,
set_flags: !read_only,
create_child: !read_only,
rename_messages: !read_only,
delete_messages: !read_only,
delete_mailbox: !read_only,
change_permissions: false,
},
unseen: Arc::new(Mutex::new(0)),
total: Arc::new(Mutex::new(0)),
index: Default::default(),
},
);
} else {
return Err(MeliError::new(format!(
"mbox mailbox configuration entry \"{}\" should have a \"path\" value set pointing to an mbox file.",
k
)));
}
}
Ok(Box::new(ret))
}
pub fn validate_config(s: &AccountSettings) -> Result<()> {
let path = Path::new(s.root_mailbox.as_str()).expand();
if !path.exists() {
return Err(MeliError::new(format!(
"\"root_mailbox\" {} for account {} is not a valid path.",
s.root_mailbox.as_str(),
s.name()
)));
}
let prefer_mbox_type: Result<String> =
get_conf_val!(s["prefer_mbox_type"], "auto".to_string());
prefer_mbox_type?;
Ok(())
}
}