tokio-fastcgi 1.3.0

#![warn(missing_docs)]
#![doc = include_str!("../README.md")]
//! # Basic Example
//! ```no_run
#![doc = include_str!("../examples/simple.rs")]
//! ```
use log::{trace, warn};
use std::fmt::Debug;
use std::marker::Unpin;
use std::io::{Cursor, Read, Write};
use std::collections::{HashMap, hash_map::Entry};
use std::sync::Arc;
use tokio::io::{AsyncRead, AsyncWrite, AsyncReadExt, AsyncWriteExt};
use tokio::sync::{Mutex, MutexGuard};
use std::convert::TryFrom;
use byteorder::{BigEndian, ReadBytesExt};
use std::future::Future;

/// The size of the record header is 8 bytes.
const RECORD_HEADER_SIZE: usize = 8;

/// Flag for FCGI_BeginRequestBody
const FCGI_KEEP_CONN: u8 = 0x01;

/// Static panic message for a failed lock.
const ERR_LOCK_FAILED: &str = "A request must not be processed by multiple threads.";

/// The type of the request id. This is always u16 but makes external code more readable.
type RequestId = u16;

/// Types for the parameter iterator
type ParamsIterator<'i> = dyn Iterator<Item=(&'i str, &'i [u8])> + 'i;

/// Types for the parameter iterator with string conversion
type StrParamsIterator<'i> = dyn Iterator<Item=(&'i str, Option<&'i str>)> + 'i;

/// Type returned by [`get_stdin`](Request::get_stdin) and [`get_data`](Request::get_data).
/// It makes passing around the streams easier.
pub type OwnedInStream<'a> = MutexGuard<'a, InStream>;

/// Error type for TryFrom on StdReqType and SysReqType
#[derive(Debug)]
enum TypeError {
	UnknownRecordType(u8)
}

/// Enum containing all request record types that can be handled by Request::Process.
#[derive(Clone, Copy, Debug, PartialEq)]
enum StdReqType {
	BeginRequest = 1,
	Params = 4,
	StdIn = 5,
	Data = 8
}

impl From<StdReqType> for u8 {
	fn from(rt: StdReqType) -> Self {
		rt as u8
	}
}

impl TryFrom<u8> for StdReqType {
	type Error = TypeError;
	fn try_from(value: u8) -> Result<Self, Self::Error> {
		match value {
			1 => Ok(Self::BeginRequest),
			4 => Ok(Self::Params),
			5 => Ok(Self::StdIn),
			8 => Ok(Self::Data),
			_ => Err(TypeError::UnknownRecordType(value))
		}
	}
}

/// Enum containing all response types generated by Request::Process.
#[derive(Clone, Copy, Debug, PartialEq)]
enum StdRespType {
	EndRequest = 3,
	StdOut = 6,
	StdErr = 7
}

impl From<StdRespType> for u8 {
	fn from(rt: StdRespType) -> Self {
		rt as u8
	}
}

/// Enum containing all request record types that must be handled by Request::process_sys.
#[derive(Clone, Copy, Debug, PartialEq)]
enum SysReqType {
	AbortRequest = 2,
	GetValues = 9
}

impl From<SysReqType> for u8 {
	fn from(rt: SysReqType) -> Self {
		rt as u8
	}
}

impl TryFrom<u8> for SysReqType {
	type Error = TypeError;
	fn try_from(value: u8) -> Result<Self, Self::Error> {
		match value {
			2 => Ok(Self::AbortRequest),
			9 => Ok(Self::GetValues),
			_ => Err(TypeError::UnknownRecordType(value))
		}
	}
}

/// Enum containing all response record types that can be generated by Request::process_sys.
#[derive(Clone, Copy, Debug, PartialEq)]
enum SysRespType {
	GetValuesResult = 10,
	UnknownType = 11
}

impl From<SysRespType> for u8 {
	fn from(rt: SysRespType) -> Self {
		rt as u8
	}
}

/// Container for std and sys request and response types.
#[derive(Clone, Copy, Debug)]
enum Category<S: Copy, T: Copy> {
	Std(S),
	Sys(T)
}

impl <S: Copy + TryFrom<u8, Error = TypeError>, T: Copy + TryFrom<u8, Error = TypeError>> TryFrom<u8> for Category<S, T> {
	type Error = TypeError;
	fn try_from(value: u8) -> Result<Self, Self::Error> {
		if let Ok(result) = S::try_from(value) {
			Ok(Self::Std(result))
		} else {
			T::try_from(value).map(Self::Sys)
		}
	}
}

impl <S: std::convert::Into<u8> + Copy, T: std::convert::Into<u8> + Copy> From<Category<S, T>> for u8 {
	fn from(cat: Category<S, T>) -> Self {
		match cat {
			Category::<S, T>::Std(std) => std.into(),
			Category::<S, T>::Sys(sys) => sys.into()
		}
	}
}

/// Type for all known request record types
type RequestType = Category<StdReqType, SysReqType>;

/// Type for all known response record types
type ResponseType = Category<StdRespType, SysRespType>;

/// Enum containing the role that is requested from the FastCGI client. See the different
/// variants for a description of the roles and their input and output streams.
#[derive(PartialEq, Debug)]
pub enum Role {
	/// A FastCGI responder receives all the information associated with an HTTP
	/// request and generates an HTTP response. A responder receives the following
	/// information from the web-server:
	///
	/// * Environment variables (see [`get_param`](Request::get_param)/[`get_str_param`](Request::get_str_param))
	/// * StdIn (see [`get_stdin`](Request::get_stdin))
	///
	/// A responder has the following communication channels at its disposal:
	/// * Result code (see [`RequestResult`])
	/// * StdOut (see [`get_stdout`](Request::get_stdout))
	/// * StdErr (see [`get_stderr`](Request::get_stderr))
	///
	/// see the [FastCGI specification](https://fastcgi-archives.github.io/FastCGI_Specification.html#S6.2) for more Information
	Responder,

