HTTP/2 with ntex-h2

ntex-h2 is a separate crate that implements HTTP/2 for ntex. Most applications use it indirectly through ntex::http or ntex::web. When a connection negotiates HTTP/2, ntex-h2 handles frames, header compression, streams, and flow control. ntex turns each stream into requests and responses.

Use ntex-h2 directly when you only need HTTP/2, need trailers or long-lived bidirectional streams, or want to manage flow control yourself. ntex-grpc uses it this way for both its client and its server.

For outgoing requests, client::Client opens and reuses a pool of connections to one server. Choose client::SimpleClient instead when you want to manage one connection yourself. On the receiving side, server::Server is an ntex service that handles HTTP/2 connections.

ntex-h2 speaks only HTTP/2 and does not negotiate the protocol. Any transport you give it must already be committed to HTTP/2. Over TLS, ALPN must have selected h2. Over plain TCP (h2c), both sides must know in advance that they will use HTTP/2.

The HTTP/1.1 Upgrade mechanism and server push are not supported.

Here is a client that sends a GET request over plain TCP and prints the response:

use ntex::SharedCfg;
use ntex::http::{HeaderMap, Method};
use ntex_h2::{MessageKind, StreamEof, client::Client};

#[ntex::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = Client::builder("127.0.0.1:8080").build(SharedCfg::new("H2-CLIENT"));

    let (_send, recv) = client
        .send(Method::GET, "/index.html".into(), HeaderMap::new(), true)
        .await?;

    while let Some(msg) = recv.recv().await {
        match msg.kind {
            MessageKind::Headers { pseudo, .. } => println!("status: {:?}", pseudo.status),
            MessageKind::Data(data, _) => println!("data: {data:?}"),
            MessageKind::Eof(StreamEof::Data(data, _)) => println!("last data: {data:?}"),
            MessageKind::Eof(_) | MessageKind::Disconnect(_) => break,
        }
    }
    Ok(())
}

This API works with HTTP/2 streams and messages. Unlike ntex::client, it has no URLs, body helpers, or Response type.

Streams and messages

Each HTTP/2 stream carries one request and its response. Many independent streams can share a connection. The client opens a stream by sending request headers. Both sides can then send data, and each closes its own half when it is done.

Either side can cut a stream short with a reset. The connection and its other streams stay open.

A StreamRef identifies a stream and lets you send on it. It is cheap to clone. Everything that arrives on a stream reaches you as a Message:

pub struct Message {
    pub stream: StreamRef,
    pub kind: MessageKind,
}

A client and a server see the same sequence. It begins with Headers { pseudo, headers, eof }, the request or response head. The pseudo field holds HTTP/2’s pseudo-headers: method, scheme, authority, and path on requests, and status on responses. If eof is true, there is no body and this is the only message.

Otherwise, any number of Data(Bytes, Capacity) messages carry body chunks. Each Capacity represents the flow-control credit used by its chunk; the flow-control section explains when to release it.

Exactly one Eof(StreamEof) or Disconnect(err) ends the sequence. StreamEof::Data carries a final body chunk, StreamEof::Trailers carries trailing headers, and StreamEof::Error reports a failed or reset stream. Disconnect means the whole connection failed while the stream was open.

Send headers first, using send on a client or send_response on a server. Then send body chunks with send_payload, as many times as needed. To finish, set eof on the last call or send trailers with send_trailers.

Configuration

Start with limits that suit the requests your service handles. A large upload may need different receive windows from a small JSON call, and a shared connection needs room for several streams. ServiceConfig sets those limits and timeouts. Add it to a SharedCfg to apply it to every connection created with that configuration:

use ntex::{SharedCfg, time::Seconds};
use ntex_h2::ServiceConfig;

fn config() -> SharedCfg {
    SharedCfg::new("H2")
        .add(
            ServiceConfig::new()
                .set_max_concurrent_streams(128)
                .set_initial_window_size(256 * 1024)
                .set_initial_connection_window_size(4 * 1024 * 1024)
                .set_ping_timeout(Seconds(30)),
        )
        .into()
}

ntex::http exposes this same type as Http2ServiceConfig. These settings also apply to HTTP/2 in an ntex web server or HTTP client.

ntex-h2 connections use ordinary ntex I/O objects. Add an IoConfig to the same SharedCfg to control buffers and backpressure, as described in the I/O chapter.

