@Test
void exceededMaxStreamsForMultipleEventLoops() {
    final ClientFactory clientFactory =
            ClientFactory.builder()
                         .connectionPoolListener(connectionPoolListenerWrapper)
                         .maxNumEventLoopsPerEndpoint(2)
                         .build();
    final WebClient client = WebClient.builder(server.uri(SessionProtocol.H2C))
                                      .factory(clientFactory)
                                      .build();
    final AtomicInteger opens = new AtomicInteger();
    connectionPoolListener = newConnectionPoolListener(opens::incrementAndGet, () -> {});

    final int numExpectedConnections = MAX_NUM_CONNECTIONS;
    final int numRequests = MAX_CONCURRENT_STREAMS * numExpectedConnections;

    runInsideEventLoop(clientFactory.eventLoopGroup(), () -> {
        for (int i = 0; i < numRequests; i++) {
            client.get(PATH).aggregate();
        }
    });

    await().untilAsserted(() -> assertThat(responses).hasSize(numRequests));
    assertThat(opens).hasValue(numExpectedConnections);
}
 

HttpClientPipelineConfigurator(HttpClientFactory clientFactory,
                               SessionProtocol sessionProtocol,
                               @Nullable SslContext sslCtx) {
    this.clientFactory = clientFactory;

    if (sessionProtocol == HTTP || sessionProtocol == HTTPS) {
        httpPreference = HttpPreference.HTTP2_PREFERRED;
    } else if (sessionProtocol == H1 || sessionProtocol == H1C) {
        httpPreference = HttpPreference.HTTP1_REQUIRED;
    } else if (sessionProtocol == H2 || sessionProtocol == H2C) {
        httpPreference = HttpPreference.HTTP2_REQUIRED;
    } else {
        // Should never reach here.
        throw new Error();
    }

    if (sessionProtocol.isTls()) {
        this.sslCtx = sslCtx;
    } else {
        this.sslCtx = null;
    }
}
 

@Test
void testActiveLocalPort() throws Exception {
    final Server server = Server.builder()
                                .http(0)
                                .https(0)
                                .tlsSelfSigned()
                                .service("/", (ctx, res) -> HttpResponse.of(""))
                                .build();

    // not started yet
    assertThatThrownBy(server::activeLocalPort)
            .isInstanceOf(IllegalStateException.class)
            .hasMessageContaining("no active local ports");

    server.start().get();

    assertThat(server.activeLocalPort()).isPositive();
    assertThat(server.activeLocalPort(SessionProtocol.HTTP)).isPositive();
    assertThat(server.activeLocalPort(SessionProtocol.HTTPS)).isPositive();
    assertThatThrownBy(() -> server.activeLocalPort(SessionProtocol.PROXY))
            .isInstanceOf(IllegalStateException.class)
            .hasMessageContaining("no active local ports for " + SessionProtocol.PROXY);
}
 

private LocalArmeriaPortElement(Member member, AnnotatedElement ae, @Nullable PropertyDescriptor pd) {
    super(member, pd);
    final LocalArmeriaPort localArmeriaPort = ae.getAnnotation(LocalArmeriaPort.class);
    final SessionProtocol protocol = localArmeriaPort.value();
    Server server = getServer();
    if (server == null) {
        server = beanFactory.getBean(Server.class);
        serServer(server);
    }

    Integer port = portCache.get(protocol);
    if (port == null) {
        port = server.activeLocalPort(protocol);
        portCache.put(protocol, port);
    }
    this.port = port;
}
 

/**
 * Returns the {@link URI} for the {@link Server} of the specified {@link SessionProtocol} and
 * {@link SerializationFormat}.
 *
 * @throws IllegalStateException if the {@link Server} is not started or
 *                               it did not open the port for the specified {@link SessionProtocol}.
 */
