http-fs 0.5.0

//! File service module
use mime_guess::{guess_mime_type};
use futures::{Future, Stream};
use bytes::Bytes;

use crate::config::{DefaultConfig, FileServeConfig};
use crate::headers::cd::{self, ContentDisposition, DispositionType};
use crate::utils;
use crate::headers::range::{HttpRange};

use std::io::{self, Seek, Read};
use std::path::Path;
use std::fs;
use std::marker::PhantomData;
use std::str::FromStr;
use std::cmp;

///Performs match of `ETag` against `If-Match`
///
///Returns `true` if has no `If-Match` header or the `Etag` doesn't match it
pub fn if_match(etag: &etag::EntityTag, headers: &http::header::HeaderMap) -> bool {
    match headers.get(http::header::IF_MATCH).and_then(|header| header.to_str().ok()) {
        Some(header) => {
            for header_tag in header.split(',').map(|tag| tag.trim()) {
                match header_tag.parse::<etag::EntityTag>() {
                    Ok(header_tag) => match etag.strong_eq(&header_tag) {
                        true => return true,
                        false => (),
                    },
                    Err(_) => ()
                }
            }
            false
        },
        None => true
    }
}

///Matches given ETag against `If-None-Match` header.
///
///Returns `true` if matching value found is not found or misses the header
pub fn if_none_match(etag: &etag::EntityTag, headers: &http::header::HeaderMap) -> bool {
    match headers.get(http::header::IF_NONE_MATCH).and_then(|header| header.to_str().ok()) {
        Some(header) => {
            for header_tag in header.split(',').map(|tag| tag.trim()) {
                match header_tag.parse::<etag::EntityTag>() {
                    Ok(header_tag) => match etag.weak_eq(&header_tag) {
                        true => return false,
                        false => (),
                    },
                    Err(_) => ()
                }
            }
            true
        },
        None => true
    }
}

///Matches `Last-Modified` against `If-Unmodified-Since`
///
///Returns true if `Last-Modified` is not after `If-Unmodified-Since`
///Or the header is missing
pub fn if_unmodified_since(last_modified: httpdate::HttpDate, headers: &http::header::HeaderMap) -> bool {
    match headers.get(http::header::IF_UNMODIFIED_SINCE).and_then(|header| header.to_str().ok()).and_then(|header| httpdate::HttpDate::from_str(header.trim()).ok()) {
        Some(header) => last_modified <= header,
        None => true,
    }
}

///Matches `Last-Modified` against `If-Modified-Since`
///
///Returns true if `Last-Modified` is before `If-Modified-Since`, header is missing or
///`If-None-Matches` is present
pub fn if_modified_since(last_modified: httpdate::HttpDate, headers: &http::header::HeaderMap) -> bool {
    if headers.contains_key(http::header::IF_NONE_MATCH) {
        return true;
    }

    match headers.get(http::header::IF_MODIFIED_SINCE).and_then(|header| header.to_str().ok()).and_then(|header| httpdate::HttpDate::from_str(header.trim()).ok()) {
        Some(header) => last_modified > header,
        None => true,
    }
}


///File service helper
pub struct ServeFile<C = DefaultConfig> {
    file: fs::File,
    meta: fs::Metadata,
    ///File's MIME type
    pub content_type: mime::Mime,
    ///File's Content-Disposition
    pub content_disposition: ContentDisposition,
    _config: PhantomData<C>,
}

impl ServeFile<DefaultConfig> {
    #[inline]
    ///Creates new instance from already opened file
    pub fn from_parts(file_name: &str, file: fs::File, meta: fs::Metadata) -> Self {
        Self::from_parts_with_cfg(file_name, file, meta)
    }
}

impl<C: FileServeConfig> ServeFile<C> {
    ///Opens file to serve.
    pub fn open(path: &Path) -> io::Result<Self> {
        let file = fs::File::open(path)?;
        let meta = file.metadata()?;

        if let Some(file_name) = path.file_name().and_then(|file_name| file_name.to_str()) {
            Ok(Self::from_parts_with_cfg(file_name, file, meta))
        } else {
            Err(io::Error::new(io::ErrorKind::InvalidInput, "Provided path has no filename"))
        }
    }

    ///Creates new instance from already opened file
    pub fn from_parts_with_cfg(file_name: &str, file: fs::File, meta: fs::Metadata) -> Self {
        let (content_type, content_disposition) = {
            let content_type = guess_mime_type(&file_name);
            let content_disposition = match C::content_disposition_map(content_type.type_()) {
                DispositionType::Inline => ContentDisposition::Inline,
                DispositionType::Attachment => ContentDisposition::Attachment(cd::Filename::with_encoded_name(file_name.into())),
            };

            (content_type, content_disposition)
        };

        Self {
            file,
            meta,
            content_type,
            content_disposition,
            _config: PhantomData,
        }
    }

    #[inline]
    ///Creates `EntityTag` for file
    pub fn etag(&self) -> etag::EntityTag {
        etag::EntityTag::from_file_meta(&self.meta)
    }

    #[inline]
    ///Creates `HttpDate` instance for file, if possible
    pub fn last_modified(&self) -> Option<httpdate::HttpDate> {
        self.meta.modified().map(|modified| modified.into()).ok()
    }

    #[inline]
    ///Returns length of File.
    pub fn len(&self) -> u64 {
        self.meta.len()
    }