The flow-control and frame settings determine how much data the peer may send:

Receive windows bound body data, but headers and stream bookkeeping also take space. Use these limits to bound the work and state a peer can create:

Finally, keep a stalled peer from holding connection resources indefinitely. These timeouts cover the handshake and ongoing protocol exchanges:

Stream limits apply only to streams the peer opens. On a server, set_max_concurrent_streams caps how many requests a client may run at once. A client is bound by the limit the server announces.

ntex-h2 also guards against rapid-reset attacks. It closes a connection with GOAWAY if at least ten streams have been opened and half of them were reset.

The client

A client sends requests to one server. You can let a pool manage its connections or manage a single connection yourself.

Connecting with a pool

Use client::Client to reuse connections without managing them yourself. Like the HTTP client, create it once and keep it for the life of the application. Clones are cheap and share the same connections.

Each request needs a free stream slot, not a connection of its own. The pool looks for a connection that can open another stream and opens a new connection when needed. You can also ask it to grow to a minimum size with minconn.

Create the pool with a builder. Pass the configuration for all its connections to build:

use ntex::{SharedCfg, time::Seconds};
use ntex_h2::{ServiceConfig, client::Client};

fn backend() -> Client {
    let cfg = SharedCfg::new("BACKEND").add(ServiceConfig::new().set_ping_timeout(Seconds(30)));

    Client::builder("backend.internal:8080")
        .connect_timeout(Seconds(3))
        .connection_limit(4)
        .lifetime(Seconds(600))
        .build(cfg)
}

The builder has these options:

When every connection is full and the pool is at its limit, send waits for a stream to free up. Errors before a stream opens, such as a failed connection attempt or a timeout, are reported as ClientError.

The pool opens new connections one at a time and spreads requests across them. It prefers connections using less than half of their stream limit.

Secure connections

To connect over TLS, give the pool a TLS connector that offers h2 through ALPN. Also set secure(true) so requests carry the https scheme. Here is an OpenSSL configuration:

use ntex::SharedCfg;
use ntex_h2::client::Client;
use openssl::ssl::{SslConnector, SslMethod};

fn secure_client() -> Result<Client, openssl::error::ErrorStack> {
    let mut builder = SslConnector::builder(SslMethod::tls())?;
    builder.set_alpn_protos(b"\x02h2")?;

    Ok(Client::builder("api.example.com:443")
        .connector(ntex::tls::openssl::SslConnector::new(builder.build()))
        .secure(true)
        .build(SharedCfg::new("API")))
}

The ALPN list b"\x02h2" is the protocol name h2 prefixed with its length. With rustls, set alpn_protocols to vec![b"h2".to_vec()] in its ClientConfig.

Unlike ntex::client, ntex-h2 cannot fall back to HTTP/1.1. If you also offer http/1.1, make sure the server will pick h2.

A single connection

Use client::SimpleClient to keep a session on one connection or run HTTP/2 over a transport you created. It wraps a single connection that you have already established:

use ntex::{SharedCfg, io::Io, util::ByteString};
use ntex_h2::client::SimpleClient;

async fn connect() -> Result<SimpleClient, Box<dyn std::error::Error>> {
    let cfg: SharedCfg = SharedCfg::new("H2").into();
    let io: Io = ntex::connect::connect_with("127.0.0.1:8080", &cfg).await?;

    Ok(SimpleClient::new(io, false, ByteString::from("127.0.0.1")))
}

The arguments are the transport, whether requests use the https scheme, and the :authority sent with each request. The connection reads its ServiceConfig from the transport’s SharedCfg.

To avoid setting this up by hand, use client::Connector. This ntex service takes an address and returns a SimpleClient.

SimpleClient is cheap to clone. The connection stays open while any clone exists and closes gracefully when the last one is dropped.

To stop earlier, call close(): open streams may finish, but no new ones can start. If you also need to wait until the connection is gone, use disconnect().await, with .disconnect_timeout(..) on the returned future to bound the wait. Reserve force_close() for cases where you must drop the connection immediately.

Use is_closed() to check whether the connection has closed. on_disconnect() tells you when it closes, whether you or the server initiated it.

A pool manages the connection lifecycle for you. If you need access to one of its connections, call Client::client(). It returns a SimpleClient from the pool that can open another stream.

Sending a request

Call send to start a request. Pass the method, the :path including any query string, the headers, and an eof flag.