public URI uri(SessionProtocol protocol, SerializationFormat format) {
    requireNonNull(protocol, "protocol");
    requireNonNull(format, "format");

    ensureStarted();

    final int port;
    if (!protocol.isTls() && hasHttp()) {
        port = httpPort();
    } else if (protocol.isTls() && hasHttps()) {
        port = httpsPort();
    } else {
        throw new IllegalStateException("can't find the specified port");
    }

    final String uriStr = protocol.uriText() + "://127.0.0.1:" + port;
    if (format == SerializationFormat.NONE) {
        return URI.create(uriStr);
    } else {
        return URI.create(format.uriText() + '+' + uriStr);
    }
}
 

@Nullable
private ServerPort activePort0(@Nullable SessionProtocol protocol) {
    ServerPort candidate = null;
    synchronized (activePorts) {
        for (ServerPort serverPort : activePorts.values()) {
            if (protocol == null || serverPort.hasProtocol(protocol)) {
                if (!isLocalPort(serverPort)) {
                    return serverPort;
                } else if (candidate == null) {
                    candidate = serverPort;
                }
            }
        }
    }
    return candidate;
}
 

@Test
void testProxyWithH2C() throws Exception {
    final int numRequests = 5;
    final ClientFactory clientFactory = ClientFactory.builder().proxyConfig(
            ProxyConfig.socks4(socksProxyServer.address())).build();
    final WebClient webClient = WebClient.builder(SessionProtocol.H2C, backendServer.httpEndpoint())
                                         .factory(clientFactory)
                                         .decorator(LoggingClient.newDecorator())
                                         .build();

    final List<CompletableFuture<AggregatedHttpResponse>> responseFutures = new ArrayList<>();
    for (int i = 0; i < numRequests; i++) {
        responseFutures.add(webClient.get(PROXY_PATH).aggregate());
    }
    await().until(() -> responseFutures.stream().allMatch(CompletableFuture::isDone));
    assertThat(responseFutures.stream().map(CompletableFuture::join))
            .allMatch(response -> response.contentUtf8().equals(SUCCESS_RESPONSE));
    assertThat(numSuccessfulProxyRequests).isGreaterThanOrEqualTo(1);
    clientFactory.close();
}
 

@BeforeEach
void setUp() {
    opened = new AtomicInteger();
    closed = new AtomicInteger();
    connectionPoolListener = new ConnectionPoolListener() {
        @Override
        public void connectionOpen(SessionProtocol protocol, InetSocketAddress remoteAddr,
                                   InetSocketAddress localAddr, AttributeMap attrs) throws Exception {
            opened.incrementAndGet();
        }

        @Override
        public void connectionClosed(SessionProtocol protocol, InetSocketAddress remoteAddr,
                                     InetSocketAddress localAddr, AttributeMap attrs) throws Exception {
            closed.incrementAndGet();
        }
    };
}
 

@Test
void respectsHttpClientUri_endpointGroup() throws Exception {
    final EndpointGroup group = EndpointGroup.of(Endpoint.of("127.0.0.1", server.httpPort()),
                                                 Endpoint.of("127.0.0.1", server.httpPort()));

    final Service service = ArmeriaRetrofit.builder("http", group)
                                           .addConverterFactory(converterFactory)
                                           .build()
                                           .create(Service.class);

    try (ClientRequestContextCaptor ctxCaptor = Clients.newContextCaptor()) {
        final Response<Pojo> response = service.postForm("Cony", 26).get();

        final RequestLog log = ctxCaptor.get().log().whenComplete().join();
        assertThat(log.sessionProtocol()).isSameAs(SessionProtocol.H2C);
        assertThat(log.requestHeaders().authority()).isEqualTo("127.0.0.1:" + server.httpPort());

        // TODO(ide) Use the actual `host:port`. See https://github.com/line/armeria/issues/379
        final HttpUrl url = response.raw().request().url();
        assertThat(url.scheme()).isEqualTo("http");
        assertThat(url.host()).startsWith("armeria-group-");
        assertThat(url.pathSegments()).containsExactly("postForm");
    }
}
 

