Struct warp::Server

pub struct Server<F> { /* private fields */ }

A Warp Server ready to filter requests.

Implementations

impl<F> Server<F> where
    F: Filter + Clone + Send + Sync + 'static,
    <F::Future as TryFuture>::Ok: Reply,
    <F::Future as TryFuture>::Error: IsReject, 

pub async fn run(self, addr: impl Into<SocketAddr>)

Run this Server forever on the current thread.

Examples found in repository

examples/returning.rs (line 19)

async fn main() {
    let routes = index_filter().or(assets_filter());
    warp::serve(routes).run(([127, 0, 0, 1], 3030)).await;
}

examples/dyn_reply.rs (line 16)

async fn main() {
    let routes = warp::path::param().and_then(dyn_reply);

    warp::serve(routes).run(([127, 0, 0, 1], 3030)).await;
}

examples/dir.rs (line 8)

async fn main() {
    pretty_env_logger::init();

    warp::serve(warp::fs::dir("examples/dir"))
        .run(([127, 0, 0, 1], 3030))
        .await;
}

examples/hello.rs (line 9)

async fn main() {
    // Match any request and return hello world!
    let routes = warp::any().map(|| "Hello, World!");

    warp::serve(routes).run(([127, 0, 0, 1], 3030)).await;
}

examples/futures.rs (line 15)

async fn main() {
    // Match `/:Seconds`...
    let routes = warp::path::param()
        // and_then create a `Future` that will simply wait N seconds...
        .and_then(sleepy);

    warp::serve(routes).run(([127, 0, 0, 1], 3030)).await;
}

examples/wrapping.rs (line 30)

async fn main() {
    // Match any request and return hello world!
    let routes = warp::any()
        .map(|| "hello world")
        .boxed()
        .recover(|_err| async { Ok("recovered") })
        // wrap the filter with hello_wrapper
        .with(warp::wrap_fn(hello_wrapper));

    warp::serve(routes).run(([127, 0, 0, 1], 3030)).await;
}

Additional examples can be found in:

• examples/file.rs
• examples/headers.rs
• examples/todos.rs
• examples/body.rs
• examples/sse.rs
• examples/websockets.rs
• examples/rejections.rs
• examples/websockets_chat.rs
• examples/compression.rs
• examples/handlebars_template.rs
• examples/sse_chat.rs
• examples/query_string.rs
• examples/tracing.rs
• examples/routing.rs

pub async fn run_incoming<I>(self, incoming: I) where
    I: TryStream + Send,
    I::Ok: AsyncRead + AsyncWrite + Send + 'static + Unpin,
    I::Error: Into<Box<dyn StdError + Send + Sync>>, 

Run this Server forever on the current thread with a specific stream of incoming connections.

This can be used for Unix Domain Sockets, or TLS, etc.

Examples found in repository

examples/unix_socket.rs (line 13)

async fn main() {
    pretty_env_logger::init();

    let listener = UnixListener::bind("/tmp/warp.sock").unwrap();
    let incoming = UnixListenerStream::new(listener);
    warp::serve(warp::fs::dir("examples/dir"))
        .run_incoming(incoming)
        .await;
}

pub fn bind(
    self,
    addr: impl Into<SocketAddr> + 'static
) -> impl Future<Output = ()> + 'static

Bind to a socket address, returning a Future that can be executed on any runtime.

Panics

Panics if we are unable to bind to the provided address.

pub async fn try_bind(self, addr: impl Into<SocketAddr>)

Bind to a socket address, returning a Future that can be executed on any runtime.

In case we are unable to bind to the specified address, resolves to an error and logs the reason.

