Skip to content

Server and client side async execution pipeline assemblers #616

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ok2c wants to merge 1 commit into
base: master
Choose a base branch
from
+3,158 −956
Open

Server and client side async execution pipeline assemblers #616

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
76 changes: 39 additions & 37 deletions .../src/test/java/org/apache/hc/core5/http2/examples/H2ConscriptRequestExecutionExample.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,9 +42,8 @@
import org.apache.hc.core5.http.Message;
import org.apache.hc.core5.http.impl.bootstrap.HttpAsyncRequester;
import org.apache.hc.core5.http.nio.AsyncClientEndpoint;
import org.apache.hc.core5.http.nio.entity.StringAsyncEntityConsumer;
import org.apache.hc.core5.http.nio.support.AsyncRequestBuilder;
import org.apache.hc.core5.http.nio.support.BasicResponseConsumer;
import org.apache.hc.core5.http.nio.support.AsyncClientPipeline;
import org.apache.hc.core5.http.protocol.HttpCoreContext;
import org.apache.hc.core5.http2.HttpVersionPolicy;
import org.apache.hc.core5.http2.config.H2Config;
import org.apache.hc.core5.http2.frame.RawFrame;
Expand Down Expand Up @@ -127,43 +126,46 @@ public void onOutputFlowControl(final HttpConnection connection, final int strea
final Future<AsyncClientEndpoint> future = requester.connect(target, Timeout.ofDays(5));
final AsyncClientEndpoint clientEndpoint = future.get();

final String[] requestUris = new String[] {"/httpbin/ip", "/httpbin/user-agent", "/httpbin/headers"};
final String[] requestUris = new String[]{"/httpbin/ip", "/httpbin/user-agent", "/httpbin/headers"};

final CountDownLatch latch = new CountDownLatch(requestUris.length);
for (final String requestUri: requestUris) {
for (final String requestUri : requestUris) {
clientEndpoint.execute(
AsyncRequestBuilder.get()
.setHttpHost(target)
.setPath(requestUri)
.build(),
new BasicResponseConsumer<>(new StringAsyncEntityConsumer()),
new FutureCallback<Message<HttpResponse, String>>() {

@Override
public void completed(final Message<HttpResponse, String> message) {
clientEndpoint.releaseAndReuse();
final HttpResponse response = message.head();
final String body = message.body();
System.out.println(requestUri + "->" + response.getCode() + " " + response.getVersion());
System.out.println(body);
latch.countDown();
}

@Override
public void failed(final Exception ex) {
clientEndpoint.releaseAndDiscard();
System.out.println(requestUri + "->" + ex);
latch.countDown();
}

@Override
public void cancelled() {
clientEndpoint.releaseAndDiscard();
System.out.println(requestUri + " cancelled");
latch.countDown();
}

});
AsyncClientPipeline.assemble()
.request()
.get(target, requestUri)
.response()
.asString()
.result(new FutureCallback<Message<HttpResponse, String>>() {

@Override
public void completed(final Message<HttpResponse, String> message) {
clientEndpoint.releaseAndReuse();
final HttpResponse response = message.head();
final String body = message.body();
System.out.println(requestUri + "->" + response.getCode() + " " + response.getVersion());
System.out.println(body);
latch.countDown();
}

@Override
public void failed(final Exception ex) {
clientEndpoint.releaseAndDiscard();
System.out.println(requestUri + "->" + ex);
latch.countDown();
}

@Override
public void cancelled() {
clientEndpoint.releaseAndDiscard();
System.out.println(requestUri + " cancelled");
latch.countDown();
}

})
.create(),
null,
HttpCoreContext.create());
}

latch.await();
Expand Down
152 changes: 73 additions & 79 deletions httpcore5-h2/src/test/java/org/apache/hc/core5/http2/examples/H2FileServerExample.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,34 +27,36 @@
package org.apache.hc.core5.http2.examples;

import java.io.File;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import org.apache.hc.core5.function.Supplier;
import org.apache.hc.core5.http.ContentType;
import org.apache.hc.core5.http.EndpointDetails;
import org.apache.hc.core5.http.EntityDetails;
import org.apache.hc.core5.http.Header;
import org.apache.hc.core5.http.HttpConnection;
import org.apache.hc.core5.http.HttpException;
import org.apache.hc.core5.http.HttpRequest;
import org.apache.hc.core5.http.HttpResponse;
import org.apache.hc.core5.http.HttpStatus;
import org.apache.hc.core5.http.Message;
import org.apache.hc.core5.http.Method;
import org.apache.hc.core5.http.ProtocolException;
import org.apache.hc.core5.http.URIScheme;
import org.apache.hc.core5.http.impl.bootstrap.HttpAsyncServer;
import org.apache.hc.core5.http.nio.AsyncRequestConsumer;
import org.apache.hc.core5.http.nio.AsyncServerRequestHandler;
import org.apache.hc.core5.http.nio.entity.AsyncEntityProducers;
import org.apache.hc.core5.http.nio.entity.DiscardingEntityConsumer;
import org.apache.hc.core5.http.nio.support.AsyncResponseBuilder;
import org.apache.hc.core5.http.nio.support.BasicRequestConsumer;
import org.apache.hc.core5.http.protocol.HttpContext;
import org.apache.hc.core5.http.impl.routing.RequestRouter;
import org.apache.hc.core5.http.message.BasicHttpResponse;
import org.apache.hc.core5.http.nio.AsyncServerExchangeHandler;
import org.apache.hc.core5.http.nio.entity.FileEntityProducer;
import org.apache.hc.core5.http.nio.entity.StringAsyncEntityProducer;
import org.apache.hc.core5.http.nio.support.AsyncServerPipeline;
import org.apache.hc.core5.http.nio.support.BasicResponseProducer;
import org.apache.hc.core5.http.protocol.HttpCoreContext;
import org.apache.hc.core5.http.protocol.HttpDateGenerator;
import org.apache.hc.core5.http2.HttpVersionPolicy;
import org.apache.hc.core5.http2.frame.RawFrame;
import org.apache.hc.core5.http2.impl.nio.H2StreamListener;
Expand Down Expand Up @@ -87,6 +89,59 @@ public static void main(final String[] args) throws Exception {
.setTcpNoDelay(true)
.build();

final Supplier<AsyncServerExchangeHandler> exchangeHandlerSupplier = AsyncServerPipeline.assemble()
// Read GET / HEAD requests ignoring their content body
.request(Method.GET, Method.HEAD)
.ignoreContent()
// Write out responses by streaming out content of a file
.response()
.<Message<HttpResponse, File>>produce(m -> {
if (m.error() == null) {
final File file = m.getBody();
final ContentType contentType;
final String filename = TextUtils.toLowerCase(file.getName());
if (filename.endsWith(".txt")) {
contentType = ContentType.TEXT_PLAIN;
} else if (filename.endsWith(".html") || filename.endsWith(".htm")) {
contentType = ContentType.TEXT_HTML;
} else if (filename.endsWith(".xml")) {
contentType = ContentType.TEXT_XML;
} else {
contentType = ContentType.DEFAULT_BINARY;
}
return new BasicResponseProducer(new FileEntityProducer(file, contentType));
} else {
return new BasicResponseProducer(new StringAsyncEntityProducer(Objects.toString(m.error()), ContentType.TEXT_PLAIN));
}
})
// Map exceptions to a response message
.errorMessage(Throwable::getMessage)
// Generate a response to the request
.handle((m, context) -> {
final HttpRequest request = m.head();
final URI requestUri;
try {
requestUri = request.getUri();
} catch (final URISyntaxException ex) {
throw new ProtocolException(ex.getMessage(), ex);
}
final String path = requestUri.getPath();
final File file = new File(docRoot, path);
if (!file.exists()) {
println("File " + file.getPath() + " not found");
return Message.error(new BasicHttpResponse(HttpStatus.SC_NOT_FOUND), "File not found");
} else if (!file.canRead() || file.isDirectory()) {
println("Cannot read file " + file.getPath());
return Message.error(new BasicHttpResponse(HttpStatus.SC_FORBIDDEN), "File cannot be accessed");
} else {
final HttpCoreContext coreContext = HttpCoreContext.cast(context);
final EndpointDetails endpoint = coreContext.getEndpointDetails();
println(endpoint + " | serving file " + file.getPath());
return Message.of(new BasicHttpResponse(HttpStatus.SC_OK), file);
}
})
.supplier();

final HttpAsyncServer server = H2ServerBootstrap.bootstrap()
.setIOReactorConfig(config)
.setVersionPolicy(HttpVersionPolicy.FORCE_HTTP_2)
Expand Down Expand Up @@ -123,75 +178,10 @@ public void onOutputFlowControl(final HttpConnection connection, final int strea
}

})
.register("*", new AsyncServerRequestHandler<Message<HttpRequest, Void>>() {

@Override
public AsyncRequestConsumer<Message<HttpRequest, Void>> prepare(
final HttpRequest request,
final EntityDetails entityDetails,
final HttpContext context) throws HttpException {
return new BasicRequestConsumer<>(entityDetails != null ? new DiscardingEntityConsumer<>() : null);
}

@Override
public void handle(
final Message<HttpRequest, Void> message,
final ResponseTrigger responseTrigger,
final HttpContext localContext) throws HttpException, IOException {
final HttpCoreContext context = HttpCoreContext.cast(localContext);
final HttpRequest request = message.head();
final URI requestUri;
try {
requestUri = request.getUri();
} catch (final URISyntaxException ex) {
throw new ProtocolException(ex.getMessage(), ex);
}
final String path = requestUri.getPath();
final File file = new File(docRoot, path);
if (!file.exists()) {

System.out.println("File " + file.getPath() + " not found");
responseTrigger.submitResponse(
AsyncResponseBuilder.create(HttpStatus.SC_NOT_FOUND)
.setEntity("<html><body><h1>File" + file.getPath() +
" not found</h1></body></html>", ContentType.TEXT_HTML)
.build(),
context);

} else if (!file.canRead() || file.isDirectory()) {

System.out.println("Cannot read file " + file.getPath());
responseTrigger.submitResponse(
AsyncResponseBuilder.create(HttpStatus.SC_FORBIDDEN)
.setEntity("<html><body><h1>Access denied</h1></body></html>", ContentType.TEXT_HTML)
.build(),
context);

} else {

final ContentType contentType;
final String filename = TextUtils.toLowerCase(file.getName());
if (filename.endsWith(".txt")) {
contentType = ContentType.TEXT_PLAIN;
} else if (filename.endsWith(".html") || filename.endsWith(".htm")) {
contentType = ContentType.TEXT_HTML;
} else if (filename.endsWith(".xml")) {
contentType = ContentType.TEXT_XML;
} else {
contentType = ContentType.DEFAULT_BINARY;
}

final EndpointDetails endpoint = context.getEndpointDetails();
System.out.println(endpoint + ": serving file " + file.getPath());
responseTrigger.submitResponse(
AsyncResponseBuilder.create(HttpStatus.SC_OK)
.setEntity(AsyncEntityProducers.create(file, contentType))
.build(),
context);
}
}

})
.setRequestRouter(RequestRouter.<Supplier<AsyncServerExchangeHandler>>builder()
.addRoute(RequestRouter.LOCAL_AUTHORITY, "*", exchangeHandlerSupplier)
.resolveAuthority(RequestRouter.LOCAL_AUTHORITY_RESOLVER)
.build())
.create();

Runtime.getRuntime().addShutdownHook(new Thread(() -> {
Expand All @@ -206,4 +196,8 @@ public void handle(
server.awaitShutdown(TimeValue.ofDays(Long.MAX_VALUE));
}

static void println(final String msg) {
System.out.println(HttpDateGenerator.INSTANCE.getCurrentDate() + " | " + msg);
}

}
Loading
Loading