@ParameterizedTest
@ArgumentsSource(ParametersProvider.class)
void contextCaptorSync(
        ClientOptions clientOptions, SerializationFormat format, SessionProtocol protocol)
        throws Exception {
    final HelloService.Iface client = Clients.builder(uri(Handlers.HELLO, format, protocol))
                                             .options(clientOptions)
                                             .build(Handlers.HELLO.iface());
    try (ClientRequestContextCaptor ctxCaptor = Clients.newContextCaptor()) {
        client.hello("kukuman");
        final ClientRequestContext ctx = ctxCaptor.get();
        final RpcRequest rpcReq = ctx.rpcRequest();
        assertThat(rpcReq).isNotNull();
        assertThat(rpcReq.method()).isEqualTo("hello");
        assertThat(rpcReq.params()).containsExactly("kukuman");
    }
}
 

@ParameterizedTest
@ArgumentsSource(ParametersProvider.class)
void testTimeServiceAsync(
        ClientOptions clientOptions, SerializationFormat format, SessionProtocol protocol)
        throws Exception {
    final TimeService.AsyncIface client =
            Clients.builder(uri(Handlers.TIME, format, protocol))
                   .options(clientOptions)
                   .build(Handlers.TIME.asyncIface());

    final BlockingQueue<Object> resQueue = new LinkedBlockingQueue<>();
    client.getServerTime(new RequestQueuingCallback(resQueue));

    final Object result = resQueue.take();
    assertThat(result).isInstanceOf(Long.class);
    assertThat((Long) result).isLessThanOrEqualTo(System.currentTimeMillis());
}
 

@Override
public void connectionClosed(SessionProtocol protocol, InetSocketAddress remoteAddr,
                             InetSocketAddress localAddr, AttributeMap attrs) throws Exception {
    final ConnectionPoolListener connectionPoolListener =
            HttpClientMaxConcurrentStreamTest.this.connectionPoolListener;
    if (connectionPoolListener != null) {
        connectionPoolListener.connectionClosed(protocol, remoteAddr, localAddr, attrs);
    }
}
 

@Override
public void serialize(ServerPort value,
                      JsonGenerator gen, SerializerProvider serializers) throws IOException {

    final InetSocketAddress localAddr = value.localAddress();
    final int port = localAddr.getPort();
    final String host;

    if (localAddr.getAddress().isAnyLocalAddress()) {
        host = "*";
    } else {
        final String hs = localAddr.getHostString();
        if (NetUtil.isValidIpV6Address(hs)) {
            // Try to get the platform-independent consistent IPv6 address string.
            host = NetUtil.toAddressString(localAddr.getAddress());
        } else {
            host = hs;
        }
    }

    gen.writeStartObject();
    gen.writeObjectFieldStart("localAddress");
    gen.writeStringField("host", host);
    gen.writeNumberField("port", port);
    gen.writeEndObject();
    gen.writeArrayFieldStart("protocols");
    for (final SessionProtocol protocol : value.protocols()) {
        gen.writeString(protocol.uriText());
    }
    gen.writeEndArray();
    gen.writeEndObject();
}
 

ArmeriaRetrofitBuilder(WebClient webClient) {
    this.webClient = webClient;
    final URI uri = webClient.uri();
    final SessionProtocol protocol = webClient.scheme().sessionProtocol();

    // Build a baseUrl that will pass Retrofit's validation.
    final HttpUrl baseUrl = HttpUrl.get((protocol.isTls() ? "https" : "http") +
                                        uri.toString().substring(protocol.uriText().length()));

    retrofitBuilder = new Retrofit.Builder().baseUrl(baseUrl);
    baseWebClientHost = baseUrl.host();
    baseWebClientPort = baseUrl.port();
}
 

@Test
void sendHelloViaHttp2() throws Exception {
    final AtomicReference<SessionProtocol> sessionProtocol = new AtomicReference<>();
    final HelloService.Iface client = newSchemeCapturingClient(http2uri(HTTP), sessionProtocol);

    assertThat(client.hello("new world")).isEqualTo("Hello, new world!");
    assertThat(sessionProtocol.get()).isEqualTo(H2C);
}
 