	/// A FastCGI authorizer receives all the information associated with an HTTP
	/// request and generates an authorized/unauthorized decision. In case of an
	/// authorized decision the authorizer can also associate name-value pairs with
	/// the HTTP request. A responder receives the following information from the
	/// web-server:
	///
	/// * Environment variables and request parameters (see
	///   [`get_param`](Request::get_param)/[`get_str_param`](Request::get_str_param))
	///
	/// An authorizer has the following communication channels at its disposal:
	///
	/// * Result code (see [`RequestResult`])
	/// * StdOut (see [`get_stdout`](Request::get_stdout))
	/// * StdErr (see [`get_stderr`](Request::get_stderr))
	///
	/// see the [FastCGI specification](https://fastcgi-archives.github.io/FastCGI_Specification.html#S6.3) for more Information
	Authorizer,

	/// A FastCGI filter receives all the information associated with an HTTP
	/// request, plus an extra stream of data from a file stored on the Web server,
	/// and generates a “filtered” version of the data stream as an HTTP response. A
	/// filter receives the following information from the web-server:
	///
	/// * Environment variables, request parameters and additional information
	///   (`FCGI_DATA_LAST_MOD` and `FCGI_DATA_LENGTH`) (see
	///   [`get_param`](Request::get_param)/[`get_str_param`](Request::get_str_param))
	/// * StdIn (see [`get_stdin`](Request::get_stdin))
	/// * File Data from the web-server (see [`get_data`](Request::get_data))
	///
	/// A filter has the following communication channels at its disposal:
	///
	/// * Result code (see [`RequestResult`])
	/// * StdOut (see [`get_stdout`](Request::get_stdout))
	/// * StdErr (see [`get_stderr`](Request::get_stderr))
	///
	/// see the [FastCGI specification](https://fastcgi-archives.github.io/FastCGI_Specification.html#S6.4) for more Information
	Filter
}

impl Role {
	fn from_number(rl_num: u16) -> Option<Self> {
		match rl_num {
			1 => Some(Role::Responder),
			2 => Some(Role::Authorizer),
			3 => Some(Role::Filter),
			_ => None
		}
	}
}


/// The result of a FastCGI request.
///
/// This enum is returned by the [`process`](Request::process) method of the
///[`Request`] struct.  The meaning of the values is defined by the FastCGI
/// specification.
#[derive(Copy, Clone)]
pub enum RequestResult {
	/// The request completed successfully. The returned status value is defined by
	/// the [role](Role) of the FastCGI application.
	///
	/// # Result codes
	///
	/// The application returns the status code that the CGI program would have
	/// returned via the `exit` system call.
	///
	Complete(u32),
	/// The application ran out of resources (for example database connections). The
	/// request is rejected.
	Overloaded,
	/// The application is not prepared to handle the role requested by the
	/// web-server. For example if a FastCGI responder is called as a filter or an
	/// authorizer.
	UnknownRole
}

impl RequestResult {
	fn app_status(self) -> u32 {
		match self {
			Self::Complete(app_status) => app_status,
			_ => 0
		}
	}
}

impl From<RequestResult> for u8 {
	/// Allow the RequestResult to be converted into a u8.
	/// This method returns the magic number that must be used as the
	/// result field of the FastCGI protocol.
	fn from(rr: RequestResult) -> Self {
		match rr {
			RequestResult::Complete(_) => 0,
			RequestResult::Overloaded => 2,
			RequestResult::UnknownRole => 3
		}
	}
}

/// Errors that can be returned by calls to [`process`](Request::process).
#[derive(Debug)]
pub enum Error {
	/// The input stream was already closed and can not be reused. This indicates
	/// an error within the call sequence, like calling `process` twice or the
	/// web-server sending more data after closing `StdIn` or `Data`.
	StreamAlreadyDone,

	/// `write` was called on an output stream that was already closed by a call to
	/// `close`.
	StreamAlreadyClosed,

	/// The web-server violated the FastCGI specification. For example by sending a
	/// `StdIn` record before sending a `BeginRequest` record.
	SequenceError,

	/// The record version is not 1. This should never happen since the FastCGI
	/// specification does not define any other record versions.
	InvalidRecordVersion,

	/// The web-server sent an unknown role number. This is most likely a bug in the
	/// FastCGI implementation of the web-server.
	InvalidRoleNumber,

	/// This error is never returned to the user of the library. It is internally
	/// handled by the `tokio-fastcgi` crate. The library returns a
	/// `FCGI_UNKNOWN_TYPE` record to the web-server.
	UnknownRecordType(RequestId, u8),

	/// An IoError occurred. Most likely the connection to the web-server got lost or
	/// was interrupted. Some I/O errors are handled by `tokio-fastcgi`. If the
	/// web-server closes the FastCGI connection after all requests have been
	/// processed no error is returned and the EOF error is just swallowed.
	IoError(std::io::Error)
}

impl std::fmt::Display for Error {
	fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
		match self {
			Error::StreamAlreadyDone => write!(f, "Input stream is already done"),
			Error::StreamAlreadyClosed => write!(f, "Output stream is already closed"),
			Error::SequenceError => write!(f, "Records out of sequence "),
			Error::InvalidRecordVersion => write!(f, "Only record version 1 supported"),
			Error::InvalidRoleNumber => write!(f, "Unkown role pass from server"),
			Error::UnknownRecordType(request_id, type_id) => write!(f, "Unkown record type {} in request {} received", type_id, request_id),
			Error::IoError(error) => write!(f, "I/O error: {}", error)
		}
	}
}

impl std::error::Error for Error {
	fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
		match self {
			Error::IoError(source) => Some(source),
			_ => None
		}
	}
}

impl From<std::io::Error> for Error {
	fn from(io_error: std::io::Error) -> Self {
		Error::IoError(io_error)
	}
}

/// Represents a record received by the web-server.
struct Record {
	record_type: RequestType,
	request_id: RequestId,
	content: Vec<u8>
}