    ///Prepares file for service
    pub fn prepare(self, path: &Path, method: http::Method, headers: &http::HeaderMap, out_headers: &mut http::HeaderMap, workers: &threadpool::ThreadPool) -> (http::StatusCode, Option<ChunkedReadFile<C>>) {
        //ETag is more reliable so it is given priority.
        if C::is_use_etag(&path) {
            let etag = self.etag();

            //As per RFC we must send useful cache related headers.
            //Since cache is from ETag let's send ETag only
            out_headers.insert(http::header::ETAG, utils::display_to_header(&etag));

            if !if_match(&etag, headers) {
                return (http::StatusCode::PRECONDITION_FAILED, None);
            } else if !if_none_match(&etag, headers) {
                return (http::StatusCode::NOT_MODIFIED, None);
            }
        }

        if C::is_use_last_modifier(&path) {
            if let Some(last_modified) = self.last_modified() {
                out_headers.insert(http::header::LAST_MODIFIED, utils::display_to_header(&last_modified));

                if !if_unmodified_since(last_modified, headers) {
                    return (http::StatusCode::PRECONDITION_FAILED, None);
                } else if !if_modified_since(last_modified, headers) {
                    return (http::StatusCode::NOT_MODIFIED, None);
                }
            }
        }

        out_headers.insert(http::header::CONTENT_TYPE, utils::display_to_header(&self.content_type));
        out_headers.insert(http::header::CONTENT_DISPOSITION, utils::display_to_header(&self.content_disposition));
        out_headers.insert(http::header::ACCEPT_RANGES, http::header::HeaderValue::from_static("bytes"));

        let mut length = self.len();
        let mut offset = 0;

        // check for range header
        if let Some(ranges) = headers.get(http::header::RANGE) {
            if let Ok(ranges_header) = ranges.to_str() {
                if let Ok(ranges_vec) = HttpRange::parse(ranges_header, length) {
                    length = ranges_vec[0].length;
                    offset = ranges_vec[0].start;
                    let content_range = utils::display_to_header(&format_args!("bytes {}-{}/{}", offset, offset + length - 1, self.len()));
                    out_headers.insert(http::header::CONTENT_RANGE, content_range);
                } else {
                    let content_range = utils::display_to_header(&format_args!("bytes */{}", length));
                    out_headers.insert(http::header::CONTENT_RANGE, content_range);
                    return (http::StatusCode::RANGE_NOT_SATISFIABLE, None);
                }
            } else {
                return (http::StatusCode::BAD_REQUEST, None);
            };
        };

        out_headers.insert(http::header::CONTENT_LENGTH, utils::display_to_header(&length));

        match method {
            http::Method::HEAD => (http::StatusCode::OK, None),
            _ => {
                let code = if offset != 0 || length != self.len() {
                    http::StatusCode::PARTIAL_CONTENT
                } else {
                    http::StatusCode::OK
                };

                let reader = ChunkedReadFile::<C>::new(length, offset, self.into(), workers.clone());
                (code, Some(reader))
            },
        }
    }
}

impl<C> Into<fs::File> for ServeFile<C> {
    fn into(self) -> fs::File {
        self.file
    }
}

///Stream to read chunks of file
pub struct ChunkedReadFile<C> {
    ///Size of file to read
    pub size: u64,
    offset: u64,
    thread_pool: threadpool::ThreadPool,
    file: Option<fs::File>,
    fut: Option<futures::sync::oneshot::Receiver<Result<(fs::File, Bytes), io::Error>>>,
    counter: u64,
    _config: PhantomData<C>,
}

impl<C: FileServeConfig> ChunkedReadFile<C> {
    ///Creates new instance
    pub fn new(size: u64, offset: u64, file: fs::File, thread_pool: threadpool::ThreadPool) -> Self {
        Self {
            size,
            offset,
            thread_pool,
            file: Some(file),
            fut: None,
            counter: 0,
            _config: PhantomData
        }
    }
}

fn map_oneshot_error(error: futures::sync::oneshot::Canceled) -> io::Error {
    io::Error::new(io::ErrorKind::Other, error)
}

impl<C: FileServeConfig> Stream for ChunkedReadFile<C> {
    type Item = Bytes;
    type Error = io::Error;

    fn poll(&mut self) -> futures::Poll<Option<Bytes>, Self::Error> {
        if self.fut.is_some() {
            return match self.fut.as_mut().unwrap().poll().map_err(map_oneshot_error)? {
                futures::Async::Ready(result) => match result {
                    Ok((file, bytes)) => {
                        self.fut.take();
                        self.file = Some(file);
                        self.offset += bytes.len() as u64;
                        self.counter += bytes.len() as u64;
                        Ok(futures::Async::Ready(Some(bytes)))
                    },
                    Err(error) => {
                        Err(error)
                    },
                }
                futures::Async::NotReady => Ok(futures::Async::NotReady),
            };
        }

        if self.size == self.counter {
            return Ok(futures::Async::Ready(None))
        }

        let size = self.size;
        let offset = self.offset;
        let counter = self.counter;

        let mut file = self.file.take().expect("Use after completion");

        let (sender, receiver) = futures::sync::oneshot::channel();
        self.thread_pool.execute(move || {
            if sender.is_canceled() {
                return
            }

            let max_bytes = cmp::min(size.saturating_sub(counter), C::max_buffer_size());
            let mut buf = Vec::with_capacity(max_bytes as usize);
            let _ = match file.seek(io::SeekFrom::Start(offset)) {
                Ok(_) => match file.by_ref().take(max_bytes).read_to_end(&mut buf) {
                    Ok(0) => sender.send(Err(io::ErrorKind::UnexpectedEof.into())),
                    Ok(_) => sender.send(Ok((file, Bytes::from(buf)))),
                    Err(error) => sender.send(Err(error)),
                },
                Err(error) => sender.send(Err(error)),
            };
        });

        self.fut = Some(receiver);

        self.poll()
    }
}