Listeners

A listener listens for incoming request, for example over the network.

A list of default listeners

Name Description
RestListener Listens for incoming JSON/REST requests.
TcpListener Listens for incoming TCP connections.
WebsocketListener Listens for websocket connections.
UdpListener Listens for UDP datagrams, request size limited to MTU.
QuicListener Listens for QUIC connections (not HTTP/3), one request per stream. Requires TLS.
ClusterListener Listens for messages over the local / clustered event bus.

The typical setup for distributed services is to use a gateway listener that listens for requests and then forward these requests over the cluster to the target service. The target service usually runs the ClusterListener.

Starting a listener

Starting a listener is done using the CoreContext,

ListenerSettings settings = new ListenerSettings()
    .setPort(8080) // not applicable to the ClusterListener.
    .setSecure(false);

core.listener(() -> new RestListener()
        .settings(settings)
        .handler(new MyHandler()));

HTTP versions and TLS

The RestListener and the WebsocketListener serve HTTP/1.1 and HTTP/2 by default. Without TLS, HTTP/2 is used by clients that start with it (prior knowledge). With TLS (secure: true, the default) HTTP/2 is negotiated with ALPN, and HTTP/3 is served as well: it uses QUIC, so UDP, on the same port number as the TCP listener. Clients connect with HTTP/1.1 or HTTP/2 first and are told about HTTP/3. HTTP/3 requires TLS: configuring it on a listener that is not secure fails the deployment with a message that says so. ALPN is enabled by default and can be turned off with setAlpn(false).

The versions can be chosen with the configuration of the listener, see below. When versions is set, HTTP/3 is not added to it.

Configuring the transport

ListenerSettings has the properties of a listener that all listeners share (port, TLS, request size, the timeout of requests: a duration, default 3s, a plain number is milliseconds). The settings of the transport itself are the config property. It is read as the configuration class of Vert.x that the listener declares in configType(), and the properties are those of that class,

Listener config is
RestListener, WebsocketListener HttpServerConfig
TcpListener TcpServerConfig
QuicListener QuicServerConfig
UdpListener DatagramSocketOptions
ClusterListener none 
port: 8080
secure: true
keystore: main
timeout: 10s
config:
  versions: [HTTP_1_1, HTTP_2, HTTP_3]
  idleTimeout: PT60S
  compressionConfig:
    compressionEnabled: false
  http1Config:
    maxHeaderSize: 8192

ListenerSettings can be a property of any configuration class, and is loaded together with it, see configuration. Durations are written as ISO-8601 (PT30S) or as a number of seconds.

The properties are checked when the listener is deployed: a property that the class does not have, or an invalid value, fails the deployment with a message that names the class and the property, a misspelled property is not ignored.

Objects that are set in code are used instead of config, which is useful when the configuration needs more than JSON can describe,

new ListenerSettings()
    .setHttpOptions(new HttpServerConfig().setIdleTimeout(Duration.ofSeconds(30)))
    .setTcp(new TcpServerConfig())
    .setSecurity(sslOptions); // TLS options to use instead of the keystore.

Behind a load balancer

Behind a load balancer or reverse proxy every request comes from the address of the proxy, so anything keyed on request.connection().remote() (rate limits, bans, logs) sees one client. List the proxies in proxies (IP addresses or CIDR ranges) and the RestListener and WebsocketListener take the address of the client from the X-Forwarded-For header, but only when the peer is one of them:

port: 6969
proxies: ["10.0.0.0/24"]   # the network of the load balancer, empty (the default) trusts no proxy.

The header is read from the right: the client is the rightmost address that is not a trusted proxy, so a chain of proxies works when all of them are listed. Addresses to the left of it were sent by the client and are ignored, and from a peer that is not trusted the header is ignored as a whole: otherwise any client could choose its own address. List only the proxies themselves, never a range that clients connect from. The proxy must set or append the header (Traefik does by default, nginx with $proxy_add_x_forwarded_for, HAProxy with option forwardfor). An entry that is not an address or a range fails the deployment of the listener. The TCP, UDP and QUIC listeners have no headers and always use the peer.

