fastcgi_client/

client.rs

// Copyright 2022 jmjoy
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! FastCGI client implementation for async communication with FastCGI servers.
//!
//! This module provides the main `Client` struct that handles communication
//! with FastCGI servers in both short connection and keep-alive modes.
//! The client can execute requests and receive responses or response streams.

use crate::{
    ClientError, ClientResult, Response,
    conn::{KeepAlive, Mode, ShortConn},
    io::{self, AsyncRead, AsyncWrite, AsyncWriteExt},
    meta::{BeginRequestRec, EndRequestRec, Header, ParamPairs, RequestType, Role},
    params::Params,
    request::Request,
    response::ResponseStream,
};
use std::marker::PhantomData;
use tracing::debug;

#[cfg(feature = "tokio")]
use crate::io::{TokioAsyncReadCompatExt, TokioCompat};

/// I refer to nginx fastcgi implementation, found the request id is always 1.
///
/// <https://github.com/nginx/nginx/blob/f7ea8c76b55f730daa3b63f5511feb564b44d901/src/http/modules/ngx_http_fastcgi_module.c>
const REQUEST_ID: u16 = 1;

/// Async client for handling communication between fastcgi server.
pub struct Client<S, M> {
    stream: S,
    _mode: PhantomData<M>,
}

impl<S, M> Client<S, M> {
    fn from_stream(stream: S) -> Self {
        Self {
            stream,
            _mode: PhantomData,
        }
    }
}

#[cfg(feature = "tokio")]
impl<S> Client<TokioCompat<S>, ShortConn>
where
    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
{
    /// Construct a `Client` Object with a Tokio stream under short connection
    /// mode.
    ///
    /// Requires the optional `tokio` feature, which exists purely as a
    /// convenience adapter: Tokio defines its own I/O traits, so the stream is
    /// wrapped with [`tokio_util::compat`](crate::io::TokioAsyncReadCompatExt)
    /// before being handed to the runtime agnostic core. Without the feature,
    /// bridge the stream yourself and call [`Client::new`].
    ///
    /// # Examples
    ///
    /// ```
    /// # #[cfg(feature = "tokio")]
    /// # async fn example() {
    /// use fastcgi_client::Client;
    /// use tokio::net::TcpStream;
    /// # #[cfg(unix)]
    /// # use tokio::net::UnixStream;
    ///
    /// let tcp_stream = TcpStream::connect(("127.0.0.1", 9000)).await.unwrap();
    /// let _tcp_client = Client::new_tokio(tcp_stream);
    ///
    /// # #[cfg(unix)]
    /// # {
    /// let unix_stream = UnixStream::connect("/run/php-fpm.sock").await.unwrap();
    /// let _unix_client = Client::new_tokio(unix_stream);
    /// # }
    /// # }
    /// # #[cfg(not(feature = "tokio"))]
    /// # fn example() {}
    /// ```
    pub fn new_tokio(stream: S) -> Self {
        Self::from_stream(stream.compat())
    }
}

impl<S: AsyncRead + AsyncWrite + Unpin> Client<S, ShortConn> {
    /// Construct a `Client` object under short connection mode.
    ///
    /// This constructor is runtime agnostic: it accepts any stream implementing
    /// the [`futures_io`](crate::io) `AsyncRead` + `AsyncWrite` traits, such as
    /// `smol::net::TcpStream`, `async_net::TcpStream`, or a Tokio stream
    /// wrapped by [`tokio_util::compat`](https://docs.rs/tokio-util/latest/tokio_util/compat/index.html).
    ///
    /// # Examples
    ///
    /// ```
    /// # async fn example() {
    /// use fastcgi_client::Client;
    /// use smol::net::TcpStream;
    ///
    /// let stream = TcpStream::connect(("127.0.0.1", 9000)).await.unwrap();
    /// let _client = Client::new(stream);
    /// # }
    /// ```
    pub fn new(stream: S) -> Self {
        Self::from_stream(stream)
    }