It returns the two halves of the new stream. Use the SendStream for the request body and the RecvStream for the response.

use ntex::http::{HeaderMap, Method, header};
use ntex::util::Bytes;
use ntex_h2::client::Client;

async fn upload(client: &Client, body: Bytes) -> Result<(), Box<dyn std::error::Error>> {
    let mut headers = HeaderMap::new();
    headers.insert(header::CONTENT_TYPE, header::HeaderValue::from_static("application/octet-stream"));

    let (send, recv) = client.send(Method::POST, "/upload".into(), headers, false).await?;
    send.send_payload(body, true).await?;

    // ... read the response from `recv`
    drop(recv);
    Ok(())
}

Set eof to true when the request has no body, as with most GET requests. Otherwise, send the body with send_payload, which accepts anything that converts to Bytes. Pass eof: true on the last call. If the request has trailers, finish with send_trailers(headers) instead.

To stream a large body, call send_payload for each chunk:

use ntex::util::Bytes;
use ntex_h2::client::SendStream;
async fn stream_body(
    send: &SendStream,
    chunks: Vec<Bytes>,
) -> Result<(), ntex::error::Error<ntex_h2::OperationError>> {
    for chunk in chunks {
        send.send_payload(chunk, false).await?;
    }
    send.send_payload(Bytes::new(), true).await
}

You can read the response while still sending the request body. The two halves are independent, which supports bidirectional streaming protocols. Only one task should send on a stream at a time.

send_payload returns once it has handed the data to the connection. It splits the data into frames that fit the peer’s maximum frame size and the flow-control window. If the window is closed, it waits for the peer to reopen it (see Flow control).

It also waits while the connection’s write buffer is full. A fast sender cannot outrun the network.

Reading the response

Read a response from its RecvStream with recv().await. Each call returns the next message, or None once the response is complete. You can also use stream combinators because RecvStream implements Stream.

This function collects a whole response:

use ntex::http::{HeaderMap, StatusCode};
use ntex::util::{Bytes, BytesMut};
use ntex_h2::{MessageKind, StreamEof, client::RecvStream};

async fn read_response(
    recv: RecvStream,
) -> Result<(StatusCode, HeaderMap, Bytes), Box<dyn std::error::Error>> {
    let mut status = StatusCode::OK;
    let mut headers = HeaderMap::new();
    let mut body = BytesMut::new();

    while let Some(msg) = recv.recv().await {
        match msg.kind {
            MessageKind::Headers { pseudo, headers: hdrs, .. } => {
                status = pseudo.status.unwrap_or(StatusCode::OK);
                headers = hdrs;
            }
            MessageKind::Data(chunk, _capacity) => body.extend_from_slice(&chunk),
            MessageKind::Eof(StreamEof::Data(chunk, _capacity)) => {
                body.extend_from_slice(&chunk);
                break;
            }
            MessageKind::Eof(StreamEof::Trailers(_)) => break,
            MessageKind::Eof(StreamEof::Error(err)) => return Err(err.into()),
            MessageKind::Disconnect(err) => return Err(err.into()),
        }
    }
    Ok((status, headers, body.freeze()))
}

Each Data message comes with a Capacity. The example drops it right away, letting the server keep sending. Hold it to slow the server down, as Flow control explains.

Resets and cancellation

To cancel a request, drop either unfinished half of its stream. Dropping the RecvStream before the response is complete resets the stream with CANCEL. Dropping the SendStream before the request body is complete does the same. The server can then stop working on the request.

A timeout needs no extra cancellation code:

use ntex_h2::client::RecvStream;
async fn read_response(_: RecvStream) -> Result<(), Box<dyn std::error::Error>> { Ok(()) }
async fn example(recv: RecvStream) {
use ntex::time::{Seconds, timeout};

// If the timeout fires, `recv` is dropped and the stream is reset.
let result = timeout(Seconds(10), read_response(recv)).await;
}

Either side can reset a stream. Each reset carries a Reason code from ntex_h2::frame::Reason, such as CANCEL or INTERNAL_ERROR. To choose a reason, call reset(reason) on the SendStream or the StreamRef.

Use is_reset() to check whether either side has reset the stream. on_reset() tells you when a reset happens.

When the server resets a stream, you receive Eof(StreamEof::Error(..)). Further sends fail. When the whole connection fails, every open stream gets a Disconnect message instead.