private int activeLocalPort0(@Nullable SessionProtocol protocol) {
    synchronized (activePorts) {
        return activePorts.values().stream()
                          .filter(activePort -> (protocol == null || activePort.hasProtocol(protocol)) &&
                                                isLocalPort(activePort))
                          .findFirst()
                          .orElseThrow(() -> new IllegalStateException(
                                  (protocol == null ? "no active local ports: "
                                                    : ("no active local ports for " + protocol + ": ")) +
                                  activePorts.values()))
                          .localAddress()
                          .getPort();
    }
}
 

private HttpResponse execute(EndpointGroup endpointGroup, HttpRequest req, SessionProtocol protocol) {
    final PathAndQuery pathAndQuery = PathAndQuery.parse(req.path());
    if (pathAndQuery == null) {
        final IllegalArgumentException cause = new IllegalArgumentException("invalid path: " + req.path());
        return abortRequestAndReturnFailureResponse(req, cause);
    }
    return execute(protocol, endpointGroup, req.method(),
                   pathAndQuery.path(), pathAndQuery.query(), null, req);
}
 

@ParameterizedTest
@EnumSource(value = SessionProtocol.class, mode = Mode.EXCLUDE, names = "PROXY")
void default_withScheme(SessionProtocol protocol) {
    final HttpRequest request = HttpRequest.of(HttpMethod.GET, server2.uri(protocol) + "/simple-client");
    try (ClientRequestContextCaptor captor = Clients.newContextCaptor()) {
        final WebClient client = WebClient.builder().factory(ClientFactory.insecure()).build();
        final HttpResponse response = client.execute(request);
        final ClientRequestContext ctx = captor.get();
        assertThat(ctx.sessionProtocol()).isEqualTo(protocol);
        assertThat(response.aggregate().join().status()).isEqualTo(OK);
    }
}
 

@Test
void configureServer() throws Exception {
    final File yml = new File(resourceFilePath("armeria-settings.yaml"));
    final ArmeriaSettings armeriaSettings = configFactory.build(yml);
    armeriaSettings.setSsl(null);
    final ServerBuilder serverBuilder = Server.builder()
            .service("/foo", (ctx, req) -> HttpResponse.of(200));
    serverBuilder.tlsSelfSigned();
    ArmeriaConfigurationUtil.configureServer(serverBuilder, armeriaSettings);
    final Server server = serverBuilder.build();
    assertThat(server.defaultHostname()).isEqualTo("host.name.com");
    assertThat(server.config().maxNumConnections()).isEqualTo(5000);
    assertThat(server.config().isDateHeaderEnabled()).isFalse();
    assertThat(server.config().isServerHeaderEnabled()).isTrue();
    assertThat(server.config().defaultVirtualHost().maxRequestLength()).isEqualTo(10485761);

    assertThat(server.config().ports()).hasSize(3);
    assertThat(server.config().ports()).containsExactly(
            new ServerPort(8080, SessionProtocol.HTTP),
            new ServerPort(new InetSocketAddress("127.0.0.1", 8081), SessionProtocol.HTTPS),
            new ServerPort(8443, SessionProtocol.HTTPS, SessionProtocol.PROXY)
    );
    assertThat(server.config().http1MaxChunkSize()).isEqualTo(4000);
    assertThat(server.config().http1MaxInitialLineLength()).isEqualTo(4096);
    assertThat(server.config().http1MaxInitialLineLength()).isEqualTo(4096);
    assertThat(server.config().http2InitialConnectionWindowSize()).isEqualTo(1024 * 1024 * 2);
    assertThat(server.config().http2InitialStreamWindowSize()).isEqualTo(1024 * 1024 * 2);
    assertThat(server.config().http2MaxFrameSize()).isEqualTo(16385);
    assertThat(server.config().http2MaxHeaderListSize()).isEqualTo(8193);
    assertThat(server.config().proxyProtocolMaxTlvSize()).isEqualTo(65320);
}
 

