AsyncHttpServer extends AbstractServer<AsyncHttpServer> { public static AsyncHttpServer create(@NotNull Eventloop eventloop, @NotNull AsyncServlet servlet) { return new AsyncHttpServer(eventloop, servlet); } private  AsyncHttpServer(@NotNull Eventloop eventloop, @NotNull AsyncServlet servlet); static AsyncHttpServer create(@NotNull Eventloop eventloop, @NotNull AsyncServlet servlet); AsyncHttpServer withKeepAliveTimeout(@NotNull Duration keepAliveTime); AsyncHttpServer withMaxKeepAliveRequests(int maxKeepAliveRequests); AsyncHttpServer withNoKeepAlive(); AsyncHttpServer withReadWriteTimeout(@NotNull Duration readWriteTimeout); AsyncHttpServer withReadWriteTimeout(@NotNull Duration readWriteTimeout, @NotNull Duration readWriteTimeoutShutdown); AsyncHttpServer withServeTimeoutShutdown(@NotNull Duration serveTimeoutShutdown); AsyncHttpServer withMaxBodySize(MemSize maxBodySize); AsyncHttpServer withMaxBodySize(int maxBodySize); AsyncHttpServer withHttpErrorFormatter(@NotNull HttpExceptionFormatter httpExceptionFormatter); AsyncHttpServer withInspector(Inspector inspector); Duration getKeepAliveTimeout(); Duration getReadWriteTimeout(); Promise<Void> getCloseNotification(); List<String> getHttpAddresses(); @JmxAttribute(description = "current number of connections", reducer = JmxReducerSum.class) int getConnectionsCount(); @JmxAttribute(reducer = JmxReducerSum.class) int getConnectionsNewCount(); @JmxAttribute(reducer = JmxReducerSum.class) int getConnectionsReadWriteCount(); @JmxAttribute(reducer = JmxReducerSum.class) int getConnectionsServingCount(); @JmxAttribute(reducer = JmxReducerSum.class) int getConnectionsKeepAliveCount(); @JmxAttribute(reducer = JmxReducerSum.class) int getConnectionsKeepAliveExpired(); @JmxAttribute(reducer = JmxReducerSum.class) int getConnectionsReadWriteExpired(); @JmxAttribute(name = "") @Nullable JmxInspector getStats(); @Override String toString(); static final Duration READ_WRITE_TIMEOUT; static final Duration READ_WRITE_TIMEOUT_SHUTDOWN; static final Duration SERVE_TIMEOUT_SHUTDOWN; static final Duration KEEP_ALIVE_TIMEOUT; static final MemSize MAX_BODY_SIZE; static final int MAX_KEEP_ALIVE_REQUESTS; }
@Test public void testClosed() throws Exception { Eventloop eventloop = Eventloop.create().withFatalErrorHandler(rethrowOnAnyError()).withCurrentThread(); int port = getFreePort(); AsyncHttpServer server = blockingHttpServer(eventloop, port); server.listen(); Thread thread = new Thread(eventloop); thread.start(); Socket socket = new Socket(); socket.connect(new InetSocketAddress("localhost", port)); writeByRandomParts(socket, "GET /abc HTTP1.1\r\nHost: localhost\r\n"); socket.close(); server.closeFuture().get(); thread.join(); }
@Test public void testBodySupplierClosingOnDisconnect() throws Exception { Eventloop eventloop = Eventloop.create().withFatalErrorHandler(rethrowOnAnyError()).withCurrentThread(); int port = getFreePort(); SettablePromise<Throwable> throwablePromise = new SettablePromise<>(); ChannelSupplier<ByteBuf> supplier = ChannelSupplier.of(() -> Promise.of(wrapAscii("Hello")), throwablePromise::set); AsyncHttpServer server = AsyncHttpServer.create(eventloop, req -> HttpResponse.ok200().withBodyStream(supplier)) .withListenPort(port) .withAcceptOnce(); server.listen(); new Thread(() -> { try { Socket socket = new Socket(); socket.connect(new InetSocketAddress("localhost", port)); writeByRandomParts(socket, "GET /abc HTTP1.1\r\nHost: localhost\r\n\r\n"); socket.close(); } catch (IOException e) { throw new AssertionError(e); } }).start(); Throwable throwable = await(throwablePromise); assertThat(throwable, instanceOf(IOException.class)); }
@Test public void testNoKeepAlive_Http_1_0() throws Exception { Eventloop eventloop = Eventloop.create().withFatalErrorHandler(rethrowOnAnyError()).withCurrentThread(); int port = getFreePort(); AsyncHttpServer server = blockingHttpServer(eventloop, port); server.withListenPort(port); server.listen(); Thread thread = new Thread(eventloop); thread.start(); Socket socket = new Socket(); socket.connect(new InetSocketAddress("localhost", port)); writeByRandomParts(socket, "GET /abc HTTP/1.0\r\nHost: localhost\r\n\r\n"); readAndAssert(socket.getInputStream(), "HTTP/1.1 200 OK\r\nConnection: close\r\nContent-Length: 4\r\n\r\n/abc"); assertEquals(0, toByteArray(socket.getInputStream()).length); socket.close(); server.closeFuture().get(); thread.join(); }
@Test public void testNoKeepAlive_Http_1_1() throws Exception { Eventloop eventloop = Eventloop.create().withFatalErrorHandler(rethrowOnAnyError()).withCurrentThread(); int port = getFreePort(); AsyncHttpServer server = blockingHttpServer(eventloop, port); server.withListenPort(port); server.listen(); Thread thread = new Thread(eventloop); thread.start(); Socket socket = new Socket(); socket.connect(new InetSocketAddress("localhost", port)); writeByRandomParts(socket, "GET /abc HTTP/1.1\r\nConnection: close\r\nHost: localhost\r\n\r\n"); readAndAssert(socket.getInputStream(), "HTTP/1.1 200 OK\r\nConnection: close\r\nContent-Length: 4\r\n\r\n/abc"); assertEquals(0, toByteArray(socket.getInputStream()).length); socket.close(); server.closeFuture().get(); thread.join(); }
@Test public void testPipelining() throws Exception { Eventloop eventloop = Eventloop.create().withFatalErrorHandler(rethrowOnAnyError()).withCurrentThread(); int port = getFreePort(); doTestPipelining(eventloop, delayedHttpServer(eventloop, port), port); }
@Test @Ignore("does not work") public void testPipelining2() throws Exception { Eventloop eventloop = Eventloop.create().withFatalErrorHandler(rethrowOnAnyError()).withCurrentThread(); int port = getFreePort(); doTestPipelining2(eventloop, delayedHttpServer(eventloop, port), port); }
@Test public void testBigHttpMessage() throws Exception { int port = getFreePort(); Eventloop eventloop = Eventloop.create().withFatalErrorHandler(rethrowOnAnyError()).withCurrentThread(); byte[] body = encodeAscii("Test big HTTP message body"); HttpRequest request = HttpRequest.post("http: .withBody(body); ByteBuf buf = ByteBufPool.allocate(request.estimateSize() + body.length); request.writeTo(buf); buf.put(body); AsyncHttpServer server = AsyncHttpServer.create(eventloop, req -> HttpResponse.ok200() .withBody(encodeAscii(req.getUrl().getPathAndQuery()))) .withListenPort(port); server.listen(); Thread thread = new Thread(eventloop); thread.start(); try (Socket socket = new Socket()) { socket.connect(new InetSocketAddress("localhost", port)); socket.getOutputStream().write(buf.array(), buf.head(), buf.readRemaining()); buf.recycle(); Thread.sleep(100); } server.closeFuture().get(); thread.join(); }
@Test public void testExpectContinue() throws Exception { Eventloop eventloop = Eventloop.create().withFatalErrorHandler(rethrowOnAnyError()).withCurrentThread(); int port = getFreePort(); AsyncHttpServer server = AsyncHttpServer.create(eventloop, request -> request.loadBody().map(body -> HttpResponse.ok200().withBody(body.slice()))) .withListenPort(port); server.listen(); Thread thread = new Thread(eventloop); thread.start(); Socket socket = new Socket(); socket.setTcpNoDelay(true); socket.connect(new InetSocketAddress("localhost", port)); writeByRandomParts(socket, "POST /abc HTTP/1.0\r\nHost: localhost\r\nContent-Length: 5\r\nExpect: 100-continue\r\n\r\n"); readAndAssert(socket.getInputStream(), "HTTP/1.1 100 Continue\r\n\r\n"); writeByRandomParts(socket, "abcde"); readAndAssert(socket.getInputStream(), "HTTP/1.1 200 OK\r\nConnection: close\r\nContent-Length: 5\r\n\r\nabcde"); assertEquals(0, toByteArray(socket.getInputStream()).length); assertTrue(socket.isClosed()); socket.close(); server.closeFuture().get(); thread.join(); }
@Test public void testBodyRecycledOnce() throws IOException, InterruptedException { int port = getFreePort(); Eventloop eventloop = Eventloop.create().withCurrentThread().withFatalErrorHandler(rethrowOnAnyError()); AsyncHttpServer server = AsyncHttpServer.create(eventloop, request -> { shutdownAllChannels(eventloop); return HttpResponse.ok200(); }) .withListenPort(port) .withAcceptOnce(true); server.listen(); Thread thread = new Thread(() -> { try (Socket socket = new Socket()) { socket.connect(new InetSocketAddress("localhost", port)); ByteBuf buf = ByteBuf.wrapForReading(encodeAscii("GET / HTTP/1.1\r\nHost: localhost\r\n" + "Connection: close\r\nContent-Length: 10\r\n\r\ntest")); socket.getOutputStream().write(buf.array(), buf.head(), buf.readRemaining()); } catch (IOException e) { throw new RuntimeException(e); } }); thread.start(); eventloop.run(); thread.join(); }