impl Record {
	async fn new<R: AsyncRead + Unpin>(rd: &mut R) -> Result<Self, Error> {
		let mut header_buffer = [0; RECORD_HEADER_SIZE];

		rd.read_exact(&mut header_buffer).await?;

		let mut header_slice = &header_buffer[..];

		// Check the FastCGI version
		if byteorder::ReadBytesExt::read_u8(&mut header_slice).unwrap() != 1 {
			return Err(Error::InvalidRecordVersion);
		}

		// Parse the remaining header fields
		// Unwrap the record_type field not yet. An error on the record_type can be handled
		// and we must read the remaining data to keep the I/O stream in sync.
		let record_type = RequestType::try_from(byteorder::ReadBytesExt::read_u8(&mut header_slice).unwrap());
		let request_id = byteorder::ReadBytesExt::read_u16::<BigEndian>(&mut header_slice)?;
		let content_length = byteorder::ReadBytesExt::read_u16::<BigEndian>(&mut header_slice).unwrap() as usize;
		let padding_length = byteorder::ReadBytesExt::read_u8(&mut header_slice).unwrap() as u64;

		// Allocate the buffer for the content and read everything asynchronously.
		// `with_capacity` can not be used, because Tokio does not support this.
		let mut content = vec![0; content_length];
		rd.read_exact(&mut content).await?;

		// If there is some padding at the end of the record. Discard it.
		if padding_length > 0 {
			tokio::io::copy(&mut rd.take(padding_length), &mut tokio::io::sink()).await?;
		}

		trace!("FastCGI: In record {{T:{:?}, ID: {}, L:{}}}", record_type, request_id, RECORD_HEADER_SIZE + content.len() + padding_length as usize);

		// Now we unwrap the record_type. If we fail now, the record as been completely read.
		// Before we can unwrap the TypeError must be translated into a full blown Error::UnknownRecordType value by adding the request_id.
		let record_type = record_type.map_err(|error| {
			let TypeError::UnknownRecordType(record_type_nr) = error;
			Error::UnknownRecordType(request_id, record_type_nr)
		})?;

		Ok(Self {
			record_type,
			request_id,
			content
		})
	}

	/// Checks if this record is a system record. If that's the case Request::update should not be called
	/// on this one. Just call Request::process_sys to process the system record.
	/// This method only returns true if the record type can be processed by Request::sys_process and
	/// the Record is complete.
	fn is_sys_record(&self) -> bool {
		matches!(self.record_type, Category::Sys(_))
	}

	fn get_content(&self) -> &[u8] {
		&self.content
	}

	fn get_request_id(&self) -> RequestId {
		self.request_id
	}
}

/// Implements a data stream from the web-server to the FastCGI application.
///
/// All data is buffered in memory before being returned to the FastCGI
/// application. Therefore only a synchronous interface is implemented.
///
/// The data of the stream can be accessed via the methods of the [`Read`
/// trait](Read).
#[derive(Debug)]
pub struct InStream {
	data: Vec<u8>,
	read_pos: Option<usize>
}

impl Read for InStream {
	/// Read implementation for Stream.
	///
	/// *Beware*: Calling read or read_exact on a stream that is not done will panic!
	fn read(&mut self, out: &mut [u8]) -> std::result::Result<usize, std::io::Error> {
		let read_pos = self.read_pos.unwrap();
		let c = std::io::Read::read(&mut &self.data[read_pos..], out)?;
		self.read_pos = Some(read_pos + c);
		Ok(c)
	}

	/// Read_exact implementation for Stream.
	///
	/// *Beware*: Calling read or read_exact on a stream that is not done will panic!
	fn read_exact(&mut self, out: &mut [u8]) -> std::result::Result<(), std::io::Error> {
		let read_pos = self.read_pos.unwrap();
		std::io::Read::read_exact(&mut &self.data[read_pos..], out)?;
		self.read_pos = Some(read_pos + out.len());
		Ok(())
	}
}

impl InStream{
	/// Creates a new stream waiting for input.
	///
	/// If `already_done` is set to true the stream is created done and no data can be added to it.
	fn new(already_done: bool) -> Self {
		InStream {
			data: Vec::new(),
			read_pos: if already_done { Some(0) } else { None }
		}
	}

	/// Appends the passed data slice to the internal vector.
	///
	/// If the slice is empty, this is interpreted as an EOF marker and marks this
	/// stream as done.
	/// Calling this method on a done stream will always return a StreamAlreadyDone
	/// error.
	fn append(&mut self, data: &[u8]) -> Result<(), Error> {
		if ! data.is_empty() {
			if self.read_pos.is_none() {
				self.data.extend_from_slice(data);
				Ok(())
			} else {
				Err(Error::StreamAlreadyDone)
			}
		} else {
			self.read_pos = Some(0);
			Ok(())
		}
	}

	/// Checks if this stream is done (EOF).
	///
	/// This is signaled by the web server by passing an empty record for this stream.
	fn is_done(&self) -> bool {
		self.read_pos.is_some()
	}
}

/// Represents a FastCGI request that can be handled by the application.
///
/// An instance of this struct is returned by the [`next`](Requests::next) function
/// of the [Requests] struct. It represents one request that should be handled
/// via FastCGI. Normally [`process`](Request::process) is called on every
/// instance that is returned. The request gets passed to the callback function
/// and can be used to get the input/output streams and environment values.
pub struct Request <W: AsyncWrite + Unpin> {
	/// Contains the role that this request is requesting from the FastCGI
	/// application.
	///
	/// If the FastCGI application can not comply to this role the callback
	/// passed to [`process`](Request::process) should return
	/// [`RequestResult::UnknownRole`].
	pub role: Role,
	keep_connection: bool,
	request_id: RequestId,
	params: HashMap<String, Vec<u8>>,
	params_done: bool,
	orw: Arc<OutRecordWriter<W>>,
	stdin: Mutex<InStream>,
	data: Mutex<InStream>
}