Stream limits

A server announces how many streams it accepts at once on each connection. At that limit, SimpleClient::send waits until a stream finishes. The pool uses another connection or opens a new one.

On a SimpleClient, compare active_streams() with max_streams() to see how many slots are in use and what limit the server announced. Use is_ready() to check for a free slot or ready().await to wait for one. If your code is driven by callbacks, on_capacity(callback) runs whenever a stream is released.

Another task may take the free slot between your check and your send. Use reserve() to claim a slot immediately. It returns a StreamReservation, or None if the connection is full. The reservation’s send opens the stream without waiting. Dropping an unused reservation releases the slot.

Flow control

Flow control keeps one slow consumer from stalling every stream on a connection. TCP’s own flow control is not enough when many HTTP/2 streams share that connection.

HTTP/2 gives each stream a window of credit, measured in bytes of body data. The connection has another window, shared by all streams. A sender can send only as much data as both windows allow. Headers do not count against either window.

Each DATA frame uses credit from both windows. When either reaches zero, the sender must wait. The receiver grants more credit with a WINDOW_UPDATE frame.

Receiving data

Keep each chunk’s Capacity until you have handled the data. It represents the window credit the chunk used. Holding it keeps that credit in use, so the sender’s pace follows your consumer’s pace.

Drop the Capacity to return the credit to both the stream and connection windows. To release only part of it, call consume(n).

When you queue data for another task, pass the Capacity along with it. The task should drop it only after handling the data:

use ntex::channel::mpsc;
use ntex::util::Bytes;
use ntex_h2::{Capacity, MessageKind, StreamEof, client::RecvStream};

// Forward the body to a consumer task. The server can only send as fast
// as the consumer drops the capacities.
async fn forward(recv: RecvStream, tx: mpsc::Sender<(Bytes, Capacity)>) {
    while let Some(msg) = recv.recv().await {
        match msg.kind {
            MessageKind::Data(data, cap) => {
                if tx.send((data, cap)).is_err() {
                    break; // consumer is gone, dropping `recv` resets the stream
                }
            }
            MessageKind::Eof(StreamEof::Data(data, cap)) => {
                let _ = tx.send((data, cap));
                break;
            }
            MessageKind::Headers { .. } => {}
            MessageKind::Eof(_) | MessageKind::Disconnect(_) => break,
        }
    }
}

ntex-h2 batches WINDOW_UPDATE frames. It waits until released credit reaches a third of initial_window_size for a stream, or a quarter of initial_connection_window_size for the connection.

Unreleased data also counts against the shared connection window. A stream that buffers a lot of data without releasing its credit can slow down the others. The default connection window is therefore much larger than the stream window: 4 MiB versus 64 KiB per stream.

Size the windows together. The stream window limits how much a single request can buffer. The connection window limits the total across all requests.

If you merge chunks into one buffer, combine their capacities with + or += so you can release the credit together. Start with StreamRef::empty_capacity() for a zero-sized capacity belonging to that stream. Only combine capacities from the same stream; mixing streams panics.

If a peer sends more data than the window allows, the stream fails with StreamError::RecvWindowExceeded.

Sending data

Use send_payload to send data within the peer’s flow-control window. ntex-h2 tracks the granted credit and sends as much as the window allows. It waits for more credit when needed, until all the data is out.

If you produce data on demand, check capacity before generating a chunk. This avoids producing data that would sit in memory while you wait for the peer. available_send_capacity() reports the current window, which is the smaller of the stream’s and the connection’s window. To wait for credit before producing the next chunk, use send_capacity().await; it returns the available window size:

use ntex::util::Bytes;
use ntex_h2::client::SendStream;
fn produce(_: usize) -> Option<Bytes> { None }
async fn send_generated(send: &SendStream) -> Result<(), ntex::error::Error<ntex_h2::OperationError>> {
    loop {
        let window = send.send_capacity().await? as usize;
        match produce(window) {
            Some(chunk) => send.send_payload(chunk, false).await?,
            None => return send.send_payload(Bytes::new(), true).await,
        }
    }
}

A peer that never opens its window could leave a sender waiting forever. set_capacity_timeout bounds that wait to 5 seconds by default. If the wait exceeds the timeout, ntex-h2 resets the stream with CANCEL. The send fails with StreamError::CapacityTimeout.