    /// Send request and receive response from fastcgi server, under short
    /// connection mode.
    pub async fn execute_once<I: AsyncRead + Unpin>(
        mut self, request: Request<'_, I>,
    ) -> ClientResult<Response> {
        self.inner_execute(request).await
    }

    /// Send request and receive response stream from fastcgi server, under
    /// short connection mode.
    ///
    /// # Examples
    ///
    /// ```
    /// # #[cfg(feature = "tokio")]
    /// # async fn stream() {
    /// use fastcgi_client::{io, response::Content, Client, Params, Request, StreamExt};
    /// use tokio::net::TcpStream;
    ///
    ///     let stream = TcpStream::connect(("127.0.0.1", 9000)).await.unwrap();
    ///     let client = Client::new_tokio(stream);
    ///     let mut stream = client
    ///         .execute_once_stream(Request::new(Params::default(), io::empty()))
    ///         .await
    ///         .unwrap();
    ///
    ///     while let Some(content) = stream.next().await {
    ///         let content = content.unwrap();
    ///
    ///         match content {
    ///             Content::Stdout(out) => todo!(),
    ///             Content::Stderr(out) => todo!(),
    ///         }
    ///     }
    /// }
    /// # #[cfg(not(feature = "tokio"))]
    /// # fn stream() {}
    /// ```
    pub async fn execute_once_stream<I: AsyncRead + Unpin>(
        mut self, request: Request<'_, I>,
    ) -> ClientResult<ResponseStream<S>> {
        self.send_request(request).await?;
        Ok(ResponseStream::new(self.stream, REQUEST_ID))
    }
}

#[cfg(feature = "tokio")]
impl<S> Client<TokioCompat<S>, KeepAlive>
where
    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
{
    /// Construct a `Client` Object with a Tokio stream under keep-alive mode.
    ///
    /// Requires the optional `tokio` feature, which exists purely as a
    /// convenience adapter: Tokio defines its own I/O traits, so the stream is
    /// wrapped with [`tokio_util::compat`](crate::io::TokioAsyncReadCompatExt)
    /// before being handed to the runtime agnostic core. Without the feature,
    /// bridge the stream yourself and call [`Client::new_keep_alive`].
    ///
    /// # Examples
    ///
    /// ```
    /// # #[cfg(feature = "tokio")]
    /// # async fn example() {
    /// use fastcgi_client::Client;
    /// use tokio::net::TcpStream;
    /// # #[cfg(unix)]
    /// # use tokio::net::UnixStream;
    ///
    /// let tcp_stream = TcpStream::connect(("127.0.0.1", 9000)).await.unwrap();
    /// let _tcp_client = Client::new_keep_alive_tokio(tcp_stream);
    ///
    /// # #[cfg(unix)]
    /// # {
    /// let unix_stream = UnixStream::connect("/run/php-fpm.sock").await.unwrap();
    /// let _unix_client = Client::new_keep_alive_tokio(unix_stream);
    /// # }
    /// # }
    /// # #[cfg(not(feature = "tokio"))]
    /// # fn example() {}
    /// ```
    pub fn new_keep_alive_tokio(stream: S) -> Self {
        Self::from_stream(stream.compat())
    }
}

impl<S: AsyncRead + AsyncWrite + Unpin> Client<S, KeepAlive> {
    /// Construct a `Client` object under keep-alive mode.
    ///
    /// This constructor is runtime agnostic: it accepts any stream implementing
    /// the [`futures_io`](crate::io) `AsyncRead` + `AsyncWrite` traits, such as
    /// `smol::net::TcpStream`, `async_net::TcpStream`, or a Tokio stream
    /// wrapped by [`tokio_util::compat`](https://docs.rs/tokio-util/latest/tokio_util/compat/index.html).
    ///
    /// # Examples
    ///
    /// ```
    /// # async fn example() {
    /// use fastcgi_client::Client;
    /// use smol::net::TcpStream;
    ///
    /// let stream = TcpStream::connect(("127.0.0.1", 9000)).await.unwrap();
    /// let _client = Client::new_keep_alive(stream);
    /// # }
    /// ```
    pub fn new_keep_alive(stream: S) -> Self {
        Self::from_stream(stream)
    }