impl <W: AsyncWrite + Unpin> Request<W> {
	fn new(record: &Record, writer: Arc<Mutex<W>>) -> Result<Self, Error> {
		let mut content = record.get_content();

		if let Category::Std(StdReqType::BeginRequest) = record.record_type {
			if let Some(role) = Role::from_number(byteorder::ReadBytesExt::read_u16::<BigEndian>(&mut content).unwrap()) { //We're reading from am memory buffer. So there is something deeply wrong if this fails.
				let keep_connection = (byteorder::ReadBytesExt::read_u8(&mut content)? & FCGI_KEEP_CONN) == FCGI_KEEP_CONN;

				Ok(Self {
					params: HashMap::new(),
					params_done: false,
					orw: Arc::from(OutRecordWriter::new(writer, record.request_id)),
					stdin: Mutex::from(InStream::new(role == Role::Authorizer)), // Authorizers do not get an stdin stream
					data: Mutex::from(InStream::new(role != Role::Filter)),      // Only filters get a data stream
					role,
					keep_connection,
					request_id: record.request_id
				})
			} else {
				Err(Error::InvalidRoleNumber)
			}
		} else {
			Err(Error::SequenceError)
		}
	}

	fn read_length<T: Read>(src: &mut T) -> Result<u32, std::io::Error> {
		let length: u32 = u32::from(src.read_u8()?);

		if length & 0x80 == 0 {
			Ok(length)
		} else {
			let length_byte2 = u32::from(src.read_u8()?);
			let length_byte10 = u32::from(src.read_u16::<BigEndian>()?);

			Ok((length & 0x7f) << 24 | length_byte2 << 16 | length_byte10)
		}
	}

	fn add_nv_pairs(params: &mut HashMap<String, Vec<u8>>, src: &[u8], lowercase_keys: bool) -> Result<(), std::io::Error>{
		let mut src_slice = src;

		while !src_slice.is_empty() {
			let name_length = Request::<W>::read_length(&mut src_slice)?;
			let value_length = Request::<W>::read_length(&mut src_slice)?;

			let mut name_buffer = vec![0; name_length as usize];
			let mut value_buffer = vec![0; value_length as usize];

			std::io::Read::read_exact(&mut src_slice, &mut name_buffer)?;
			std::io::Read::read_exact(&mut src_slice, &mut value_buffer)?;

			let key = String::from_utf8_lossy(&name_buffer);
			let key = if lowercase_keys {
				key.to_ascii_lowercase()
			} else {
				key.into_owned()
			};

			trace!("FastCGI: NV-Pair[\"{}\"]=\"{}\"", key, String::from_utf8_lossy(&value_buffer));

			params.insert(key, value_buffer);
		}

		Ok(())
	}

	/// Returns the parameter with the given name as a byte vector.
	///
	/// Parameters are passed to the FastCGI application as name value pairs.
	/// Parameters can contain environment variables or other parameters that
	/// the web-server wants to pass to the application.
	///
	/// If the parameter does not exist `None` is returned.
	///
	/// ## Example
	///
	/// ```rust
	/// # use tokio::io::{empty, sink};
	/// # use tokio_fastcgi::{Requests, RequestResult};
	/// # #[tokio::main]
	/// # async fn main() {
	/// # let mut requests = Requests::new(empty(), sink(), 1, 1);
	/// # if let Some(request) = requests.next().await.expect("Request could not be constructed.") {
	/// request.process(|request| async move {
	///   if let Some(binary_data) = request.get_param("BINARY_DATA") {
	///     assert_eq!(binary_data, &[10, 20, 30]);
	///   }
	///
	///   RequestResult::Complete(0)
	/// });
	/// # } }
	/// ```
	pub fn get_param(&self, name: &str) -> Option<&Vec<u8>> {
		if self.params_done {
			self.params.get(&name.to_ascii_lowercase())
		} else {
			None
		}
	}

	/// Returns the parameter with the given name as a UTF-8 string.
	///
	/// Parameters are passed to the FastCGI application as name value pairs.
	/// Parameters can contain environment variables or other parameters that
	/// the web-server wants to pass to the application.
	///
	/// If the parameter does not exist or is not valid UTF-8 `None` is returned.
	///
	/// ## Example
	///
	/// ```rust
	/// # use tokio::io::{empty, sink};
	/// # use tokio_fastcgi::{Requests, RequestResult};
	/// # #[tokio::main]
	/// # async fn main() {
	/// # let mut requests = Requests::new(empty(), sink(), 1, 1);
	/// # if let Some(request) = requests.next().await.expect("Request could not be constructed.") {
	/// request.process(|request| async move {
	///   if let Some(uri) = request.get_str_param("REQUEST_URI") {
	///     assert_eq!(uri, "/index.html");
	///   }
	///
	///   RequestResult::Complete(0)
	/// });
	/// # } }
	/// ```
	pub fn get_str_param(&self, name: &str) -> Option<&str> {
		if self.params_done {
			match self.params.get(&name.to_ascii_lowercase()).map(|v| std::str::from_utf8(v)) {
				None => None,
				Some(Ok(value)) => Some(value),
				Some(Err(_)) => {
					warn!("FastCGI: Parameter {} is not valid utf8.", name);
					None
				}
			}
		} else {
			None
		}
	}

	/// Returns an iterator over all parameters.
	///
	/// The parameter value is a [u8] slice containing the raw data for the parameter.
	/// If you need the parameter values as string, take a look at [str_params_iter](Request::str_params_iter).
	///
	/// ## Example
	///
	/// ```rust
	/// # use tokio::io::{empty, sink};
	/// # use tokio_fastcgi::{Requests, RequestResult};
	/// # #[tokio::main]
	/// # async fn main() {
	/// # let mut requests = Requests::new(empty(), sink(), 1, 1);
	/// # if let Some(request) = requests.next().await.expect("Request could not be constructed.") {
	/// request.process(|request| async move {
	///   if let Some(params) = request.params_iter() {
	///     // Output a list of all parameters
	///     for param in params {
	///       println!("{}: {:?}", param.0, param.1);
	///     }
	///   }
	///
	///   RequestResult::Complete(0)
	/// });
	/// # } }
	/// ```
	pub fn params_iter(&self) -> Option<Box<ParamsIterator>> {
		if self.params_done {
			Some(Box::new(self.params.iter().map(|v| {
				(v.0.as_str(), &v.1[..])
			})))
		} else {
			None
		}
	}