The server

An ntex-h2 server gives your handlers direct access to stream messages. Your handlers send responses and keep any state they need for each stream.

Running a server

Use server::Server to serve HTTP/2 on an I/O object until the connection closes. It is an ntex service, so you can run it with ntex::server::build(), as described in the server chapter:

use ntex::http::{HeaderMap, StatusCode};
use ntex::{SharedCfg, error::Error};
use ntex_h2::{Message, MessageKind, OperationError, ServiceConfig, server};

#[ntex::main]
async fn main() -> std::io::Result<()> {
    let cfg = SharedCfg::new("H2-SRV").add(ServiceConfig::new().set_max_concurrent_streams(100));

    ntex::server::build()
        .bind("h2", "127.0.0.1:8080", cfg, async |_| {
            server::Server::new(async |msg: Message| {
                if let MessageKind::Headers { pseudo, .. } = msg.kind {
                    println!("{:?} {:?}", pseudo.method, pseudo.path);
                    msg.stream.send_response(StatusCode::OK, HeaderMap::new(), false)?;
                    msg.stream.send_payload("hello world", true).await?;
                }
                Ok::<_, Error<OperationError>>(())
            })
        })?
        .run()
        .await
}

Connections take their ServiceConfig from the SharedCfg passed to bind. This listener accepts plain TCP, so clients must know in advance that it speaks HTTP/2. For example, use curl --http2-prior-knowledge http://127.0.0.1:8080/.

For TLS, put a TLS acceptor in front of the server and have it select h2 during ALPN:

use ntex::{Service, SharedCfg, error::Error};
use ntex::tls::openssl::SslAcceptor;
use ntex_h2::{Message, OperationError, server};
use openssl::ssl::{self, AlpnError, SslFiletype, SslMethod};

#[ntex::main]
async fn main() -> std::io::Result<()> {
    let mut builder = ssl::SslAcceptor::mozilla_intermediate(SslMethod::tls()).unwrap();
    builder.set_private_key_file("key.pem", SslFiletype::PEM).unwrap();
    builder.set_certificate_chain_file("cert.pem").unwrap();
    builder.set_alpn_select_callback(|_, protos| {
        if protos.windows(3).any(|w| w == b"\x02h2") {
            Ok(b"h2")
        } else {
            Err(AlpnError::NOACK)
        }
    });
    let acceptor = builder.build();

    ntex::server::build()
        .bind("h2", "127.0.0.1:8443", SharedCfg::new("H2-TLS"), async move |_| {
            SslAcceptor::new(acceptor.clone())
                .map_err(|e| server::ServerError::from(std::io::Error::other(e)))
                .and_then(server::Server::new(async |_msg: Message| {
                    // handle messages
                    Ok::<_, Error<OperationError>>(())
                }))
        })?
        .run()
        .await
}

To serve an I/O object you already have, call server::handle_one(io, publish, control) with services you have already created. The I/O object might come from your own acceptor or a connection that switched protocols. ntex-grpc runs its server this way.

Handling streams

Handle incoming messages in the publish service you pass to Server::new. ntex-h2 creates a new publish service for each connection. It calls the service for every message on every stream the client opens. The request headers, each body chunk, and the end of the stream arrive in separate calls.

These calls run concurrently. ntex-h2 starts the next call without waiting for the previous one to finish, so a slow request does not hold up others. The call that receives the headers cannot await the request body as part of that same call. Body data arrives through separate calls to the service.

Keep per-stream state in the publish service and route each message to where it is needed. A map keyed by StreamRef works for this because the handle implements Hash and Eq.

This service buffers each request body up to a size limit. Once the request is complete, it echoes the body back:

use std::{cell::RefCell, collections::HashMap};

use ntex::http::{HeaderMap, StatusCode};
use ntex::service::{Ctx, Service};
use ntex::util::BytesMut;
use ntex_h2::{Message, MessageKind, OperationError, StreamEof, StreamRef, frame::Reason};

const MAX_BODY: usize = 1024 * 1024;

#[derive(Default)]
struct Echo {
    bodies: RefCell<HashMap<StreamRef, BytesMut>>,
}

impl Service<(), Message> for Echo {
    type Res = ();
    type Error = ntex::error::Error<OperationError>;

