Введение в hyper: серверная часть

hyper — это низкоуровневая HTTP-библиотека на Rust. Обычно она служит строительным блоком для более высокоуровневых библиотек, таких как axum, tonic, reqwest и других. Но иногда знание hyper могут требовать в вакансиях.

hyper предоставляет api как для клиента, так и для сервера, но в этой статье мы сосредоточимся исключительно на серверной части.

Первый «Hello, World!»

Создаём новый проект:

cargo new hyper-server

Добавим зависимости в Cargo.toml:

[dependencies] hyper = { version = "1.6.0", features = ["server", "http1"] }      hyper-util = { version = "0.1", features = ["full"] } http-body-util = "0.1.3"   tokio = { version = "1.44.2", features = ["rt-multi-thread", "macros"] }

Что делает каждая библиотека:

• hyper — основная HTTP-библиотека, включаем фичи поддержки HTTP/1.1-сервера

• hyper-util — предоставляет утилиты и адаптеры поверх hyper, упрощающие его использование

• http-body-util — вспомогательная библиотека для работы с HTTP-body

• tokio — асинхронный рантайм. С версии 1.0 hyper был отвязан от рантайма tokio, поэтому при желании можно использовать альтернативный рантайм

Необходимые импорты:

use hyper::{       Request, Response,       body::{Bytes, Incoming},       server::conn::http1,       service::service_fn,   };   use hyper_util::{Full, rt::TokioIo};   use std::{convert::Infallible, net::SocketAddr};   use tokio::net::TcpListener;

Пояснения:

• Request — представляет HTTP-запрос

• Response — представляет HTTP-ответ

• Bytes — тип для эффективного представления неизменяемых бинарных данных

• Incoming — асинхронный поток байтов, представляющий тело HTTP-запроса, поступающее от клиента

• http1 — модуль для работы с HTTP/1.1-соединениями

• service_fn — функция адаптер, позволяющая создать Service из обычной асинхронной функции

• Full — вспомогательная обёртка для заранее известного содержимого Body

• TokioIo — адаптер, который оборачивает tokio::net::TcpStream в тип, совместимый с hyper

• Infallible — тип ошибки, которая никогда не происходит

• SocketAddr — структура, представляющая IP-адрес и порт

• TcpListener— асинхронный TCP-сервер

Обработчик запроса:

async fn hello(_: Request<Incoming>) -> Result<Response<Full<Bytes>>, Infallible> {       Ok(Response::new(Full::new(Bytes::from("Hello, World!"))))   }

С первого взгляда обработчик может показаться достаточно громоздким:

• _: Request<Incoming> — входящий HTTP-запрос от клиента. Используем нижнее подчёркивание, чтобы явно показать, что игнорируем содержимое запроса, так как ответ всегда один и тот же

• Result<Response<Full<Bytes>>, Infallible> — тип возвращаемого значения обработчика:

  • Response<Full<Bytes>> — HTTP-ответ с телом, представленным блок байтов

  • Infallible — обозначает, что наш обработчик не может вернуть ошибку

• Response::new(Full::new(Bytes::from("Hello, World!"))) — оборачивает строку в байты и преобразует их в тело ответа

Запуск сервера:

#[tokio::main]   async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {       // IP-адрес и порт, на которых будет работать сервер    let addr = SocketAddr::from(([0, 0, 0, 0], 3000));        // Создаём асинхронный TCP-сервер       let listener = TcpListener::bind(addr).await?;          // Бесконечный цикл: принимаем входящие соединения       loop {          // Ждём новое соединение от клиента           let (stream, _) = listener.accept().await?;              // Адаптер, который оборачивает `TcpStream` и делает его совместимым с hyper         let io = TokioIo::new(stream);              // Для каждого соединения создаётся отдельная асинхронная задача           tokio::task::spawn(async move {              // HTTP/1.1 сервер, привязанный к нашему обработчику `hello`               if let Err(err) = http1::Builder::new()                   .serve_connection(io, service_fn(hello))                   .await               {               // обрабатываем возможные ошибки                 eprintln!("Error serving connection: {:?}", err);               }           });       }   }

Запускаем сервер:

cargo run

Проверяем работу сервера:

curl http://127.0.0.1:3000

В данном примере используется модуль http1. Если ван нужен http2, то вы можете посмотреть его реализацию в официальном примере.

Эхо сервер

Новые импорты:

use hyper::{ body::{Body, Frame}, Method, StatusCode } use http_body_util::{Empty, combinators::BoxBody, BodyExt}; 

Пояснения:

• Body — трейт, описывающий HTTP тело

• Frame — единица данных в теле HTTP-сообщения. Используется для представления либо части тела (Bytes), либо сигнала конца потока

• Method — перечисление всех возможных HTTP-методов (GET, POST, PUT, DELETE и т.д.)

• StatusCode — перечисление стандартных HTTP-статусов (200 OK, 404 Not Found, 500 Internal Server Error и т.д.)

• Empty — вспомогательная обёртка для пустого содержимого Body

• BoxBody — обобщённый тип тела ответа. Позволяет вернуть из функции разные типы Body