	/// Returns an iterator over all parameters that tries to convert the parameter
	/// values into strings.
	///
	/// The parameter value is an [Option] containing a [String] reference.
	/// If the parameter could not be converted into a string (because it is not valid UTF8)
	/// the [Option] will be [None](Option::None).
	///
	/// ## Example
	///
	/// ```rust
	/// # use tokio::io::{empty, sink};
	/// # use tokio_fastcgi::{Requests, RequestResult};
	/// # #[tokio::main]
	/// # async fn main() {
	/// # let mut requests = Requests::new(empty(), sink(), 1, 1);
	/// # if let Some(request) = requests.next().await.expect("Request could not be constructed.") {
	/// request.process(|request| async move {
	///   if let Some(params) = request.str_params_iter() {
	///     // Output a list of all parameters
	///     for param in params {
	///       println!("{}: {}", param.0, param.1.unwrap_or("[Invalid UTF8]"));
	///     }
	///   }
	///
	///   RequestResult::Complete(0)
	/// });
	/// # } }
	/// ```
	pub fn str_params_iter(&self) -> Option<Box<StrParamsIterator>> {
		if self.params_done {
			Some(Box::new(self.params.iter().map(|v| {
				(v.0.as_str(), std::str::from_utf8(v.1).ok())
			})))
		} else {
			None
		}
	}

	/// Checks if this record is ready for processing by the client application.
	/// A record is ready if the stdin, the data and the params stream are done (EOF).
	fn check_ready(&mut self) -> bool {
		self.get_stdin().is_done() && self.get_data().is_done() && self.params_done
	}

	/// Updates the state of the Request instance.
	/// If the Request instance is ready for processing by the client application this method will
	/// return true.
	/// Calling update on an already ready request Err(Error::SequenceError) is returned.
	fn update(&mut self, record: &Record) -> Result<bool, Error> {
		assert!(record.request_id == self.request_id);

		if self.check_ready() {
			return Err(Error::SequenceError);
		}

		if let Category::Std(record_type) = record.record_type {
			match record_type {
				StdReqType::BeginRequest => {
					return Err(Error::SequenceError);
				},

				StdReqType::Params => {
					if record.content.is_empty() {
						self.params_done = true;
					} else {
						if self.params_done { warn!("FastCGI: Protocol error. Params received after params stream was marked as done."); }

						Self::add_nv_pairs(&mut self.params, record.get_content(), true)?;
					}
				},

				StdReqType::StdIn => {
					self.get_stdin().append(record.get_content())?;
				},

				StdReqType::Data => {
					self.get_data().append(record.get_content())?;
				}
			};

			Ok(self.check_ready())
		} else {
			Err(Error::SequenceError)
		}
	}

	/// Returns the request id of this request.
	///
	/// This id is unique within the current connection. It is managed by the
	/// web-server.
	pub fn get_request_id(&self) -> RequestId {
		self.request_id
	}

	/// Allows the process closure to write to StdOut.
	///
	/// Returns an `OutStream` instance that will send `StdOut` records back to
	/// the web-server.
	///
	///
	/// ## Example
	///
	/// ```rust
	/// # use tokio::io::{empty, sink};
	/// # use tokio_fastcgi::{Requests, RequestResult};
	/// # #[tokio::main]
	/// # async fn main() {
	/// # let mut requests = Requests::new(empty(), sink(), 1, 1);
	/// # if let Some(request) = requests.next().await.expect("Request could not be constructed.") {
	/// request.process(|request| async move {
	///   let mut stdout = request.get_stdout();
	///
	///   assert!(stdout.write(b"Hello World").await.is_ok());
	///
	///   RequestResult::Complete(0)
	/// });
	/// # } }
	/// ```
	pub fn get_stdout(&self) -> OutStream<W> {
		OutStream::new(Category::Std(StdRespType::StdOut), self.orw.clone())
	}

	/// Allows the process closure to write to StdErr.
	///
	/// Returns an `OutStream` instance that will send `StdErr` records back to
	/// the web-server. What is done with the data that is sent to StdErr depends
	/// on the web-server.
	///
	/// ## Example
	///
	/// ```rust
	/// # use tokio::io::{empty, sink, AsyncWriteExt};
	/// # use tokio_fastcgi::{Requests, RequestResult};
	/// # #[tokio::main]
	/// # async fn main() {
	/// # let mut requests = Requests::new(empty(), sink(), 1, 1);
	/// # if let Some(request) = requests.next().await.expect("Request could not be constructed.") {
	/// request.process(|request| async move {
	///   let mut stderr = request.get_stderr();
	///
	///   assert!(stderr.write(b"Hello World").await.is_ok());
	///
	///   RequestResult::Complete(0)
	/// });
	/// # } }
	/// ```
	pub fn get_stderr(&self) -> OutStream<W> {
		OutStream::new(Category::Std(StdRespType::StdErr), self.orw.clone())
	}

	/// Allows the process closure to read from StdIn.
	///
	/// Returns an `InStream` instance that will read the data passed as StdIn
	/// by the web-server.
	///
	/// ## Example
	///
	/// ```rust
	/// # use tokio::io::{empty, sink};
	/// # use std::io::Read;
	/// # use tokio_fastcgi::{Requests, RequestResult};
	/// # #[tokio::main]
	/// # async fn main() {
	/// # let mut requests = Requests::new(empty(), sink(), 1, 1);
	/// # if let Some(request) = requests.next().await.expect("Request could not be constructed.") {
	/// request.process(|request| async move {
	///   let mut stdin = request.get_stdin();
	///
	///   let mut buffer = Vec::with_capacity(10);
	///   assert!(stdin.read(&mut buffer).is_ok());
	///
	///   assert_eq!(buffer.len(), 10);
	///
	///   RequestResult::Complete(0)
	/// });
	/// # } }
	/// ```
	pub fn get_stdin(&self) -> OwnedInStream {
		self.stdin.try_lock().expect(ERR_LOCK_FAILED)
	}