    /// Send request and receive response from fastcgi server, under keep alive
    /// connection mode.
    pub async fn execute<I: AsyncRead + Unpin>(
        &mut self, request: Request<'_, I>,
    ) -> ClientResult<Response> {
        self.inner_execute(request).await
    }

    /// Send request and receive response stream from fastcgi server, under
    /// keep alive connection mode.
    ///
    /// # Examples
    ///
    /// ```
    /// # #[cfg(feature = "tokio")]
    /// # async fn stream() {
    /// use fastcgi_client::{io, response::Content, Client, Params, Request, StreamExt};
    /// use tokio::net::TcpStream;
    ///
    ///     let stream = TcpStream::connect(("127.0.0.1", 9000)).await.unwrap();
    ///     let mut client = Client::new_keep_alive_tokio(stream);
    ///
    ///     for _ in (0..3) {
    ///         let mut stream = client
    ///             .execute_stream(Request::new(Params::default(), io::empty()))
    ///             .await
    ///             .unwrap();
    ///
    ///         while let Some(content) = stream.next().await {
    ///             let content = content.unwrap();
    ///
    ///             match content {
    ///                 Content::Stdout(out) => todo!(),
    ///                 Content::Stderr(out) => todo!(),
    ///             }
    ///         }
    ///     }
    /// }
    /// # #[cfg(not(feature = "tokio"))]
    /// # fn stream() {}
    /// ```
    pub async fn execute_stream<I: AsyncRead + Unpin>(
        &mut self, request: Request<'_, I>,
    ) -> ClientResult<ResponseStream<&mut S>> {
        self.send_request(request).await?;
        Ok(ResponseStream::new(&mut self.stream, REQUEST_ID))
    }
}

impl<S: AsyncRead + AsyncWrite + Unpin, M: Mode> Client<S, M> {
    /// Internal method to execute a request and return a complete response.
    ///
    /// # Arguments
    ///
    /// * `request` - The request to execute
    async fn inner_execute<I: AsyncRead + Unpin>(
        &mut self, request: Request<'_, I>,
    ) -> ClientResult<Response> {
        self.send_request(request).await?;
        handle_response(&mut self.stream, REQUEST_ID).await
    }

    /// Sends a complete request, using the `FCGI_KEEP_CONN` flag of the
    /// connection mode `M`.
    ///
    /// This is the only place where the connection mode is observed; everything
    /// else in the protocol handling is mode agnostic.
    ///
    /// # Arguments
    ///
    /// * `request` - The request to send
    async fn send_request<I: AsyncRead + Unpin>(
        &mut self, request: Request<'_, I>,
    ) -> ClientResult<()> {
        handle_request(
            &mut self.stream,
            REQUEST_ID,
            M::KEEP_CONN,
            request.params,
            request.stdin,
        )
        .await
    }
}

/// Handles the complete request process.
///
/// # Arguments
/// * `stream` - The stream to write to
/// * `id` - The request ID
/// * `keep_conn` - Value of the `FCGI_KEEP_CONN` flag
/// * `params` - The request parameters
/// * `body` - The request body stream
async fn handle_request<S: AsyncWrite + Unpin, I: AsyncRead + Unpin>(
    stream: &mut S, id: u16, keep_conn: bool, params: Params<'_>, mut body: I,
) -> ClientResult<()> {
    handle_request_start(stream, id, keep_conn).await?;
    handle_request_params(stream, id, params).await?;
    handle_request_body(stream, id, &mut body).await?;
    handle_request_flush(stream).await?;
    Ok(())
}