    async fn call(&self, msg: Message, _: Ctx<'_, Self, ()>) -> Result<(), Self::Error> {
        let Message { stream, kind } = msg;

        let body = match kind {
            MessageKind::Headers { eof: true, .. } => BytesMut::new(),
            MessageKind::Headers { .. } => {
                self.bodies.borrow_mut().insert(stream, BytesMut::new());
                return Ok(());
            }
            MessageKind::Data(data, _) => {
                let mut bodies = self.bodies.borrow_mut();
                if let Some(body) = bodies.get_mut(&stream) {
                    body.extend_from_slice(&data);
                    if body.len() > MAX_BODY {
                        bodies.remove(&stream);
                        stream.reset(Reason::CANCEL);
                    }
                }
                return Ok(());
            }
            MessageKind::Eof(StreamEof::Data(data, _)) => {
                let Some(mut body) = self.bodies.borrow_mut().remove(&stream) else {
                    return Ok(());
                };
                body.extend_from_slice(&data);
                body
            }
            MessageKind::Eof(StreamEof::Trailers(_)) => {
                let Some(body) = self.bodies.borrow_mut().remove(&stream) else {
                    return Ok(());
                };
                body
            }
            MessageKind::Eof(StreamEof::Error(_)) | MessageKind::Disconnect(_) => {
                self.bodies.borrow_mut().remove(&stream);
                return Ok(());
            }
        };

        if body.len() > MAX_BODY {
            stream.reset(Reason::CANCEL);
            return Ok(());
        }
        stream.send_response(StatusCode::OK, HeaderMap::new(), false)?;
        stream.send_payload(body.freeze(), true).await
    }
}

The final Eof can carry data too, so the size limit must cover that last chunk as well as ordinary Data messages. Release each RefCell borrow before an await: other calls on the same service can run while this call is waiting.

Remove the map entry on Eof(StreamEof::Error(..)) or Disconnect. This keeps the map from leaking when a client goes away mid-request.

Wrap a service type like this in a factory. fn_factory takes an async function and calls it for each new connection:

use ntex::service::{Ctx, Service};
use ntex_h2::{Message, OperationError};
#[derive(Default)] struct Echo;
impl Service<(), Message> for Echo {
    type Res = ();
    type Error = ntex::error::Error<OperationError>;
    async fn call(&self, _: Message, _: Ctx<'_, Self, ()>) -> Result<(), Self::Error> { Ok(()) }
}
use ntex::service::fn_factory;
use ntex_h2::server;

fn server() -> server::Server<ntex::io::IoBoxed, ntex::error::Error<OperationError>, std::convert::Infallible> {
server::Server::new(fn_factory(async |_: &()| {
    Ok::<_, std::convert::Infallible>(Echo::default())
}))
}

For large or streaming bodies, store a channel sender in the map instead of buffering the body. Pass each chunk to a task that handles the request. Send the Capacity with the data, as in the flow control example. The client’s pace will then follow the task’s pace.

ntex::http builds request payloads on top of ntex-h2 this way.

Sending the response

Send a response through the request’s StreamRef. Start with send_response(status, headers, eof), setting eof to true if there is no body. Otherwise, follow it with send_payload(data, eof), just as on the client. You can finish with send_trailers(headers) instead of a final data chunk; gRPC uses trailers to carry its status.

If you need to send a 1xx response, such as 103 Early Hints, call send_informational(status, headers) before the final response head.

You can respond after the call that received the request has returned. Clone the StreamRef to store it or move it into another task.

StreamRef also has the client’s flow-control helpers. Use available_send_capacity to check capacity or send_capacity to wait for it.

Errors and resets

For a problem with one request, send an error response or reset its stream with stream.reset(reason). The connection and its other streams carry on. REFUSED_STREAM tells the client it may safely retry. CANCEL and INTERNAL_ERROR tell it the request failed.

Returning an error from the publish service affects the whole connection. ntex-h2 treats it as a connection fault and sends GOAWAY with INTERNAL_ERROR. It then closes the connection, failing every stream on it.

You can still use ? on send_response and send_payload. They fail only when the stream or connection is already gone. For request-level problems, reset the stream instead of returning an error.

If the client resets a stream while one of your calls for it is still running, that call is cancelled. You still receive an Eof message with the error, so you can clean up per-stream state as shown above.

The control service

Use a control service to log or react to connection-level events. The default control service accepts everything, and most servers can keep it. Pass your own service to .control(..) when you need different handling:

use ntex::error::Error;
use ntex_h2::{Control, Message, OperationError, server};

fn server() -> server::Server<ntex::io::IoBoxed, Error<OperationError>, std::io::Error> {
server::Server::new(async |_msg: Message| Ok::<_, Error<OperationError>>(()))
    .control(async |msg: Control<Error<OperationError>>| {
        if let Control::Disconnect(reason) = &msg {
            log::info!("connection closed: {reason:?}");
        }
        Ok::<_, std::io::Error>(msg.ack())
    })
}

Answer every control message with msg.ack().

Control::Disconnect tells you why a connection ended. The cause may be a protocol error, a GOAWAY from the client, the client going away, or an error returned by the publish service.

Server state

A request handler often needs both application data and connection details, such as a request counter and the peer address. Give the publish and control services that context through connection state. It travels with the I/O object and becomes the state of their pipelines, as described in Carrying state with a request.

A plain Io carries no state, so its state is (). This is why Echo implements Service<(), Message> and its factory takes &().

Worker state created by build_with_config() does not reach these services automatically. To pass it on, put a service in front of Server that reads the worker state and wraps the connection in a State. That is also a convenient place to record the peer address. See Worker and request state for how worker state is created.

The connection state must implement Clone because ntex-h2 clones it for the publish and control factories. Keep shared resources behind an Rc so those copies refer to the same data.

State-aware functions receive the connection state as their first argument. Service factories receive it in create:

use std::{cell::Cell, convert::Infallible, net::SocketAddr, rc::Rc};

use ntex::http::{HeaderMap, StatusCode};
use ntex::io::{Io, types::PeerAddr};
use ntex::service::{State, service};
use ntex::{SharedCfg, error::Error};
use ntex_h2::{Message, MessageKind, OperationError, server};

/// Worker state, created once per worker.
#[derive(Clone)]
struct AppState {
    requests: Rc<Cell<u64>>,
}

/// Connection state, shared by all streams of one connection.
#[derive(Clone)]
struct Conn {
    app: AppState,
    peer: Option<SocketAddr>,
}

async fn accept(
    app: &AppState,
    io: Io,
) -> Result<State<Conn, Io>, server::ServerError<Infallible>> {
    let peer = io.query::<PeerAddr>().get().map(|addr| addr.into_inner());
    let conn = Conn { app: app.clone(), peer };
    Ok(State { req: io, state: conn })
}

async fn publish(conn: &Conn, msg: Message) -> Result<(), Error<OperationError>> {
    if let MessageKind::Headers { pseudo, .. } = msg.kind {
        let requests = &conn.app.requests;
        requests.set(requests.get() + 1);
        println!("{:?} from {:?}, request #{}", pseudo.path, conn.peer, requests.get());

        msg.stream.send_response(StatusCode::OK, HeaderMap::new(), false)?;
        msg.stream.send_payload("hello world", true).await?;
    }
    Ok(())
}

#[ntex::main]
async fn main() -> std::io::Result<()> {
    ntex::server::build_with_config(async || {
        Ok(AppState {
            requests: Rc::new(Cell::new(0)),
        })
    })
    .bind("h2", "127.0.0.1:8080", SharedCfg::default(), async |_: &AppState| {
        service(accept).and_then(server::Server::new(publish))
    })?
    .run()
    .await
}

Each worker has its own AppState. The requests counter therefore counts only requests handled by that worker. Use an Arc for values that must be shared across workers.

Keep per-stream state in the publish service itself, as in the Echo example. A request body that is still arriving belongs there.

With handle_one, you create the publish and control pipelines yourself, so they hold whatever state you gave them.

Shutting down

During shutdown, let existing requests finish without accepting new work. Call ServiceConfig::shutdown() to shut down HTTP/2 connections gracefully. Servers on the current thread then refuse new streams, let active streams finish, and close their connections.

When the ntex server stops, its workers wait for connections to finish, as described in the server chapter.

What you take on

When you use ntex-h2 directly, you take responsibility for policies that ntex::http would otherwise handle. Plan to check that each body matches its content-length, enforce body size limits, and guard against floods of empty DATA frames. You also map stream messages onto requests and responses yourself.

The HTTP/2 example shows a server with buffered and streaming request bodies, backpressure, and trailers, plus a client that calls it. For more examples, see the ntex-h2 examples and the gRPC chapter.