	/// Allows the process closure to read from the Data stream.
	///
	/// Returns an `InStream` instance that will read the data passed as a Data
	/// stream by the web-server.
	///
	/// ## Example
	///
	/// ```rust
	/// # use tokio::io::{empty, sink};
	/// # use std::io::Read;
	/// # use tokio_fastcgi::{Requests, RequestResult};
	/// # #[tokio::main]
	/// # async fn main() {
	/// # let mut requests = Requests::new(empty(), sink(), 1, 1);
	/// # if let Some(request) = requests.next().await.expect("Request could not be constructed.") {
	/// request.process(|request| async move {
	///   let mut data = request.get_data();
	///
	///   let mut buffer = Vec::with_capacity(10);
	///   assert!(data.read(&mut buffer).is_ok());
	///
	///   assert_eq!(buffer.len(), 10);
	///
	///   RequestResult::Complete(0)
	/// });
	/// # } }
	/// ```
	pub fn get_data(&self) -> OwnedInStream {
		self.data.try_lock().expect(ERR_LOCK_FAILED)
	}

	/// Processes a FastCGI request.
	///
	/// As soon as a request is completely received it is returned by
	/// [`Requests::next`]. Calling `process` on this request allows the request
	/// to be processed. The application logic is passed to `process` via a
	/// callback function.
	///
	/// The callback function gets a reference to the [`Request`] instance that
	/// contains all necessary information (input-/output-streams, parameters,
	/// etc.) for processing the request.
	///
	/// See the examples directory for a complete example for using this function.
	///
	/// ## Callback function
	///
	/// The callback function can access all information about the request via
	/// the passed `request` parameter. The return value can be one of the
	/// following values:
	///
	/// - [`RequestResult::Complete`]
	/// - [`RequestResult::Overloaded`]
	/// - [`RequestResult::UnknownRole`]
	///
	/// ## Example
	///
	/// ```rust
	/// # use tokio::io::{empty, sink};
	/// # use std::io::Read;
	/// # use tokio_fastcgi::{Requests, RequestResult};
	/// # #[tokio::main]
	/// # async fn main() {
	/// # let instream = empty();
	/// # let outstream = sink();
	/// let mut requests = Requests::new(instream, outstream, 1, 1);
	///
	/// while let Some(request) = requests.next().await.expect("Request could not be constructed.") {
	///   request.process(|request| async move {
	///
	///     // Process request
	///
	///     RequestResult::Complete(0)
	///   });
	/// }
	/// # }
	/// ```
	pub async fn process<F: Future<Output = RequestResult>, C: FnOnce(Arc<Self>) -> F>(self, callback: C) -> Result<(), Error> {
		let rc_self = Arc::from(self);

		let result = callback(rc_self.clone()).await;

		if let Ok(this) = Arc::try_unwrap(rc_self) {
			this.get_stdout().close().await?;
			this.get_stderr().close().await?;

			this.orw.write_finish(result).await?;
		} else {
			panic!("StdErr or StdOut leaked out of process.")
		}

		Ok(())
	}
}

/// Processes records form an input and output stream.
///
/// FastCGI allow multiple requests to be interleaved within one data-stream.
/// This struct reads the FastCGI-records from an input stream and assembles
/// them into a [`Request`].
///
/// *Beware*: Requests are built in memory. Having huge requests can eat up all
/// of your systems memory.
pub struct Requests <R: AsyncRead + Unpin + Send, W: AsyncWrite + Unpin + Send> {
	reader: R,
	writer: Arc<Mutex<W>>,
	requests: HashMap<RequestId, Request<W>>,
	close_on_next: bool,
	max_conns: u8,
	max_reqs: u8
}