QUIC

The QuicListener uses QUIC directly, without HTTP/3. Each request is sent on its own bidirectional stream: the client writes the JSON request and ends its side of the stream, and the response is written back on the same stream, which the server then ends. A client can run many requests at once over one connection without head-of-line blocking.

ListenerSettings settings = new ListenerSettings()
    .setPort(4433)
    .setKeystore("main"); // QUIC always uses TLS 1.3, this also sets secure: true.

core.listener(() -> new QuicListener()
        .settings(settings)
        .handler(new MyHandler()));

A Vert.x client:

QuicClient client = vertx.createQuicClient(new ClientSSLOptions()
        .setApplicationLayerProtocols(List.of("chili")));

client.connect(4433, "localhost")
        .compose(QuicConnection::openStream)
        .onSuccess(stream -> {
            Buffer response = Buffer.buffer();
            stream.handler(response::appendBuffer);
            stream.endHandler(done -> System.out.println(response.toJsonObject()));
            stream.end(new JsonObject().put("route", "list").toBuffer());
        });

Using custom listeners

A custom listener can be implemented with the following

// sample listener that retrieves unread smses from an sms gateway API.
public class SmsListener implements CoreListener {
    // SmsGateway is a third-party implementation of an sms gateway.
    private SmsGateway gateway = new SmsGateway("https://some-gateway-service.com/", "api-key");
    private ListenerSettings settings;
    private CoreHandler handler;
    private CoreContext core;
    private boolean running = false;

    @Override
    public void init(CoreContext context) {
        // optional override: only if we need the context.
        this.core = context;            
    }

    @Override
    public CoreListener settings(ListenerSettings settings) {
        this.settings = settings;
        return this;
    }

    @Override
    public CoreListener handler(CoreHandler handler) {
        this.handler = handler;
        return this;
    }

    // optional: the class of the transport configuration that the listener reads from settings.config(..)
    // @Override
    // public Class<?> configType() { return SmsGatewayConfig.class; }
    
    @Override
    public void start(Promise<Void> start) {
        core.periodic(TimerSource.ofMS(1000, "smsGatewayPoll"), (id) -> {
            if (running) {
                List<JsonObject> smses = smsGateway.fetchUnread();
                
                for (JsonObject sms: smses) {
                    handler.handle(new SmsRequest(smsGateway, sms));
                }
            } else {
                core.cancel(id);
            }
        });
        
        running = true;
        start.complete();
    }
    
    @Override
    public void stop(Promise<Void> stop) {
        // optional override: only if cleanup is required.
        running = false;
        stop.complete();
    }

}

A request object is required to handle responses and usage in the handler/protocol classes.

public class Smsrequest implements Request {
    private AtomicBoolean written = new AtomicBoolean(false);
    private SmsGateway gateway;
    private JsonObject body;
    
    // the constructor needs a context and the message body.
    public SmsRequest(SmsGateway gateway, JsonObject body) {
        this.gateway = gateway;
        this.body = body;
    }
    
    @Override
    public JsonObject data() {
        // the body can use any format, as long as we can use the JsonObject
        // API to interact with it.
        return body;
    }
    
    @Override
    public void write(Object object) {
        // the object can be an error or a response message.
        // serialize it to whichever format is used by the API.
        gateway.send(Serializer.json(object));
        
        written.set(true);    
    }
    
    @Override
    public void connection() { 
        // there is no connection - our fictive API is using REST.
        // we can create a connection-like object anyways before a response is written.
        return new Connection((body) -> {
            
            if (written.get()) {
                throw new CoreRuntimeException("Connection closed: response already written.");
            } else {
                write(body);            
                written.set(true);
            }
            
        }, UUID.randomUUID().toString());    
    }
    
    @Override
    public int size() {
        // should return the number of bytes in the message body.
        return body.size();
    }
}

Other examples that may be usable