• BodyExt — набор удобных методов для работы с телом запроса

Вспомогательные функции для создания полных и пустых тел запросов:

/// Создаёт пустое тело ответа fn empty() -> BoxBody<Bytes, hyper::Error> {       Empty::<Bytes>::new()           .map_err(|never| match never {})           .boxed()   }    /// Оборачивает переданный chunk (например, `&str` или `Vec<u8>`) в тело ответа fn full<T: Into<Bytes>>(chunk: T) -> BoxBody<Bytes, hyper::Error> {       Full::new(chunk.into())           .map_err(|never| match never {})           .boxed()   } 

Обработчики:

/// Возвращает клиенту тело запроса без изменений.    fn echo(req: Request<Incoming>) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {       Ok(Response::new(req.into_body().boxed()))   }      /// Обработчик HTTP-запроса, который возвращает тело запроса,   /// преобразованное в верхний ASCII-регистр   async fn echo_uppercase(req: Request<Incoming>) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {       // Преобразуем поток тела запроса по частям       let frame_stream = req.into_body().map_frame(|frame| {           let frame = if let Ok(data) = frame.into_data() {               // Преобразуем каждый байт в верхний ASCII-регистр               data.iter()                   .map(|byte| byte.to_ascii_uppercase())                   .collect::<Bytes>()           } else {               Bytes::new()           };              Frame::data(frame)       });          Ok(Response::new(frame_stream.boxed()))   }      /// Обработчик HTTP-запроса, который возвращает тело запроса в перевёрнутом виде   async fn echo_reversed(req: Request<Incoming>) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {       // Защита от слишком больших тел       let upper = req.body().size_hint().upper().unwrap_or(u64::MAX);       // Если размер тела запроса превышает 64 КБ, возвращает ошибку `413 Payload Too Large`       if upper > 1024 * 64 {           let response = Response::builder()               .status(StatusCode::PAYLOAD_TOO_LARGE)               .body(full("Body too big"))               .unwrap();              return Ok(response);       }          // Читаем всё тело целиком       let whole_body = req.collect().await?.to_bytes();          // Разворачиваем байты в обратном порядке       let reversed_body = whole_body.iter()           .rev()           .cloned()           .collect::<Vec<u8>>();          Ok(Response::new(full(reversed_body)))   }      /// Возвращает ответ `404 Not Found` с пустым телом async fn not_found() -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {       let response = Response::builder()           .status(StatusCode::NOT_FOUND)           .body(empty())           .unwrap();          Ok(response)   }

Как говорилось в начале статьи, hyper — это достаточно низкоуровневая библиотека, и в отличие от axum или actix-web, не предоставляет встроенную маршрутизацию. Поэтому её нужно реализовать вручную:

async fn router(      req: Request,  ) -> Result>, hyper::Error> {  // Сопоставляем HTTP-метод и путь    match (req.method(), req.uri().path()) {          (&Method::POST, "/echo") => echo(req),          (&Method::POST, "/echo/uppercase") => echo_uppercase(req).await,          (&Method::POST, "/echo/reversed") => echo_reversed(req).await,          _ => not_found().await,      }  } 

В функции main обновляем service_fn :

.serve_connection(io, service_fn(router))

Запускаем сервер и проверяем его работу:

curl -X POST http://127.0.0.1:3000/echo -d "hello world" curl -X POST http://127.0.0.1:3000/echo/uppercase -d "hello world" curl -X POST http://127.0.0.1:3000/echo/reversed -d "hello world"

Middleware

Добавим к текущему эхо серверу middleware, который будет логировать метод и путь каждого HTTP-запроса

Новая зависимость в Cargo.toml:

[dependencies] tower = "0.5.2" # предоставляет абстракции для middleware

Новые импорты:

use hyper::service::Service; use tower::ServiceBuilder;  

• Service — используется для подключения пользовательской логики к сетевому соединению (например, service_fn(router) создаёт имплементацию этого трейта)

• ServiceBuilder — позволяет подключить middleware к сервису

Пишем простой middleware для логирования запросов:

/// Простая обёртка, логирующая HTTP-запросы   #[derive(Debug, Clone)]   pub struct Logger<S> {       /// Сервис, к которому делегируется обработка запроса       inner: S,   }      impl<S> Logger<S> {       /// Создание нового логера, оборачивающего другой сервис       pub fn new(inner: S) -> Self {           Logger { inner }       }   }      type Req = Request<Incoming>;      /// Реализация трейта `Service` для `Logger`,   /// позволяющая использовать его как middleware   impl<S> Service<Req> for Logger<S>   where       S: Service<Req>,  // Внутренний сервис должен реализовывать трейт `Service`   {       // Тип ответа будет таким же, как у внутреннего сервиса       type Response = S::Response;       // Тип ошибки - тоже такой же       type Error = S::Error;       // Тип возвращаемого future (обработчика запроса)       type Future = S::Future;          /// Метод, вызываемый при каждом запросе       fn call(&self, req: Req) -> Self::Future {           // Логируем метод и путь запроса           println!("processing request: {} {}", req.method(), req.uri().path());              // Передаём запрос дальше во внутренний сервис           self.inner.call(req)       }   }

Обновляем обработчик в функции main:

tokio::task::spawn(async move {       // Оборачиваем обработчик маршрутов в Service                     let svc = service_fn(router);        // Добавляем наш middleware к обработчику       let svc = ServiceBuilder::new().layer_fn(Logger::new).service(svc);       // Ожидаем завершение соединения (или ошибку)      if let Err(err) = http1::Builder::new().serve_connection(io, svc).await {           eprintln!("server error: {}", err);       }   });

Теперь, при каждом запросе, будут выводиться записи вида:

processing request: POST /echo/reversed processing request: POST /echo 

Для упрощения примера, запросы логируются с помощью println!. Если вам нужен полноценный логер, то ознакомьтесь с крейтом tracing

Gracefully Shutdown

Graceful shutdown — процесс, при котором сервер:

• перестаёт принимать новые подключения

• позволяет завершить текущие соединения

• корректно освобождает ресурсы (файлы, БД, сокеты и т.п.)

Что нужно для реализации:

• Сигнал завершения

• Цикл, обрабатывающий входящие соединения

• Наблюдатель, координирующий завершение соединений

Добавим в tokio новую фичу для обработки сигналов:

[dependencies] tokio = { version = "1.44.2", features = [..., "signal"] } # добавляем фичу для обработки сигналов

Обработка сигнала:

async fn shutdown_signal() {     // Ожидание сигнала CTRL+C      tokio::signal::ctrl_c()         .await         .expect("failed to install CTRL+C signal handler"); }

Обновляем функцию main для отслеживания сигнала отключения и подключаем отслеживание сигнала завершения для соединений:

#[tokio::main]   async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {       // IP-адрес и порт, где будет слушать сервер   let addr = SocketAddr::from(([0, 0, 0, 0], 3000));        // Создаём асинхронный TCP-сервер       let listener = TcpListener::bind(addr).await?;       // Создаём объект для управления "graceful shutdown" (мягкое завершение)       let graceful = hyper_util::server::graceful::GracefulShutdown::new();     // Подключаем сигнал завершения       // После его получения начнётся завершение работы сервера         let mut signal = std::pin::pin!(shutdown_signal());          // Запускаем цикл, чтобы постоянно принимать входящие соединения       loop {           tokio::select! {               // Обработка новых соединений               Ok((stream, _addr)) = listener.accept() => {                   // Адаптер, который оборачивает `TcpStream` и делает его совместимым с hyper                   let io = TokioIo::new(stream);                   // Оборачиваем обработчик маршрутов в Service                                 let svc = service_fn(router);                   // Добавляем наш middleware к обработчику                   let svc = ServiceBuilder::new().layer_fn(Logger::new).service(svc);                   // Создаём соединение HTTP/1.1                   let conn = http1::Builder::new().serve_connection(io, svc);                   // Подключаем отслеживание сигнала завершения для соединения                   let fut = graceful.watch(conn);                     // Для каждого соединения создаётся отдельная асинхронная задача                   tokio::task::spawn(async move {                       // Ожидаем завершение соединения (или ошибку)                       if let Err(err) = fut.await {                           eprintln!("server error: {}", err);                       }                   });               },               // Получен сигнал завершения               _ = &mut signal => {                   // Закрываем listener, чтобы не принимать новые соединения                   drop(listener);                   eprintln!("graceful shutdown signal received");                   // Прерываем основной цикл                   break;               }           }       }          // После выхода из цикла — ждём завершения всех активных соединений       tokio::select! {           // Успешное завершение всех соединений           _ = graceful.shutdown() => {               eprintln!("all connections gracefully closed");           },           // Если соединения не закрылись за 10 секунд — принудительно завершаем           _ = tokio::time::sleep(std::time::Duration::from_secs(10)) => {               eprintln!("timed out wait for all connections to close");           }       }          Ok(())   }

Теперь, при сигнале отключения, будет выводиться:

graceful shutdown signal received all connections gracefully closed

Собираем сервер в docker контейнер

Dockerfile

FROM rust:1.86-slim AS builder   RUN apt-get update && apt-get install musl-tools -y && rustup target add x86_64-unknown-linux-musl   WORKDIR /usr/src/app   COPY Cargo.toml Cargo.lock ./   COPY src src   RUN cargo build --target x86_64-unknown-linux-musl --release      FROM scratch   COPY --from=builder /usr/src/app/target/x86_64-unknown-linux-musl/release/hyper-server /usr/local/bin/hyper-server   EXPOSE 3000   CMD ["hyper-server"]

Сборка контейнера:

docker build -t hyper-server .

Запуск контейнера:

docker run --rm -p 3000:3000 hyper-server

Проверяем работу сервера:

curl -X POST http://127.0.0.1:3000/echo -d "hello world"

Заключение

Если вы хотите продолжить изучение hyper, то ознакомитесь с официальной документацией и примерами.

Это моя первая статья на Хабре, если вы нашли какие-то ошибки или неточности, буду рад уточнениям в комментариях.