/// Handles the start of a request by sending the begin request record.
///
/// # Arguments
///
/// * `stream` - The stream to write to
/// * `id` - The request ID
/// * `keep_conn` - Value of the `FCGI_KEEP_CONN` flag
async fn handle_request_start<S: AsyncWrite + Unpin>(
    stream: &mut S, id: u16, keep_conn: bool,
) -> ClientResult<()> {
    debug!(id, "Start handle request");

    let begin_request_rec = BeginRequestRec::new(id, Role::Responder, keep_conn).await?;

    debug!(id, ?begin_request_rec, "Send to stream.");

    begin_request_rec.write_to_stream(stream).await?;

    Ok(())
}

/// Handles sending request parameters to the stream.
///
/// # Arguments
///
/// * `stream` - The stream to write to
/// * `id` - The request ID
/// * `params` - The request parameters
async fn handle_request_params<S: AsyncWrite + Unpin>(
    stream: &mut S, id: u16, params: Params<'_>,
) -> ClientResult<()> {
    let param_pairs = ParamPairs::new(params);
    debug!(id, ?param_pairs, "Params will be sent.");

    Header::write_to_stream_batches(
        RequestType::Params,
        id,
        stream,
        &mut &param_pairs.to_content().await?[..],
        Some(|header| {
            debug!(id, ?header, "Send to stream for Params.");
            header
        }),
    )
    .await?;

    Header::write_to_stream_batches(
        RequestType::Params,
        id,
        stream,
        &mut io::empty(),
        Some(|header| {
            debug!(id, ?header, "Send to stream for Params.");
            header
        }),
    )
    .await?;

    Ok(())
}

/// Handles sending the request body to the stream.
///
/// # Arguments
///
/// * `stream` - The stream to write to
/// * `id` - The request ID
/// * `body` - The request body stream
async fn handle_request_body<S: AsyncWrite + Unpin, I: AsyncRead + Unpin>(
    stream: &mut S, id: u16, body: &mut I,
) -> ClientResult<()> {
    Header::write_to_stream_batches(
        RequestType::Stdin,
        id,
        stream,
        body,
        Some(|header| {
            debug!(id, ?header, "Send to stream for Stdin.");
            header
        }),
    )
    .await?;

    Header::write_to_stream_batches(
        RequestType::Stdin,
        id,
        stream,
        &mut io::empty(),
        Some(|header| {
            debug!(id, ?header, "Send to stream for Stdin.");
            header
        }),
    )
    .await?;

    Ok(())
}

/// Flushes the stream to ensure all data is sent.
///
/// # Arguments
///
/// * `stream` - The stream to flush
async fn handle_request_flush<S: AsyncWrite + Unpin>(stream: &mut S) -> ClientResult<()> {
    stream.flush().await?;

    Ok(())
}

/// Handles reading and processing the response from the stream.
///
/// # Arguments
///
/// * `stream` - The stream to read from
/// * `id` - The request ID to match
async fn handle_response<S: AsyncRead + Unpin>(stream: &mut S, id: u16) -> ClientResult<Response> {
    let mut response = Response::default();

    let mut stderr = Vec::new();
    let mut stdout = Vec::new();

    loop {
        let header = Header::new_from_stream(stream).await?;
        if header.request_id != id {
            return Err(ClientError::ResponseNotFound { id });
        }
        debug!(id, ?header, "Receive from stream.");

        match header.r#type {
            RequestType::Stdout => {
                stdout.extend(header.read_content_from_stream(stream).await?);
            }
            RequestType::Stderr => {
                stderr.extend(header.read_content_from_stream(stream).await?);
            }
            RequestType::EndRequest => {
                let end_request_rec = EndRequestRec::from_header(&header, stream).await?;
                debug!(id, ?end_request_rec, "Receive from stream.");

                end_request_rec
                    .end_request
                    .protocol_status
                    .convert_to_client_result(end_request_rec.end_request.app_status)?;

                response.stdout = if stdout.is_empty() {
                    None
                } else {
                    Some(stdout)
                };
                response.stderr = if stderr.is_empty() {
                    None
                } else {
                    Some(stderr)
                };

                return Ok(response);
            }
            r#type => {
                return Err(ClientError::UnknownRequestType {
                    request_type: r#type,
                });
            }
        }
    }
}