impl <'w, R: AsyncRead + Unpin + Send, W: AsyncWrite + Unpin + Send> Requests<R, W> {
	/// Creates a new [`Requests`] instance.
	///
	/// As soon as a new connection is accepted the read and write parts of this
	/// connection should be passed to this function. It will create a new
	/// [`Requests`] instance that will handle the communication between the
	/// web-server and the FastCGI application.
	///
	/// In addition to the read and write side of the connection two more
	/// parameters must be passed:
	///
	/// - max_conns \
	///   Maximum number of concurrent connections. This value will be returned
	///   to the web-server to allow it to adjust its connection handling.
	/// - max_reqs \
	///   Maximum number of concurrent requests. Concurrent requests are
	///   handled by tokio-fastcgi but they consume memory. This value is used
	///   to tell the web-server how many concurrent requests he can use per
	///   connection.
	pub fn new(rd: R, wr: W, max_conns: u8, max_reqs: u8) -> Self {
		Self {
			requests: HashMap::with_capacity(1),
			reader: rd,
			writer: Arc::from(Mutex::from(wr)),
			close_on_next: false,
			max_conns,
			max_reqs
		}
	}

	/// Same as [`new`](Requests::new) but takes a tuple containing the read and write
	/// side of the socket instead of two distinct variables
	///
	/// This is more convenient in combination with the `split` function.
	///
	/// # Example
	///
	/// ```rust
	/// # use tokio::net::TcpListener;
	/// # use tokio_fastcgi::{Requests, RequestResult};
	/// # #[tokio::main]
	/// # async fn main() {
	/// # let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
	/// # let server = async move {
	/// if let Ok(mut socket) = listener.accept().await {
	///   tokio::spawn(async move {
	///   // Directly use the result of the split() call to construct a Requests instance
	///     let mut requests = Requests::from_split_socket(socket.0.split(), 10, 10);
	///
	///     // Process the requests
	///   });
	/// }
	/// # }; }
	/// ```
	pub fn from_split_socket(split_socket: (R, W), max_conns: u8, max_reqs: u8) -> Self {
		Self::new(split_socket.0, split_socket.1, max_conns, max_reqs)
	}

	/// Processes and answers system records.
	/// If this method returns Error::Canceled an FCGI_END_REQUEST was already sent. Just discard the Request instance
	/// assigned for this connection.
	async fn process_sys(&self, record: Record) -> Result<Option<RequestId>, Error> {
		if let Category::Sys(record_type) = record.record_type {
			let output_stream = OutRecordWriter::new(self.writer.clone(), record.request_id);

			let result = match record_type {
				SysReqType::GetValues => {
					let mut params = HashMap::new();

					//TODO: Is this function correctly placed in request?
					Request::<W>::add_nv_pairs(&mut params, record.get_content(), false)?;

					// If we're testing this library we have to make sure that the output is sorted.
					// Otherwise the binary compare of the produced FastCGI response is not stable.
					// For production we will not do this, because it is an unnecessary performance bottleneck.
					#[cfg(debug_assertions)]
					let mut params: Vec<(String, _)> = params.into_iter().collect();
					#[cfg(debug_assertions)]
					params.sort_by(|a, b| { a.0.cmp(&b.0) });

					// Construct a vector containing the known parameters.
					// All other parameters are simply ignored.
					let mut output = Vec::with_capacity(128);
					for (name, _) in params {
						let result = match &*name {
							"FCGI_MAX_CONNS" => Some(self.max_conns),
							"FCGI_MAX_REQS" => Some(self.max_reqs),
							"FCGI_MPXS_CONNS" => Some(1),
							_ => None
						};

						if let Some(result) = result {
							let result_str = result.to_string();
							// We can get away with casting here because we know that the names and values can not get longer than what fits into 7 Bits of an u8.
							Write::write_all(&mut output, &[name.len() as u8])?;
							Write::write_all(&mut output, &[result_str.len() as u8])?;

							Write::write_all(&mut output, name.as_bytes())?;
							Write::write_all(&mut output, result_str.as_bytes())?;
						}
					}

					output_stream.write_data(Category::Sys(SysRespType::GetValuesResult), &output[..]).await?;

					Ok(None)
				},
				SysReqType::AbortRequest => {
					output_stream.write_finish(RequestResult::Complete(0)).await?;

					Ok(Some(record.get_request_id()))
				}
			};

			output_stream.flush().await?;

			result
		} else {
			panic!("process_sys called with non sys record.");
		}
	}

	/// Fetches the next request from this connection
	///
	/// This function asynchronously fetches FastCGI records and assembles them
	/// into requests. It does the de-interlacing to allow multiple requests to
	/// be processed in parallel. As soon as the information for a request is
	/// complete, it returns a [`Request`] instance for further processing.
	///
	/// This function will do the book keeping and process system requests like
	/// `FCGI_GET_VALUES` or `FCGI_ABORT_REQUEST`.
	pub async fn next(&mut self) -> Result<Option<Request<W>>, Error> {
		if self.close_on_next {
			if !self.requests.is_empty() {
				warn!("FastCGI: The web-server interleaved requests on this connection but did not use the FCGI_KEEP_CONN flag. {} requests will get lost.", self.requests.len());
			}

			// Signal to the user that this connection should be closed.
			Ok(None)
		} else {
			loop
			{
				match Record::new(&mut self.reader).await {
					// Success, a new record hast to be added to its request...
					Ok(record) => {
						if record.is_sys_record() {
							if let Some(canceled_request_id) = self.process_sys(record).await? {
								// The request got canceled. Remove it from the list
								self.requests.remove(&canceled_request_id);
							}
						} else {
							let request_ready = match self.requests.entry(record.get_request_id()) {
								Entry::Occupied(mut e) => { e.get_mut().update(&record) },
								Entry::Vacant(e) => { e.insert(Request::new(&record, self.writer.clone())?); Ok(false) }
							}?;

							if request_ready {
								let request = self.requests.remove(&record.get_request_id()).unwrap();

								// Store if we should close the connection after handling this request.
								self.close_on_next = !request.keep_connection;

								// Calling unwrap here is ok because we made sure there is an object for this id.
								return Ok(Some(request));
							}
						}
					},
					// IoError UnexpectedEof: May be ok or an error. Depends on if requests have been processed.
					Err(Error::IoError(err)) if err.kind() == std::io::ErrorKind::UnexpectedEof => {
						// An I/O-error signals the end of the stream. On record construction this is ok as long
						// as there are no other requests in flight.
						if self.requests.is_empty() {
							return Ok(None)
						} else {
							return Err(Error::from(err));
						}
					},
					// UnknownRecordType must be transmitted back to the server. This is not a fatal error.
					Err(Error::UnknownRecordType(request_id, type_id)) => {
						let output_stream = OutRecordWriter::new(self.writer.clone(), request_id);
						output_stream.write_unkown_type(type_id).await?;
					},
					// An error occurred, exit the loop and return it...
					Err(err) => {
						return Err(err);
					}
				}
			}
		}
	}
}

// Generate nicer debug output for Request. This is useful if you look at the request
// from within the `process` function.
impl <W: AsyncWrite + Unpin> Debug for Request<W> {
	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
		write!(f, "Request {{ request_id: {}, keep_connection: {:?}, stdin: {:?}, data: {:?}, params: {{", self.request_id, self.keep_connection, self.stdin, self.data)?;

		for (param_index, param_key) in self.params.keys().enumerate() {
			let delimiter = if param_index > 0 { ", " } else { "" };

			if let Some(str_value) = self.get_str_param(param_key) {
				write!(f, "{}{}: \"{}\"", delimiter, param_key, str_value)?;
			} else {
				write!(f, "{}{}: {:?}", delimiter, param_key, self.get_param(param_key))?;
			}
		}

		writeln!(f, "}}")
	}
}

/// Sends output records to the web-server.
#[derive(Debug)]
struct OutRecordWriter<W: AsyncWrite> {
	inner_stream: Arc<Mutex<W>>,
	request_id: RequestId,
}

impl <W: AsyncWrite + Unpin> OutRecordWriter<W> {
	fn new(inner_stream: Arc<Mutex<W>>, request_id: RequestId) -> Self {
		Self {
			inner_stream,
			request_id
		}
	}