@Override
public void connectionClosed(SessionProtocol protocol,
                             InetSocketAddress remoteAddr,
                             InetSocketAddress localAddr,
                             AttributeMap attrs) throws Exception {
    final int activeChannels = this.activeChannels.decrementAndGet();
    if (logger.isInfoEnabled() && attrs.hasAttr(OPEN_NANOS)) {
        final long closeNanos = ticker.read();
        final long elapsedNanos = closeNanos - attrs.attr(OPEN_NANOS).get();
        logger.info("[L:{} ! R:{}][{}] CLOSED (lasted for: {}, active channels: {})",
                    localAddr, remoteAddr, protocol.uriText(),
                    TextFormatter.elapsed(elapsedNanos), activeChannels);
    }
}
 

@Nullable
private static Port obtainManagementServerPort(Integer port) {
    int actualPort = requireNonNull(port, "port");
    if (actualPort < 0) {
        return null;
    }
    if (actualPort == 0) {
        actualPort = SocketUtils.findAvailableTcpPort();
    }
    return new Port().setPort(actualPort).setProtocol(SessionProtocol.HTTP);
}
 

@ParameterizedTest
@ArgumentsSource(ClientAndProtocolProvider.class)
void testExpect100ContinueDoesNotBreakHttp1Decoder(WebClient client, SessionProtocol protocol)
        throws Exception {
    assumeThat(protocol).isSameAs(H1C);

    final int port = server.httpPort();
    try (Socket s = new Socket(NetUtil.LOCALHOST, port)) {
        s.setSoTimeout(10000);
        final InputStream in = s.getInputStream();
        final OutputStream out = s.getOutputStream();
        // Send 4 pipelined requests with 'Expect: 100-continue' header.
        out.write((Strings.repeat("POST /head-headers-only HTTP/1.1\r\n" +
                                  "Expect: 100-continue\r\n" +
                                  "Content-Length: 0\r\n\r\n", 3) +
                   "POST /head-headers-only HTTP/1.1\r\n" +
                   "Expect: 100-continue\r\n" +
                   "Content-Length: 0\r\n" +
                   "Connection: close\r\n\r\n").getBytes(StandardCharsets.US_ASCII));

        // '100 Continue' responses must appear once for each '200 OK' response.
        assertThat(new String(ByteStreams.toByteArray(in)))
                .isEqualTo(Strings.repeat("HTTP/1.1 100 Continue\r\n\r\n" +
                                          "HTTP/1.1 200 OK\r\n" +
                                          "content-type: text/plain; charset=utf-8\r\n" +
                                          "content-length: 6\r\n\r\n200 OK", 4));
    }
}
 

private static void acquireTenEntries(DefaultEventLoopScheduler s,
                                      SessionProtocol sessionProtocol,
                                      EndpointGroup endpointGroup,
                                      Endpoint endpoint) {
    for (int i = 0; i < 10; i++) {
        s.acquire(sessionProtocol, endpointGroup, endpoint);
    }
}
 

private static HelloService.Iface newSchemeCapturingClient(
        String uri, AtomicReference<SessionProtocol> sessionProtocol) {

    return Clients.builder(uri)
                  .rpcDecorator((delegate, ctx, req) -> {
                      ctx.log().whenAvailable(RequestLogProperty.SESSION)
                         .thenAccept(log -> sessionProtocol.set(log.sessionProtocol()));
                      return delegate.execute(ctx, req);
                  })
                  .build(HelloService.Iface.class);
}
 

@Override
public void connectionClosed(SessionProtocol protocol,
                             InetSocketAddress remoteAddr,
                             InetSocketAddress localAddr,
                             AttributeMap attrs) throws Exception {
    delegate().connectionClosed(protocol, remoteAddr, localAddr, attrs);
}
 

@Override
public void connectionOpen(SessionProtocol protocol,
                           InetSocketAddress remoteAddr,
                           InetSocketAddress localAddr,
                           AttributeMap attrs) throws Exception {
    final int activeChannels = this.activeChannels.incrementAndGet();
    if (logger.isInfoEnabled()) {
        attrs.attr(OPEN_NANOS).set(ticker.read());
        logger.info("[L:{} - R:{}][{}] OPEN (active channels: {})",
                    localAddr, remoteAddr, protocol.uriText(), activeChannels);
    }
}
 

