content: package io.activej.http;
  
  import io.activej.async.process.AsyncCloseable;
  import io.activej.bytebuf.ByteBuf;
  import io.activej.bytebuf.ByteBufPool;
  import io.activej.common.exception.TruncatedDataException;
  import io.activej.common.exception.UnknownFormatException;
  import io.activej.common.ref.Ref;
  import io.activej.csp.supplier.ChannelSupplier;
  import io.activej.csp.supplier.ChannelSuppliers;
  import io.activej.eventloop.Eventloop;
  import io.activej.http.HttpServer.JmxInspector;
  import io.activej.jmx.stats.ExceptionStats;
  import io.activej.promise.Promise;
  import io.activej.promise.Promises;
  import io.activej.promise.SettablePromise;
  import io.activej.reactor.Reactor;
  import io.activej.test.rules.ByteBufRule;
  import io.activej.test.rules.EventloopRule;
  import org.junit.Before;
  import org.junit.ClassRule;
  import org.junit.Test;
  
  import java.io.IOException;
  import java.io.InputStream;
  import java.net.InetSocketAddress;
  import java.net.Socket;
  import java.nio.channels.Selector;
  import java.time.Duration;
  import java.util.*;
  import java.util.concurrent.CountDownLatch;
  import java.util.concurrent.ExecutionException;
  
  import static io.activej.bytebuf.ByteBufStrings.*;
  import static io.activej.http.TestUtils.assertEmpty;
  import static io.activej.http.TestUtils.readFully;
  import static io.activej.promise.TestUtils.await;
  import static io.activej.test.TestUtils.getFreePort;
  import static java.lang.Math.min;
  import static org.hamcrest.CoreMatchers.instanceOf;
  import static org.hamcrest.MatcherAssert.assertThat;
  import static org.junit.Assert.*;
  
  public final class HttpServerTest {
  	@ClassRule
  	public static final ByteBufRule byteBufRule = new ByteBufRule();
  
  	@ClassRule
  	public static final EventloopRule eventloopRule = new EventloopRule();
  
  	private Eventloop eventloop;
  	private int port;
  
  	@Before
  	public void setUp() {
  		eventloop = Reactor.getCurrentReactor();
  		port = getFreePort();
  	}
  
  	public HttpServer blockingHttpServer() {
  		return HttpServer.builder(eventloop,
  				request ->
  					HttpResponse.ok200().withBody(encodeAscii(request.getUrl().getPathAndQuery())).toPromise())
  			.withListenPort(port)
  			.build();
  	}
  
  	public HttpServer httpServer() {
  		return HttpServer.builder(eventloop,
  				request ->
  					HttpResponse.ok200().withBody(encodeAscii(request.getUrl().getPathAndQuery())).toPromise().async())
  			.withListenPort(port)
  			.build();
  	}
  
  	static final Random RANDOM = new Random();
  
  	public HttpServer delayedHttpServer() {
  		return HttpServer.builder(eventloop,
  				request -> Promises.delay(RANDOM.nextInt(3),
  					HttpResponse.ok200().withBody(encodeAscii(request.getUrl().getPathAndQuery())).build()))
  			.withListenPort(port)
  			.build();
  	}
  
  	public static void writeByRandomParts(Socket socket, String string) throws IOException {
  		writeByRandomParts(socket, ByteBuf.wrapForReading(encodeAscii(string)));
  	}
  
  	public static void writeByRandomParts(Socket socket, ByteBuf buf) throws IOException {
  		Random random = new Random();
  		while (buf.canRead()) {
  			int count = min(1 + random.nextInt(5), buf.readRemaining());
  			socket.getOutputStream().write(buf.array(), buf.head(), count);
  			buf.moveHead(count);
  		}
  		buf.recycle();
  	}
  
  	public static void readAndAssert(InputStream is, String expected) {
  		byte[] bytes = new byte[expected.length()];
  		readFully(is, bytes);
  		String actual = decodeAscii(bytes);
  		assertEquals(new LinkedHashSet<>(List.of(expected.split("\r\n"))), new LinkedHashSet<>(List.of(actual.split("\r\n"))));
  	}
  
  	@Test
  	public void testKeepAlive_Http_1_0() throws Exception {
  		doTestKeepAlive_Http_1_0(blockingHttpServer());
  		doTestKeepAlive_Http_1_0(httpServer());
  		doTestKeepAlive_Http_1_0(delayedHttpServer());
  	}
  
  	private void doTestKeepAlive_Http_1_0(HttpServer server) throws Exception {
  		server.listen();
  		Thread thread = new Thread(eventloop);
  		thread.start();
  
  		try (Socket socket = new Socket()) {
  			socket.setTcpNoDelay(true);
  			socket.connect(new InetSocketAddress("localhost", port));
  
  			for (int i = 0; i < 200; i++) {
  				writeByRandomParts(socket, """
  					GET /abc HTTP/1.0\r
  					Host: localhost\r
  					Connection: keep-alive\r
  					\r
  					""");
  				readAndAssert(socket.getInputStream(), """
  					HTTP/1.1 200 OK\r
  					Connection: keep-alive\r
  					Content-Length: 4\r
  					\r
  					/abc""");
  			}
  
  			writeByRandomParts(socket, """
  				GET /abc HTTP/1.0\r
  				Host: localhost\r
  				\r
  				""");
  			readAndAssert(socket.getInputStream(), """
  				HTTP/1.1 200 OK\r
  				Connection: close\r
  				Content-Length: 4\r
  				\r
  				/abc"""); // ?
  
  			assertEmpty(socket.getInputStream());
  			assertTrue(socket.isClosed());
  		}
  
  		server.closeFuture().get();
  		thread.join();
  		resetPort();
  	}
  
  	@Test
  	public void testKeepAlive_Http_1_1() throws Exception {
  		doTestKeepAlive_Http_1_1(blockingHttpServer());
  		doTestKeepAlive_Http_1_1(httpServer());
  		doTestKeepAlive_Http_1_1(delayedHttpServer());
  	}
  
  	private void doTestKeepAlive_Http_1_1(HttpServer server) throws Exception {
  		server.listen();
  		Thread thread = new Thread(eventloop);
  		thread.start();
  
  		try (Socket socket = new Socket()) {
  			socket.setTcpNoDelay(true);
  			socket.connect(new InetSocketAddress("localhost", port));
  
  			for (int i = 0; i < 200; i++) {
  				writeByRandomParts(socket, """
  					GET /abc HTTP/1.1\r
  					Host: localhost\r
  					\r
  					""");
  				readAndAssert(socket.getInputStream(), """
  					HTTP/1.1 200 OK\r
  					Connection: keep-alive\r
  					Content-Length: 4\r
  					\r
  					/abc""");
  			}
  
  			writeByRandomParts(socket, """
  				GET /abc HTTP/1.1\r
  				Host: localhost\r
  				Connection: close\r
  				\r
  				""");
  			readAndAssert(socket.getInputStream(), """
  				HTTP/1.1 200 OK\r
  				Connection: close\r
  				Content-Length: 4\r
  				\r
  				/abc"""); // ?
  
  			assertEmpty(socket.getInputStream());
  			assertTrue(socket.isClosed());
  		}
  
  		server.closeFuture().get();
  		thread.join();
  		resetPort();
  	}
  
  	@Test
  	public void testClosed() throws Exception {
  		HttpServer server = blockingHttpServer();
  		server.listen();
  		Thread thread = new Thread(eventloop);
  		thread.start();
  
  		try (Socket socket = new Socket()) {
  			socket.connect(new InetSocketAddress("localhost", port));
  			writeByRandomParts(socket, """
  				GET /abc HTTP1.1\r
  				Host: localhost\r
  				""");
  		}
  
  		server.closeFuture().get();
  		thread.join();
  	}
  
  	@Test
  	public void testBodySupplierClosingOnDisconnect() throws Exception {
  		SettablePromise<Exception> exceptionPromise = new SettablePromise<>();
  		ChannelSupplier<ByteBuf> supplier = ChannelSuppliers.ofAsyncSupplier(() -> Promise.of(wrapAscii("Hello")), AsyncCloseable.of(exceptionPromise::set));
  		HttpServer.builder(eventloop, req -> HttpResponse.ok200().withBodyStream(supplier).toPromise())
  			.withListenPort(port)
  			.withAcceptOnce()
  			.build()
  			.listen();
  		new Thread(() -> {
  			try (Socket socket = new Socket()) {
  				socket.connect(new InetSocketAddress("localhost", port));
  				writeByRandomParts(socket, """
  					GET /abc HTTP/1.1\r
  					Host: localhost\r
  					\r
  					""");
  			} catch (IOException e) {
  				throw new AssertionError(e);
  			}
  		}).start();
  		Exception exception = await(exceptionPromise);
  		assertThat(exception, instanceOf(IOException.class));
  	}
  
  	@Test
  	public void testNoKeepAlive_Http_1_0() throws Exception {
  		HttpServer server = blockingHttpServer();
  		server.listen();
  		Thread thread = new Thread(eventloop);
  		thread.start();
  
  		try (Socket socket = new Socket()) {
  			socket.connect(new InetSocketAddress("localhost", port));
  			writeByRandomParts(socket, """
  				GET /abc HTTP/1.0\r
  				Host: localhost\r
  				\r
  				""");
  			readAndAssert(socket.getInputStream(), """
  				HTTP/1.1 200 OK\r
  				Connection: close\r
  				Content-Length: 4\r
  				\r
  				/abc""");
  			assertEmpty(socket.getInputStream());
  		}
  
  		server.closeFuture().get();
  		thread.join();
  	}
  
  	@Test
  	public void testNoKeepAlive_Http_1_1() throws Exception {
  		HttpServer server = blockingHttpServer();
  		server.listen();
  		Thread thread = new Thread(eventloop);
  		thread.start();
  
  		try (Socket socket = new Socket()) {
  			socket.connect(new InetSocketAddress("localhost", port));
  			writeByRandomParts(socket, """
  				GET /abc HTTP/1.1\r
  				Connection: close\r
  				Host: localhost\r
  				\r
  				""");
  			readAndAssert(socket.getInputStream(), """
  				HTTP/1.1 200 OK\r
  				Connection: close\r
  				Content-Length: 4\r
  				\r
  				/abc""");
  			assertEmpty(socket.getInputStream());
  		}
  
  		server.closeFuture().get();
  		thread.join();
  	}
  
  	@Test
  	public void testPipelining() throws Exception {
  		doTestPipelining(blockingHttpServer());
  		doTestPipelining(httpServer());
  		doTestPipelining(delayedHttpServer());
  	}
  
  	@Test
  	public void testActiveRequestsWithPipelining() throws Exception {
  		JmxInspector inspector = new JmxInspector();
  		HttpServer server = HttpServer.builder(eventloop,
  				request ->
  					HttpResponse.ok200().withBody(encodeAscii(request.getUrl().getPathAndQuery())).toPromise())
  			.withListenPort(port)
  			.withInspector(inspector)
  			.build();
  
  		doTestPipelining(server);
  
  		assertEquals(400, inspector.getTotalRequests().getTotalCount());
  		assertEquals(0, inspector.getActiveRequests());
  	}
  
  	private void doTestPipelining(HttpServer server) throws Exception {
  		server.listen();
  		Thread thread = new Thread(eventloop);
  		thread.start();
  
  		try (Socket socket = new Socket()) {
  			socket.connect(new InetSocketAddress("localhost", port));
  
  			for (int i = 0; i < 100; i++) {
  				writeByRandomParts(socket, """
  					GET /abc HTTP/1.1\r
  					Connection: Keep-Alive\r
  					Host: localhost\r
  					\r
  					GET /123456 HTTP/1.1\r
  					Host: localhost\r
  					\r
  					POST /post1 HTTP/1.1\r
  					Host: localhost\r
  					Content-Length: 8\r
  					Content-Type: application/json\r
  					\r
  					{"at":2}POST /post2 HTTP/1.1\r
  					Host: localhost\r
  					Content-Length: 8\r
  					Content-Type: application/json\r
  					\r
  					{"at":2}""");
  			}
  
  			for (int i = 0; i < 100; i++) {
  				readAndAssert(socket.getInputStream(), """
  					HTTP/1.1 200 OK\r
  					Connection: keep-alive\r
  					Content-Length: 4\r
  					\r
  					/abc""");
  				readAndAssert(socket.getInputStream(), """
  					HTTP/1.1 200 OK\r
  					Connection: keep-alive\r
  					Content-Length: 7\r
  					\r
  					/123456""");
  				readAndAssert(socket.getInputStream(), """
  					HTTP/1.1 200 OK\r
  					Connection: keep-alive\r
  					Content-Length: 6\r
  					\r
  					/post1""");
  				readAndAssert(socket.getInputStream(), """
  					HTTP/1.1 200 OK\r
  					Connection: keep-alive\r
  					Content-Length: 6\r
  					\r
  					/post2""");
  			}
  		}
  
  		server.closeFuture().get();
  		thread.join();
  		resetPort();
  	}
  
  	@Test
  	public void testBigHttpMessage() throws Exception {
  		byte[] body = encodeAscii("Test big HTTP message body");
  		HttpRequest request = HttpRequest.post("http://127.0.0.1:" + port)
  			.withBody(body)
  			.build();
  
  		ByteBuf buf = ByteBufPool.allocate(request.estimateSize() + body.length);
  		request.writeTo(buf);
  		buf.put(body);
  
  		HttpServer server = HttpServer.builder(eventloop,
  				req -> HttpResponse.ok200()
  					.withBody(encodeAscii(req.getUrl().getPathAndQuery()))
  					.toPromise())
  			.withListenPort(port)
  			.build();
  		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();
  //		assertEquals(1, server.getStats().getHttpErrors().getTotal());
  //		System.out.println(server.getStats().getHttpErrors().getLastException());
  //		assertEquals(AbstractHttpConnection.TOO_BIG_HTTP_MESSAGE,
  //				server.getStats().getHttpErrors().getLastException());
  	}
  
  	@Test
  	public void testExpectContinue() throws Exception {
  		HttpServer server = HttpServer.builder(eventloop,
  				request -> request.loadBody()
  					.then(body -> HttpResponse.ok200().withBody(body.slice()).toPromise()))
  			.withListenPort(port)
  			.build();
  
  		server.listen();
  		Thread thread = new Thread(eventloop);
  		thread.start();
  
  		try (Socket socket = new Socket()) {
  			socket.setTcpNoDelay(true);
  			socket.connect(new InetSocketAddress("localhost", port));
  
  			writeByRandomParts(socket, """
  				POST /abc HTTP/1.0\r
  				Host: localhost\r
  				Content-Length: 5\r
  				Expect: 100-continue\r
  				\r
  				""");
  			readAndAssert(socket.getInputStream(), "HTTP/1.1 100 Continue\r\n\r\n");
  
  			writeByRandomParts(socket, "abcde");
  			readAndAssert(socket.getInputStream(), """
  				HTTP/1.1 200 OK\r
  				Connection: close\r
  				Content-Length: 5\r
  				\r
  				abcde""");
  
  			assertEmpty(socket.getInputStream());
  			assertTrue(socket.isClosed());
  		}
  
  		server.closeFuture().get();
  		thread.join();
  	}
  
  	@Test
  	public void testBodyRecycledOnce() throws IOException, InterruptedException {
  		HttpServer.builder(eventloop,
  				request -> {
  					// imitate network problems
  					shutdownAllChannels();
  					return HttpResponse.ok200().toPromise();
  				})
  			.withListenPort(port)
  			.withAcceptOnce()
  			.build()
  			.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
  					Host: localhost\r
  					Connection: close\r
  					Content-Length: 10\r
  					\r
  					test"""));
  				socket.getOutputStream().write(buf.array(), buf.head(), buf.readRemaining());
  			} catch (IOException e) {
  				throw new RuntimeException(e);
  			}
  		});
  
  		thread.start();
  		eventloop.run();
  		thread.join();
  	}
  
  	@Test
  	public void testPostParameters() throws IOException, ExecutionException, InterruptedException {
  		HttpServer server = HttpServer.builder(eventloop, request ->
  				request.loadBody()
  					.then(() -> {
  						Map<String, String> postParameters = request.getPostParameters();
  						StringBuilder sb = new StringBuilder();
  						for (Map.Entry<String, String> entry : postParameters.entrySet()) {
  							sb.append(entry.getKey())
  								.append("=")
  								.append(entry.getValue())
  								.append(";");
  						}
  
  						return HttpResponse.ok200()
  							.withBody(encodeAscii(sb.toString()))
  							.toPromise();
  					}))
  			.withListenPort(port)
  			.build();
  		server.listen();
  		Thread thread = new Thread(eventloop);
  		thread.start();
  
  		try (Socket socket = new Socket()) {
  			socket.connect(new InetSocketAddress("localhost", port));
  			writeByRandomParts(socket, """
  				POST / HTTP/1.1\r
  				Host: localhost\r
  				Connection: close\r
  				Content-Type: application/x-www-form-urlencoded\r
  				Content-Length: 27\r
  				\r
  				field1=value1&field2=value2""");
  
  			readAndAssert(socket.getInputStream(), """
  				HTTP/1.1 200 OK\r
  				Connection: close\r
  				Content-Length: 28\r
  				\r
  				field1=value1;field2=value2;""");
  			assertEmpty(socket.getInputStream());
  		}
  
  		server.closeFuture().get();
  		thread.join();
  	}
  
  	@Test
  	public void testIncompleteRequest() throws IOException, ExecutionException, InterruptedException {
  		JmxInspector inspector = new JmxInspector();
  		HttpServer server = HttpServer.builder(eventloop, request ->
  				request.loadBody()
  					.map(($, e) -> {
  						assertTrue(e instanceof MalformedHttpException);
  						assertEquals("Incomplete HTTP message", e.getMessage());
  						assertEquals(1, inspector.getMalformedHttpExceptions().getTotal());
  						assertEquals("Incomplete HTTP message", inspector.getMalformedHttpExceptions().getLastMessage());
  						assertEquals(0, inspector.getHttpErrors().getTotal());
  
  						assertFalse(request.isRecycled());
  						assertTrue(request.getConnection().isClosed());
  
  						assertEquals("localhost", request.getHeader(HttpHeaders.HOST));
  						return HttpResponse.ofCode(400).build();
  					}))
  			.withInspector(inspector)
  			.withListenPort(port)
  			.build();
  		server.listen();
  		Thread thread = new Thread(eventloop);
  		thread.start();
  
  		try (Socket socket = new Socket()) {
  			socket.connect(new InetSocketAddress("localhost", port));
  			writeByRandomParts(socket, """
  				POST / HTTP/1.1\r
  				Host: localhost\r
  				Connection: close\r
  				Content-Type: application/x-www-form-urlencoded\r
  				Content-Length: 100\r
  				\r
  				field1=value1&field2=value2""");
  			socket.shutdownOutput();
  
  			readAndAssert(socket.getInputStream(),
  				"""
  					HTTP/1.1 400 Bad Request\r
  					Connection: close\r
  					Content-Length: 0\r
  					\r
  					""");
  
  			assertEmpty(socket.getInputStream());
  		}
  
  		server.closeFuture().get();
  		thread.join();
  	}
  
  	@Test
  	public void testMalformedStartLineNoLF() throws IOException, ExecutionException, InterruptedException {
  		JmxInspector inspector = new JmxInspector();
  		HttpServer server = HttpServer.builder(eventloop, $ -> {
  				throw new IllegalArgumentException("Should not be called");
  			})
  			.withListenPort(port)
  			.withInspector(inspector)
  			.build();
  		server.listen();
  		Thread thread = new Thread(eventloop);
  		thread.start();
  
  		// malformed header
  		String malformedHeaderRequest = "AAABBBCCC";
  		doTestMalformedRequest(malformedHeaderRequest);
  
  		ExceptionStats malformedHttpExceptions = inspector.getMalformedHttpExceptions();
  		assertEquals(1, malformedHttpExceptions.getTotal());
  		assertEquals("Invalid start line prefix", malformedHttpExceptions.getLastMessage());
  		String context = (String) malformedHttpExceptions.getContext();
  		assertNotNull(context);
  		assertFalse(context.isEmpty());
  		assertTrue(malformedHeaderRequest.contains(context.substring("/127.0.0.1: ".length())));
  
  		server.closeFuture().get();
  		thread.join();
  	}
  
  	@Test
  	public void testMalformedUri() throws IOException, ExecutionException, InterruptedException {
  		JmxInspector inspector = new JmxInspector();
  		HttpServer server = HttpServer.builder(eventloop, $ -> {
  				throw new IllegalArgumentException("Should not be called");
  			})
  			.withListenPort(port)
  			.withInspector(inspector)
  			.build();
  		server.listen();
  		Thread thread = new Thread(eventloop);
  		thread.start();
  
  		String malformedUriRequest = """
  			GET /malformed uri HTTP/1.1\r
  			Host: localhost\r
  			Connection: keep-alive\r
  			\r
  			""";
  		doTestMalformedRequest(malformedUriRequest);
  
  		ExceptionStats malformedHttpExceptions = inspector.getMalformedHttpExceptions();
  		assertEquals(1, malformedHttpExceptions.getTotal());
  		assertEquals("Unsupported HTTP version", malformedHttpExceptions.getLastMessage());
  		String context = (String) malformedHttpExceptions.getContext();
  		assertNotNull(context);
  		assertFalse(context.isEmpty());
  		assertTrue(malformedUriRequest.startsWith(context.substring("/127.0.0.1: ".length())));
  
  		server.closeFuture().get();
  		thread.join();
  	}
  
  	@Test
  	public void testMalformedHeaders() throws IOException, ExecutionException, InterruptedException {
  		JmxInspector inspector = new JmxInspector();
  		HttpServer server = HttpServer.builder(eventloop, $ -> {
  				throw new IllegalArgumentException("Should not be called");
  			})
  			.withListenPort(port)
  			.withInspector(inspector)
  			.build();
  		server.listen();
  		Thread thread = new Thread(eventloop);
  		thread.start();
  
  		// malformed header
  		String malformedHeaderRequest = """
  			GET / HTTP/1.1\r
  			Host: localhost\r
  			Connection: keep-alive\r
  			Content-Length: error\r
  			\r
  			""";
  		doTestMalformedRequest(malformedHeaderRequest);
  
  		ExceptionStats malformedHttpExceptions = inspector.getMalformedHttpExceptions();
  		assertEquals(1, malformedHttpExceptions.getTotal());
  		assertEquals("Not a decimal value: error", malformedHttpExceptions.getLastMessage());
  		String context = (String) malformedHttpExceptions.getContext();
  		assertNotNull(context);
  		assertFalse(context.isEmpty());
  		assertTrue(malformedHeaderRequest.contains(context.substring("/127.0.0.1: ".length())));
  
  		server.closeFuture().get();
  		thread.join();
  	}
  
  	@Test
  	public void testMalformedFirstRequestPipelined() throws IOException, ExecutionException, InterruptedException {
  		JmxInspector inspector = new JmxInspector();
  		HttpServer server = HttpServer.builder(eventloop, $ -> {
  				throw new IllegalArgumentException("Should not be called");
  			})
  			.withListenPort(port)
  			.withInspector(inspector)
  			.build();
  		server.listen();
  		Thread thread = new Thread(eventloop);
  		thread.start();
  
  		// pipeline malformed first request
  		String malformedPipelinedRequest = """
  			GET /malformed uri HTTP/1.1\r
  			Host: localhost\r
  			Connection: keep-alive\r
  			\r
  			GET / HTTP/1.1\r
  			Host: localhost\r
  			Connection: keep-alive\r
  			\r
  			""";
  		doTestMalformedRequest(malformedPipelinedRequest);
  
  		ExceptionStats malformedHttpExceptions = inspector.getMalformedHttpExceptions();
  		assertEquals(1, malformedHttpExceptions.getTotal());
  		assertEquals("Unsupported HTTP version", malformedHttpExceptions.getLastMessage());
  		String context = (String) malformedHttpExceptions.getContext();
  		assertNotNull(context);
  		assertFalse(context.isEmpty());
  		assertTrue(malformedPipelinedRequest.contains(context.substring("/127.0.0.1: ".length())));
  
  		server.closeFuture().get();
  		thread.join();
  	}
  
  	@Test
  	public void testMalformedSecondRequestPipelined() throws IOException, ExecutionException, InterruptedException {
  		JmxInspector inspector = new JmxInspector();
  		HttpServer server = HttpServer.builder(eventloop, $ -> HttpResponse.ok200()
  				.withPlainText("Hello, world!")
  				.toPromise())
  			.withListenPort(port)
  			.withInspector(inspector)
  			.build();
  		server.listen();
  		Thread thread = new Thread(eventloop);
  		thread.start();
  
  		String normalPipelinedrequest = """
  			GET / HTTP/1.1\r
  			Host: localhost\r
  			Connection: keep-alive\r
  			\r
  			""";
  		String malformedPipelinedRequest = """
  			GET /malformed uri HTTP/1.1\r
  			Host: localhost\r
  			Connection: keep-alive\r
  			\r
  			""";
  
  		try (Socket socket = new Socket()) {
  			socket.connect(new InetSocketAddress("localhost", port));
  			writeByRandomParts(socket, normalPipelinedrequest + malformedPipelinedRequest);
  			socket.shutdownOutput();
  
  			readAndAssert(socket.getInputStream(),
  				"""
  					HTTP/1.1 200 OK\r
  					Connection: keep-alive\r
  					Content-Type: text/plain; charset=utf-8\r
  					Content-Length: 13\r
  					\r
  					Hello, world!HTTP/1.1 400 Bad Request\r
  					Connection: close\r
  					Content-Length: 0\r
  					\r
  					""");
  
  			assertEmpty(socket.getInputStream());
  		}
  
  		ExceptionStats malformedHttpExceptions = inspector.getMalformedHttpExceptions();
  		assertEquals(1, malformedHttpExceptions.getTotal());
  		assertEquals("Unsupported HTTP version", malformedHttpExceptions.getLastMessage());
  		String context = (String) malformedHttpExceptions.getContext();
  		assertNotNull(context);
  		assertFalse(context.isEmpty());
  		assertTrue(malformedPipelinedRequest.startsWith(context.substring("/127.0.0.1: ".length())));
  
  		server.closeFuture().get();
  		thread.join();
  	}
  
  	@Test
  	public void testActiveRequestsCounter() throws IOException, ExecutionException, InterruptedException {
  		JmxInspector inspector = new JmxInspector();
  		HttpServer server = HttpServer.builder(eventloop, request -> {
  				assertEquals(1, inspector.getTotalRequests().getTotalCount());
  				assertEquals(1, inspector.getActiveConnections());
  				assertEquals(1, inspector.getActiveRequests());
  				return HttpResponse.ok200()
  					.withBodyStream(request.takeBodyStream())
  					.toPromise();
  			})
  			.withListenPort(port)
  			.withReadWriteTimeout(Duration.ofMillis(20))
  			.withInspector(inspector)
  			.build();
  		server.listen();
  
  		Thread thread = new Thread(eventloop);
  		thread.start();
  
  		try (Socket socket = new Socket()) {
  			socket.connect(new InetSocketAddress("localhost", port));
  			writeByRandomParts(socket, """
  				POST / HTTP/1.1\r
  				Host: localhost\r
  				Connection: keep-alive\r
  				Transfer-Encoding: chunked\r
  				\r
  				""");
  
  			skipResponse(socket);
  		}
  
  		assertEquals(1, inspector.getTotalRequests().getTotalCount());
  		assertEquals(0, inspector.getActiveConnections());
  		assertEquals(0, inspector.getActiveRequests());
  
  		server.closeFuture().get();
  		thread.join();
  	}
  
  	@Test
  	public void testActiveRequestsCounterServing() throws IOException, ExecutionException, InterruptedException {
  		JmxInspector inspector = new JmxInspector();
  		CountDownLatch latch = new CountDownLatch(1);
  		HttpServer server = HttpServer.builder(eventloop, request -> {
  				assertEquals(1, inspector.getTotalRequests().getTotalCount());
  				assertEquals(1, inspector.getActiveConnections());
  				assertEquals(1, inspector.getActiveRequests());
  				return request.loadBody()
  					.whenException(e -> latch.countDown())
  					.map($ -> HttpResponse.ok200().build());
  			})
  			.withListenPort(port)
  			.withInspector(inspector)
  			.build();
  		server.listen();
  
  		Thread thread = new Thread(eventloop);
  		thread.start();
  
  		try (Socket socket = new Socket()) {
  			socket.connect(new InetSocketAddress("localhost", port));
  			writeByRandomParts(socket, """
  				POST / HTTP/1.1\r
  				Host: localhost\r
  				Connection: keep-alive\r
  				Content-Length: 1000\r
  				\r
  				""");
  
  			writeByRandomParts(socket, "Truncated request");
  		}
  
  		latch.await();
  
  		assertEquals(1, inspector.getTotalRequests().getTotalCount());
  		assertEquals(0, inspector.getActiveConnections());
  		assertEquals(0, inspector.getActiveRequests());
  
  		server.closeFuture().get();
  		thread.join();
  	}
  
  	@Test
  	public void testEmptyGzipRequestChunked() throws IOException, InterruptedException {
  		JmxInspector inspector = new JmxInspector();
  		HttpServer server = HttpServer.builder(eventloop, request -> request.loadBody()
  				.map($ -> HttpResponse.ok200().build()))
  			.withAcceptOnce()
  			.withListenPort(port)
  			.withInspector(inspector)
  			.build();
  		server.listen();
  
  		Thread thread = new Thread(eventloop);
  		thread.start();
  
  		try (Socket socket = new Socket()) {
  			socket.connect(new InetSocketAddress("localhost", port));
  			writeByRandomParts(socket, """
  				POST / HTTP/1.1\r
  				Host: localhost\r
  				Connection: close\r
  				Transfer-Encoding: chunked\r
  				Content-Encoding: gzip\r
  				\r
  				0\r
  				\r
  				""");
  
  			readAndAssert(socket.getInputStream(),
  				"""
  					HTTP/1.1 200 OK\r
  					Connection: close\r
  					Content-Length: 0\r
  					\r
  					""");
  
  			assertEmpty(socket.getInputStream());
  		}
  
  		thread.join();
  	}
  
  	@Test
  	public void testMalformedGzipRequest() throws IOException, InterruptedException {
  		JmxInspector inspector = new JmxInspector();
  		HttpServer server = HttpServer.builder(eventloop, request -> request.loadBody()
  				.map($ -> HttpResponse.ok200().build()))
  			.withAcceptOnce()
  			.withListenPort(port)
  			.withInspector(inspector)
  			.build();
  		server.listen();
  
  		Thread thread = new Thread(eventloop);
  		thread.start();
  
  		try (Socket socket = new Socket()) {
  			socket.connect(new InetSocketAddress("localhost", port));
  			writeByRandomParts(socket, """
  				POST / HTTP/1.1\r
  				Host: localhost\r
  				Connection: close\r
  				Content-Length: 3\r
  				Content-Encoding: gzip\r
  				\r
  				abc
  				""");
  
  			skipResponse(socket);
  		}
  
  		Throwable lastException = inspector.getHttpErrors().getLastException();
  
  		assertThat(lastException, instanceOf(HttpException.class));
  		//noinspection DataFlowIssue
  		assertThat(lastException.getCause(), instanceOf(TruncatedDataException.class));
  
  		thread.join();
  	}
  
  	@Test
  	public void testMalformedGzipHeader() throws IOException, InterruptedException {
  		Ref<Exception> exceptionRef = new Ref<>();
  		HttpServer server = HttpServer.builder(eventloop, request -> request.loadBody()
  				.map($ -> HttpResponse.ok200().build())
  				.whenException(exceptionRef::set))
  			.withAcceptOnce()
  			.withListenPort(port)
  			.build();
  		server.listen();
  
  		Thread thread = new Thread(eventloop);
  		thread.start();
  
  		ByteBuf buf = wrapUtf8("""
  			POST / HTTP/1.1\r
  			Host: localhost\r
  			Connection: close\r
  			Content-Length: 10\r
  			Content-Encoding: gzip\r
  			\r
  			""");
  
  		byte[] bytes = new byte[10];
  		Arrays.fill(bytes, (byte) 0x1f);
  
  		buf = ByteBufPool.append(buf, bytes);
  		buf = ByteBufPool.append(buf, new byte[]{CR, LF});
  
  		try (Socket socket = new Socket()) {
  			socket.connect(new InetSocketAddress("localhost", port));
  			writeByRandomParts(socket, buf);
  
  			skipResponse(socket);
  		}
  
  		thread.join();
  
  		Exception exception = exceptionRef.get();
  
  		assertNotNull(exception);
  		assertThat(exception, instanceOf(HttpException.class));
  		assertThat(exception.getCause(), instanceOf(UnknownFormatException.class));
  	}
  
  	@Test
  	public void testGetFullUri() throws IOException {
  		Ref<String> fullUriRef = new Ref<>();
  		HttpServer server = HttpServer.builder(eventloop, request -> {
  				String fullUri = HttpUtils.getFullUri(request);
  				fullUriRef.set(fullUri);
  				return HttpResponse.ok200().toPromise();
  			})
  			.withListenPort(port)
  			.withAcceptOnce()
  			.build();
  		server.listen();
  
  		Thread thread = new Thread(eventloop);
  		thread.start();
  
  		try (Socket socket = new Socket()) {
  			socket.connect(new InetSocketAddress("localhost", port));
  			writeByRandomParts(socket, "GET /a/b/c?q=1&u=2#fragment HTTP/1.1\r\n" +
  									   "Host: localhost:" + port + "\r\n" +
  									   "Connection: close\r\n" +
  									   "Content-Length: 0\r\n" +
  									   "\r\n");
  
  			skipResponse(socket);
  		}
  
  		assertEquals("http://localhost:" + port + "/a/b/c?q=1&u=2#fragment", fullUriRef.get());
  	}
  
  	private static void skipResponse(Socket socket) throws IOException {
  		int read;
  		do {
  			read = socket.getInputStream().read();
  		} while (read != -1);
  	}
  
  	private void doTestMalformedRequest(String string) throws IOException {
  		try (Socket socket = new Socket()) {
  			socket.connect(new InetSocketAddress("localhost", port));
  			writeByRandomParts(socket, string);
  			socket.shutdownOutput();
  
  			readAndAssert(socket.getInputStream(), """
  				HTTP/1.1 400 Bad Request\r
  				Connection: close\r
  				Content-Length: 0\r
  				\r
  				""");
  		}
  	}
  
  	private void shutdownAllChannels() {
  		try {
  			Selector selector = eventloop.getSelector();
  			assert selector != null;
  			selector.keys().iterator().next().channel().close();
  		} catch (IOException e) {
  			throw new RuntimeException(e);
  		}
  	}
  
  	private void resetPort() {
  		port = getFreePort();
  	}
  }
  