HTTP Channel Adapter

The http-channel-adapter module implements an inbound HTTP server based on Undertow 2.2.x.

Dependency

<dependency>
    <groupId>io.pipelite</groupId>
    <artifactId>http-channel-adapter</artifactId>
    <version>1.0.0-SNAPSHOT</version>
</dependency>

Usage as source (inbound)

Pipelite.defineFlow("http-inbound")
    .fromSource("http://ingress")     (1)
    .process("handle", (exchange, contribution) -> {
        String body = exchange.getInputPayloadAs(String.class);
        exchange.setOutputPayload("OK");
    })
    .toSink("slf4j://log")
    .build();
1 Opens an HTTP listener on port 8080 (default) at path /ingress.

A request is answered only once the framework has accepted its exchange (written to the durable inbox and queued, applying the backpressure of the flow). If the exchange cannot be accepted, the request is answered with 500.

The same method restriction can be applied through the typed HttpSourceConfigurer instead of a query string (see Custom Channel Adapters for the general mechanism):

.fromSource("http://orders", (HttpSourceConfigurer c) -> c.method("POST"))

URL format

http://<path>[?method=<HTTP method>]
Parameter Default Description

path

(required)

The HTTP path on which the server listens (e.g. /api/v1/events)

method

(unset - any method accepted)

Restricts this resource to a single HTTP method (e.g. POST). A request with a different method receives a 405; a request to an unmapped path entirely receives a 404.

The Undertow server is started automatically by HttpChannelAdapter.onContextStarted() upon the first registration of an HTTP consumer.

Adapter-wide configuration

Port, host and the maximum size of a single request body are shared by every http:// source in a context - one Undertow instance serves all of them - so they are configured once, adapter-wide, the same way Kafka’s bootstrapServers is:

context.addChannelConfigurer((HttpChannelConfigurer) configuration -> {
    configuration.setPort(8443);
    configuration.setHost("127.0.0.1");
    configuration.setMaxEntitySize(1024 * 1024); // bytes
});

With Spring Boot, register it from a bean implementing PipeliteContextInitializer instead - see Configuration Properties. A plain @Bean returning an HttpChannelConfigurer is not picked up automatically.

Setting Default Description

port

8080

The Undertow listener’s TCP port.

host

0.0.0.0

The interface Undertow binds to.

maxEntitySize

10 MiB

Upper bound on a single request’s body size, in bytes. A request exceeding it receives a 413 response instead of being read into memory unbounded.
HttpChannelAdapter has no built-in authentication and does not terminate TLS - it is not safe to expose directly on an untrusted network. Put a reverse proxy (nginx, an API gateway, a load balancer) in front of it to handle TLS termination and authentication before traffic reaches this adapter.

Usage as sink (outbound)

Not implemented yet: the HTTP endpoint has no producer (it keeps the default of Endpoint#createProducer(), which throws UnsupportedOperationException), so .toSink("http://…​")/.toChannel("http://…​") fail at flow-registration time. http-channel-adapter is inbound-only today.

Multiple path registration

Multiple paths can be registered on the same server:

context.registerFlowDefinition(
    Pipelite.defineFlow("flow-orders")
        .fromSource("http://orders").toSink("slf4j://orders").build()
);
context.registerFlowDefinition(
    Pipelite.defineFlow("flow-payments")
        .fromSource("http://payments").toSink("slf4j://payments").build()
);

HttpChannelAdapter maintains a per-path consumer map and shares a single Undertow instance.

Registering two flows on the same HTTP path is not allowed: IllegalStateException will be thrown.