Http2ClientConnectionHandler(Http2ConnectionDecoder decoder, Http2ConnectionEncoder encoder,
                             Http2Settings initialSettings, Channel channel,
                             HttpClientFactory clientFactory, SessionProtocol protocol) {

    super(decoder, encoder, initialSettings);
    this.clientFactory = clientFactory;

    if (clientFactory.idleTimeoutMillis() > 0 || clientFactory.pingIntervalMillis() > 0) {
        final Timer keepAliveTimer =
                MoreMeters.newTimer(clientFactory.meterRegistry(), "armeria.client.connections.lifespan",
                                    ImmutableList.of(Tag.of("protocol", protocol.uriText())));
        keepAliveHandler = new Http2ClientKeepAliveHandler(
                channel, encoder.frameWriter(), keepAliveTimer,
                clientFactory.idleTimeoutMillis(), clientFactory.pingIntervalMillis());
    } else {
        keepAliveHandler = null;
    }

    responseDecoder = new Http2ResponseDecoder(channel, encoder(), clientFactory, keepAliveHandler);
    connection().addListener(responseDecoder);
    decoder().frameListener(responseDecoder);

    // Setup post build options
    final long timeout = clientFactory.idleTimeoutMillis();
    if (timeout > 0) {
        gracefulShutdownTimeoutMillis(timeout);
    } else {
        // Timeout disabled
        gracefulShutdownTimeoutMillis(-1);
    }
}
 

/**
 * Creates a new instance.
 */
HealthCheckedEndpointGroup(
        EndpointGroup delegate, SessionProtocol protocol, int port,
        Backoff retryBackoff, ClientOptions clientOptions,
        Function<? super HealthCheckerContext, ? extends AsyncCloseable> checkerFactory,
        HealthCheckStrategy healthCheckStrategy) {

    super(requireNonNull(delegate, "delegate").selectionStrategy());

    this.delegate = delegate;
    this.protocol = requireNonNull(protocol, "protocol");
    this.port = port;
    this.retryBackoff = requireNonNull(retryBackoff, "retryBackoff");
    this.clientOptions = requireNonNull(clientOptions, "clientOptions");
    this.checkerFactory = requireNonNull(checkerFactory, "checkerFactory");
    this.healthCheckStrategy = requireNonNull(healthCheckStrategy, "healthCheckStrategy");

    clientOptions.factory().whenClosed().thenRun(this::closeAsync);
    delegate.addListener(this::updateCandidates);
    updateCandidates(delegate.whenReady().join());

    // Wait until the initial health of all endpoints are determined.
    final List<DefaultHealthCheckerContext> snapshot;
    synchronized (contexts) {
        snapshot = ImmutableList.copyOf(contexts.values());
    }
    snapshot.forEach(ctx -> ctx.initialCheckFuture.join());

    // If all endpoints are unhealthy, we will not have called setEndpoints even once, meaning listeners
    // aren't notified that we've finished an initial health check. We make sure to refresh endpoints once
    // on initialization to ensure this happens, even if the endpoints are currently empty.
    refreshEndpoints();
}
 

@CsvSource({ "H1C", "H2C" })
@ParameterizedTest
void shouldNotDisconnect(SessionProtocol protocol) throws InterruptedException {
    final WebClient client = newWebClient(serverKeepAlive.uri(protocol));

    for (int i = 0; i < 10; i++) {
        assertThat(client.get("/").aggregate().join().status()).isEqualTo(OK);
        assertThat(opened).hasValue(1);
        assertThat(closed).hasValue(0);
        Thread.sleep(100);
    }
}
 

@Test
void testCustomAuthority() throws Exception {
    final WebClient client = WebClient.builder(SessionProtocol.HTTPS,
                                               Endpoint.of("a.com", httpsPort)
                                                       .withIpAddr("127.0.0.1"))
                                      .factory(clientFactory)
                                      .build();

    final AggregatedHttpResponse response = client.get("/").aggregate().get();

    assertThat(response.status()).isEqualTo(HttpStatus.OK);
    assertThat(response.contentUtf8()).isEqualTo("a.com: CN=a.com");
}