	async fn write_data(&self, record_type: ResponseType, data: &[u8]) -> std::result::Result<usize, std::io::Error> {
		trace!("FastCGI: Out record {{T:{:?}, ID: {}, L:{}}}", record_type, self.request_id, RECORD_HEADER_SIZE + data.len());

		// Construct the header
		// We use unwrap here because we're writing to a vec. This must never fail.
		let mut message_header = Vec::with_capacity(8);
		byteorder::WriteBytesExt::write_u8(&mut message_header, 1).unwrap();                                // Version
		byteorder::WriteBytesExt::write_u8(&mut message_header, record_type.into()).unwrap();               // Record Type
		byteorder::WriteBytesExt::write_u16::<BigEndian>(&mut message_header, self.request_id).unwrap();    // Request ID
		byteorder::WriteBytesExt::write_u16::<BigEndian>(&mut message_header, data.len() as u16).unwrap();  // Content length
		byteorder::WriteBytesExt::write_u8(&mut message_header, 0).unwrap();                                // No padding.
		byteorder::WriteBytesExt::write_u8(&mut message_header, 0).unwrap();                                // Reserved

		// Aquire the mutext guard to prevent the header and the payload to pe torn apart.
		let mut is = self.inner_stream.try_lock().expect(ERR_LOCK_FAILED);

		// Write the messge header
		is.write_all_buf(&mut Cursor::new(message_header)).await?;

		// Write the data
		// Writing empty data blocks breaks tokio-test. Therefore we only call write if the data-buffer is not empty.
		if !data.is_empty() {
			is.write_all_buf(&mut Cursor::new(data)).await?;
			Ok(data.len())
		} else {
			Ok(0)
		}

		// Write no padding
	}

	/// Sends an `EndRequest` response to the web-server and ends the current
	/// request.
	async fn write_finish(&self, result: RequestResult) -> Result<(), std::io::Error> {
		let mut end_message = Vec::with_capacity(8);

		// Unwrap is safe here because we're writing to an in memory buffer. This must never fail.
		byteorder::WriteBytesExt::write_u32::<BigEndian>(&mut end_message, result.app_status()).unwrap();
		byteorder::WriteBytesExt::write_u8(&mut end_message, result.into()).unwrap();
		// Write 3 reserved bytes
		std::io::Write::write_all(&mut end_message, &[0u8; 3]).unwrap();

		self.write_data(Category::Std(StdRespType::EndRequest), &end_message[..]).await?;

		Ok(())
	}

	/// Sends an `UnknownType` response to the web-server.
	async fn write_unkown_type(&self, type_id: u8) -> Result<(), std::io::Error> {
		let mut ut_message = Vec::with_capacity(8);

		// Unwrap is safe here because we're writing to an in memory buffer. This must never fail.
		byteorder::WriteBytesExt::write_u8(&mut ut_message, type_id).unwrap();
		// Write 7 reserved bytes
		std::io::Write::write_all(&mut ut_message, &[0u8; 7]).unwrap();

		self.write_data(ResponseType::Sys(SysRespType::UnknownType), &ut_message[..]).await?;

		Ok(())
	}

	async fn flush(&self) -> std::result::Result<(), std::io::Error> {
		self.inner_stream.try_lock().expect(ERR_LOCK_FAILED).flush().await
	}
}

/// Implements a data stream from the FastCGI application to the web-server.
///
/// The maximum chunk size is 64k. The calls made by this
/// interface may block if the web-server is not receiving the data fast enough.
/// Therefore all calls are implemented as async functions.
pub struct OutStream<W: AsyncWrite + Unpin> {
	orw: Arc<OutRecordWriter<W>>,
	record_type: ResponseType,
	closed: bool
}

impl <W: AsyncWrite + Unpin> OutStream<W> {
	fn new(record_type: ResponseType, orw: Arc<OutRecordWriter<W>>) -> Self {
		Self {
			orw,
			record_type,
			closed: false
		}
	}

	/// Send data to the web-server.
	///
	/// If the data is bigger than 64k the transfer is automatically split into
	/// chunks of 64k.
	/// If the stream is already closed, the function will always return
	/// [`StreamAlreadyClosed`](Error::StreamAlreadyClosed).
	pub async fn write(&mut self, data: &[u8]) -> std::result::Result<usize, Error> {
		if self.closed {
			return Err(Error::StreamAlreadyClosed);
		}

		// Check if the data can be transmitted in one chunk.
		// If not, split the data in chunks of u16 - 1 size.
		if data.len() < u16::max_value() as usize {
			Ok(self.orw.write_data(self.record_type, data).await?)
		} else {
			// Transmit large streams in junks of 64k
			const JUNK_SIZE: usize = (u16::max_value() - 1) as usize;
			for offset in (0..data.len()).step_by(JUNK_SIZE) {
				self.orw.write_data(self.record_type, &data[offset..(offset + JUNK_SIZE).min(data.len())]).await?;
			}

			Ok(data.len())
		}
	}

	/// Flushes the data to the web-server immediately.
	///
	/// This function also calls flush on the underlying stream.
	pub async fn flush(&self) -> std::result::Result<(), std::io::Error> {
		self.orw.flush().await
	}

	/// Closes the output stream
	///
	/// FastCGI closes a stream by sending an empty packet. After calling this
	/// method, further calls to [`write`] will fail with
	/// [`StreamAlreadyClosed`](Error::StreamAlreadyClosed).
	async fn close(&mut self) -> Result<(), Error>{
		// Send an empty record to close the stream.
		self.write(&[0u8; 0]).await?;

		self.flush().await?;

		// Now mark this stream as closed. Do not do it any earlier, because
		// we need to call write on the stream to close it.
		self.closed = true;

		Ok(())
	}
}

#[cfg(test)]
mod tests {
	use super::*;
	use tokio_test::io::Builder;

	fn is_send<T: Send>(_: T) { }

	/// Verify that the future created by process is Send to allow using it
	/// with Tokio.
	#[test]
	fn check_send() {
		is_send(async move {
			let mut requests = Requests::new(Builder::new().build(), Builder::new().build(), 10, 10);

			is_send(&requests);

			while let Ok(Some(request)) = requests.next().await {
				request.process(|_request| async move {
					RequestResult::Complete(0)
				}).await.unwrap();
			}
		});
	}
}