Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -76,15 +76,18 @@ protected boolean addMockHttpTransport() {
public void testThatNettyHttpServerSupportsHttp2OrHttp3Get() throws Exception {
assumeThat("HTTP/3 is not available on this arch/platform", Http3Utils.isHttp3Available(), is(true));

ensureGreen();
ensureFullyConnectedCluster();

String[] requests = new String[] { "/", "/_nodes/stats", "/", "/_cluster/state", "/" };
HttpServerTransport httpServerTransport = internalCluster().getInstance(HttpServerTransport.class);
TransportAddress[] boundAddresses = httpServerTransport.boundAddress().boundAddresses();
TransportAddress transportAddress = randomFrom(boundAddresses);

@SuppressWarnings("unchecked")
final Tuple<Netty4HttpClient, String> client = randomFrom(
Tuple.tuple(Netty4HttpClient.http3(), "h2="),
Tuple.tuple(Netty4HttpClient.https(), "h3=")
Tuple.tuple(Netty4HttpClient.http3().withLogger(logger), "h2="),
Tuple.tuple(Netty4HttpClient.https().withLogger(logger), "h3=")
);

try (Netty4HttpClient nettyHttpClient = client.v1()) {
Expand All @@ -107,15 +110,18 @@ public void testThatNettyHttpServerSupportsHttp2OrHttp3Get() throws Exception {
public void testThatNettyHttpServerSupportsHttp2OrHttp3Post() throws Exception {
assumeThat("HTTP/3 is not available on this arch/platform", Http3Utils.isHttp3Available(), is(true));

ensureGreen();
ensureFullyConnectedCluster();

final List<Tuple<String, CharSequence>> requests = List.of(Tuple.tuple("/_search", "{\"query\":{ \"match_all\":{}}}"));
HttpServerTransport httpServerTransport = internalCluster().getInstance(HttpServerTransport.class);
TransportAddress[] boundAddresses = httpServerTransport.boundAddress().boundAddresses();
TransportAddress transportAddress = randomFrom(boundAddresses);

@SuppressWarnings("unchecked")
final Tuple<Netty4HttpClient, String> client = randomFrom(
Tuple.tuple(Netty4HttpClient.http3(), "h2="),
Tuple.tuple(Netty4HttpClient.https(), "h3=")
Tuple.tuple(Netty4HttpClient.http3().withLogger(logger), "h2="),
Tuple.tuple(Netty4HttpClient.https().withLogger(logger), "h3=")
);

try (Netty4HttpClient nettyHttpClient = client.v1()) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -231,6 +231,8 @@ protected void doStart() {
@Override
protected HttpServerChannel bind(InetSocketAddress socketAddress) throws Exception {
ChannelFuture future = bootstrap.bind(socketAddress).sync();
logger.info("Bound to {}", socketAddress);

Channel channel = future.channel();
Netty4HttpServerChannel httpServerChannel = new Netty4HttpServerChannel(channel);
channel.attr(HTTP_SERVER_CHANNEL_KEY).set(httpServerChannel);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,8 @@

package org.opensearch.http.netty4;

import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.Logger;
import org.opensearch.common.TriFunction;
import org.opensearch.common.collect.Tuple;
import org.opensearch.common.settings.Settings;
Expand Down Expand Up @@ -117,7 +119,6 @@
* Tiny helper to send http requests over netty.
*/
public class Netty4HttpClient implements Closeable {

static Collection<String> returnHttpResponseBodies(Collection<FullHttpResponse> responses) {
List<String> list = new ArrayList<>(responses.size());
for (FullHttpResponse response : responses) {
Expand All @@ -137,6 +138,7 @@ static Collection<String> returnOpaqueIds(Collection<FullHttpResponse> responses
private final Bootstrap clientBootstrap;
private final TriFunction<CountDownLatch, Collection<FullHttpResponse>, Boolean, AwaitableChannelInitializer<?>> handlerFactory;
private final boolean secure;
private Logger logger;

Netty4HttpClient(
Bootstrap clientBootstrap,
Expand Down Expand Up @@ -188,6 +190,11 @@ public static Netty4HttpClient http3() {
);
}

public Netty4HttpClient withLogger(Logger logger) {
this.logger = logger;
return this;
}

public List<FullHttpResponse> get(SocketAddress remoteAddress, String... uris) throws InterruptedException {
List<HttpRequest> requests = new ArrayList<>(uris.length);
for (int i = 0; i < uris.length; i++) {
Expand Down Expand Up @@ -244,6 +251,7 @@ private synchronized List<FullHttpResponse> sendRequests(final SocketAddress rem
final List<FullHttpResponse> content = Collections.synchronizedList(new ArrayList<>(requests.size()));

final AwaitableChannelInitializer<?> handler = handlerFactory.apply(latch, content, secure);
handler.logger = logger;
clientBootstrap.handler(handler);

ChannelFuture channelFuture = null;
Expand Down Expand Up @@ -341,6 +349,14 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E
*
*/
private static abstract class AwaitableChannelInitializer<C extends Channel> extends ChannelInitializer<C> {
private Logger logger;

void log(Level level, String message, Object... params) {
if (logger != null) {
logger.log(level, message, params);
}
}

void await() {
// do nothing
}
Expand Down Expand Up @@ -518,6 +534,8 @@ protected void initChannel(DatagramChannel ch) {

@Override
Channel prepare(Bootstrap clientBootstrap, Channel channel) throws InterruptedException {
log(Level.INFO, "[QuicChannel] Connecting to: {}", channel.remoteAddress());

final QuicChannel quicChannel = QuicChannel.newBootstrap(channel)
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 30000) // 30 seconds
.handler(new Http3ClientConnectionHandler())
Expand Down
Loading