pub fn bind_ephemeral(
    self,
    addr: impl Into<SocketAddr>
) -> (SocketAddr, impl Future<Output = ()> + 'static)

Bind to a possibly ephemeral socket address.

Returns the bound address and a Future that can be executed on any runtime.

Panics

Panics if we are unable to bind to the provided address.

Examples found in repository

src/server.rs (line 132)

≺ ≻↕

    pub async fn run(self, addr: impl Into<SocketAddr>) {
        let (addr, fut) = self.bind_ephemeral(addr);
        let span = tracing::info_span!("Server::run", ?addr);
        tracing::info!(parent: &span, "listening on http://{}", addr);

        fut.instrument(span).await;
    }

    /// Run this `Server` forever on the current thread with a specific stream
    /// of incoming connections.
    ///
    /// This can be used for Unix Domain Sockets, or TLS, etc.
    pub async fn run_incoming<I>(self, incoming: I)
    where
        I: TryStream + Send,
        I::Ok: AsyncRead + AsyncWrite + Send + 'static + Unpin,
        I::Error: Into<Box<dyn StdError + Send + Sync>>,
    {
        self.run_incoming2(incoming.map_ok(crate::transport::LiftIo).into_stream())
            .instrument(tracing::info_span!("Server::run_incoming"))
            .await;
    }

    async fn run_incoming2<I>(self, incoming: I)
    where
        I: TryStream + Send,
        I::Ok: Transport + Send + 'static + Unpin,
        I::Error: Into<Box<dyn StdError + Send + Sync>>,
    {
        let fut = self.serve_incoming2(incoming);

        tracing::info!("listening with custom incoming");

        fut.await;
    }

    /// Bind to a socket address, returning a `Future` that can be
    /// executed on any runtime.
    ///
    /// # Panics
    ///
    /// Panics if we are unable to bind to the provided address.
    pub fn bind(self, addr: impl Into<SocketAddr> + 'static) -> impl Future<Output = ()> + 'static {
        let (_, fut) = self.bind_ephemeral(addr);
        fut
    }

src/test.rs (line 494)

↕

    pub async fn handshake<F>(self, f: F) -> Result<WsClient, WsError>
    where
        F: Filter + Clone + Send + Sync + 'static,
        F::Extract: Reply + Send,
        F::Error: IsReject + Send,
    {
        let (upgraded_tx, upgraded_rx) = oneshot::channel();
        let (wr_tx, wr_rx) = mpsc::unbounded_channel();
        let wr_rx = UnboundedReceiverStream::new(wr_rx);
        let (rd_tx, rd_rx) = mpsc::unbounded_channel();

        tokio::spawn(async move {
            use tokio_tungstenite::tungstenite::protocol;

            let (addr, srv) = crate::serve(f).bind_ephemeral(([127, 0, 0, 1], 0));

            let mut req = self
                .req
                .header("connection", "upgrade")
                .header("upgrade", "websocket")
                .header("sec-websocket-version", "13")
                .header("sec-websocket-key", "dGhlIHNhbXBsZSBub25jZQ==")
                .req;

            let query_string = match req.uri().query() {
                Some(q) => format!("?{}", q),
                None => String::from(""),
            };

            let uri = format!("http://{}{}{}", addr, req.uri().path(), query_string)
                .parse()
                .expect("addr + path is valid URI");

            *req.uri_mut() = uri;

            // let mut rt = current_thread::Runtime::new().unwrap();
            tokio::spawn(srv);

            let upgrade = ::hyper::Client::builder()
                .build(AddrConnect(addr))
                .request(req)
                .and_then(|res| hyper::upgrade::on(res));

            let upgraded = match upgrade.await {
                Ok(up) => {
                    let _ = upgraded_tx.send(Ok(()));
                    up
                }
                Err(err) => {
                    let _ = upgraded_tx.send(Err(err));
                    return;
                }
            };
            let ws = crate::ws::WebSocket::from_raw_socket(
                upgraded,
                protocol::Role::Client,
                Default::default(),
            )
            .await;

            let (tx, rx) = ws.split();
            let write = wr_rx.map(Ok).forward(tx).map(|_| ());

            let read = rx
                .take_while(|result| match result {
                    Err(_) => future::ready(false),
                    Ok(m) => future::ready(!m.is_close()),
                })
                .for_each(move |item| {
                    rd_tx.send(item).expect("ws receive error");
                    future::ready(())
                });

            future::join(write, read).await;
        });

        match upgraded_rx.await {
            Ok(Ok(())) => Ok(WsClient {
                tx: wr_tx,
                rx: rd_rx,
            }),
            Ok(Err(err)) => Err(WsError::new(err)),
            Err(_canceled) => panic!("websocket handshake thread panicked"),
        }
    }

pub fn try_bind_ephemeral(
    self,
    addr: impl Into<SocketAddr>
) -> Result<(SocketAddr, impl Future<Output = ()> + 'static), Error>

Tried to bind a possibly ephemeral socket address.

Returns a Result which fails in case we are unable to bind with the underlying error.

Returns the bound address and a Future that can be executed on any runtime.

pub fn bind_with_graceful_shutdown(
    self,
    addr: impl Into<SocketAddr> + 'static,
    signal: impl Future<Output = ()> + Send + 'static
) -> (SocketAddr, impl Future<Output = ()> + 'static)

Create a server with graceful shutdown signal.

When the signal completes, the server will start the graceful shutdown process.

Returns the bound address and a Future that can be executed on any runtime.

Example

use warp::Filter;
use futures::future::TryFutureExt;
use tokio::sync::oneshot;

let routes = warp::any()
    .map(|| "Hello, World!");

let (tx, rx) = oneshot::channel();

let (addr, server) = warp::serve(routes)
    .bind_with_graceful_shutdown(([127, 0, 0, 1], 3030), async {
         rx.await.ok();
    });

// Spawn the server into a runtime
tokio::task::spawn(server);

// Later, start the shutdown...
let _ = tx.send(());

pub fn try_bind_with_graceful_shutdown(
    self,
    addr: impl Into<SocketAddr> + 'static,
    signal: impl Future<Output = ()> + Send + 'static
) -> Result<(SocketAddr, impl Future<Output = ()> + 'static), Error>

Create a server with graceful shutdown signal.

When the signal completes, the server will start the graceful shutdown process.

pub fn serve_incoming<I>(self, incoming: I) -> impl Future<Output = ()> where
    I: TryStream + Send,
    I::Ok: AsyncRead + AsyncWrite + Send + 'static + Unpin,
    I::Error: Into<Box<dyn StdError + Send + Sync>>, 

Setup this Server with a specific stream of incoming connections.

This can be used for Unix Domain Sockets, or TLS, etc.

Returns a Future that can be executed on any runtime.

pub fn serve_incoming_with_graceful_shutdown<I>(
    self,
    incoming: I,
    signal: impl Future<Output = ()> + Send + 'static
) -> impl Future<Output = ()> where
    I: TryStream + Send,
    I::Ok: AsyncRead + AsyncWrite + Send + 'static + Unpin,
    I::Error: Into<Box<dyn StdError + Send + Sync>>, 

Setup this Server with a specific stream of incoming connections and a signal to initiate graceful shutdown.

This can be used for Unix Domain Sockets, or TLS, etc.

When the signal completes, the server will start the graceful shutdown process.

Returns a Future that can be executed on any runtime.

Trait Implementations

impl<F: Debug> Debug for Server<F>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.