HttpClientConfig.java

/*
 * Copyright (c) 2020-2022 VMware, Inc. or its affiliates, All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package reactor.netty5.http.client;

import java.net.SocketAddress;
import java.net.URI;
import java.nio.charset.Charset;
import java.time.Duration;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.BiPredicate;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.regex.Pattern;

import io.netty5.channel.Channel;
import io.netty5.channel.ChannelHandler;
import io.netty5.channel.ChannelHandlerAdapter;
import io.netty5.channel.ChannelHandlerContext;
import io.netty5.channel.ChannelInitializer;
import io.netty5.channel.ChannelOption;
import io.netty5.channel.ChannelPipeline;
import io.netty5.handler.codec.http.DefaultHttpHeaders;
import io.netty5.handler.codec.http.HttpClientCodec;
import io.netty5.handler.codec.http.HttpClientUpgradeHandler;
import io.netty5.handler.codec.http.HttpContentDecompressor;
import io.netty5.handler.codec.http.HttpHeaders;
import io.netty5.handler.codec.http.HttpMethod;
import io.netty5.handler.codec.http.cookie.ClientCookieDecoder;
import io.netty5.handler.codec.http.cookie.ClientCookieEncoder;
import io.netty5.handler.codec.http2.Http2ClientUpgradeCodec;
import io.netty5.handler.codec.http2.Http2FrameCodec;
import io.netty5.handler.codec.http2.Http2FrameCodecBuilder;
import io.netty5.handler.codec.http2.Http2FrameLogger;
import io.netty5.handler.codec.http2.Http2MultiplexHandler;
import io.netty5.handler.codec.http2.Http2Settings;
import io.netty5.handler.codec.http2.Http2StreamFrameToHttpObjectCodec;
import io.netty5.handler.flush.FlushConsolidationHandler;
import io.netty5.handler.logging.LogLevel;
import io.netty5.handler.logging.LoggingHandler;
import io.netty5.handler.ssl.ApplicationProtocolNames;
import io.netty5.handler.ssl.SslHandler;
import io.netty5.handler.timeout.ReadTimeoutHandler;
import io.netty5.resolver.AddressResolverGroup;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Mono;
import reactor.netty5.ChannelPipelineConfigurer;
import reactor.netty5.Connection;
import reactor.netty5.ConnectionObserver;
import reactor.netty5.NettyOutbound;
import reactor.netty5.NettyPipeline;
import reactor.netty5.ReactorNetty;
import reactor.netty5.channel.ChannelMetricsRecorder;
import reactor.netty5.channel.ChannelOperations;
import reactor.netty5.http.Http2SettingsSpec;
import reactor.netty5.http.HttpProtocol;
import reactor.netty5.http.HttpResources;
import reactor.netty5.resources.ConnectionProvider;
import reactor.netty5.resources.LoopResources;
import reactor.netty5.tcp.SslProvider;
import reactor.netty5.transport.ClientTransportConfig;
import reactor.netty5.transport.ProxyProvider;
import reactor.netty5.transport.logging.AdvancedByteBufFormat;
import reactor.util.Logger;
import reactor.util.Loggers;
import reactor.util.annotation.Nullable;
import reactor.util.context.Context;

import static reactor.netty5.ReactorNetty.format;
import static reactor.netty5.http.client.Http2ConnectionProvider.OWNER;

/**
 * Encapsulate all necessary configuration for HTTP client transport. The public API is read-only.
 *
 * @author Stephane Maldini
 * @author Violeta Georgieva
 */
public final class HttpClientConfig extends ClientTransportConfig<HttpClientConfig> {

	/**
	 * Return the configured base URL to use for this request/response or null.
	 *
	 * @return the configured base URL to use for this request/response or null
	 */
	@Nullable
	public String baseUrl() {
		return baseUrl;
	}

	@Override
	public int channelHash() {
		return Objects.hash(super.channelHash(), acceptGzip, decoder, _protocols, sslProvider, uriTagValue);
	}

	@Override
	public ChannelOperations.OnSetup channelOperationsProvider() {
		return (ch, c, msg) -> new HttpClientOperations(ch, c, cookieEncoder, cookieDecoder);
	}

	/**
	 * Return the configured {@link ClientCookieDecoder} or the default {@link ClientCookieDecoder#STRICT}.
	 *
	 * @return the configured {@link ClientCookieDecoder} or the default {@link ClientCookieDecoder#STRICT}
	 */
	public ClientCookieDecoder cookieDecoder() {
		return cookieDecoder;
	}

	/**
	 * Return the configured {@link ClientCookieEncoder} or the default {@link ClientCookieEncoder#STRICT}.
	 *
	 * @return the configured {@link ClientCookieEncoder} or the default {@link ClientCookieEncoder#STRICT}
	 */
	public ClientCookieEncoder cookieEncoder() {
		return cookieEncoder;
	}

	/**
	 * Return the configured HTTP response decoder options or the default.
	 *
	 * @return the configured HTTP request decoder options or the default
	 */
	public HttpResponseDecoderSpec decoder() {
		return decoder;
	}

	/**
	 * Return the configured follow redirect predicate or null.
	 *
	 * @return the configured follow redirect predicate or null
	 */
	@Nullable
	public BiPredicate<HttpClientRequest, HttpClientResponse> followRedirectPredicate() {
		return followRedirectPredicate;
	}

	/**
	 * Return a copy of the request headers.
	 *
	 * @return a copy of the request headers
	 */
	public HttpHeaders headers() {
		return headers.copy();
	}

	/**
	 * Return the HTTP/2 configuration
	 *
	 * @return the HTTP/2 configuration
	 */
	public Http2SettingsSpec http2SettingsSpec() {
		return http2Settings;
	}

	/**
	 * Return whether GZip compression is enabled.
	 *
	 * @return whether GZip compression is enabled
	 */
	public boolean isAcceptGzip() {
		return acceptGzip;
	}

	/**
	 * Return true if {@code retry once} is disabled, false otherwise.
	 *
	 * @return true if {@code retry once} is disabled, false otherwise
	 */
	public boolean isRetryDisabled() {
		return retryDisabled;
	}

	/**
	 * Returns true if that {@link HttpClient} secured via SSL transport
	 *
	 * @return true if that {@link HttpClient} secured via SSL transport
	 */
	public boolean isSecure() {
		return sslProvider != null;
	}

	/**
	 * Return the configured request method.
	 *
	 * @return the configured request method
	 */
	public HttpMethod method() {
		return method;
	}

	/**
	 * Return the HTTP protocol to support. Default is {@link HttpProtocol#HTTP11}.
	 *
	 * @return the HTTP protocol to support
	 */
	public HttpProtocol[] protocols() {
		return protocols;
	}

	/**
	 * Return the configured redirect request {@link BiConsumer} or null.
	 *
	 * @return the configured redirect request {@link BiConsumer} or null
	 */
	@Nullable
	public BiConsumer<HttpHeaders, HttpClientRequest> redirectRequestBiConsumer() {
		return redirectRequestBiConsumer;
	}

	/**
	 * Return the configured redirect request consumer or null.
	 *
	 * @return the configured redirect request consumer or null
	 */
	@Nullable
	public Consumer<HttpClientRequest> redirectRequestConsumer() {
		return redirectRequestConsumer;
	}

	/**
	 * Return the configured response timeout or null
	 *
	 * @return the configured response timeout or null
	 */
	@Nullable
	public Duration responseTimeout() {
		return responseTimeout;
	}

	/**
	 * Returns the current {@link SslProvider} if that {@link HttpClient} secured via SSL
	 * transport or null
	 *
	 * @return the current {@link SslProvider} if that {@link HttpClient} secured via SSL
	 * transport or null
	 */
	@Nullable
	public SslProvider sslProvider() {
		return sslProvider;
	}

	/**
	 * Return the configured request uri.
	 *
	 * @return the configured request uri
	 */
	public String uri() {
		return uri == null ? uriStr : uri.toString();
	}

	/**
	 * Returns the configured function that receives the actual uri and returns the uri tag value
	 * that will be used for the metrics with {@link reactor.netty5.Metrics#URI} tag
	 *
	 * @return the configured function that receives the actual uri and returns the uri tag value
	 * that will be used for the metrics with {@link reactor.netty5.Metrics#URI} tag
	 */
	@Nullable
	public Function<String, String> uriTagValue() {
		return uriTagValue;
	}

	/**
	 * Return the configured websocket client configuration.
	 *
	 * @return the configured websocket client configuration
	 */
	public WebsocketClientSpec websocketClientSpec() {
		return websocketClientSpec;
	}


	// Protected/Package private write API

	boolean acceptGzip;
	String baseUrl;
	BiFunction<? super HttpClientRequest, ? super NettyOutbound, ? extends Publisher<Void>> body;
	Function<? super Mono<? extends Connection>, ? extends Mono<? extends Connection>> connector;
	ClientCookieDecoder cookieDecoder;
	ClientCookieEncoder cookieEncoder;
	HttpResponseDecoderSpec decoder;
	Function<Mono<HttpClientConfig>, Mono<HttpClientConfig>> deferredConf;
	BiConsumer<? super HttpClientRequest, ? super Connection> doAfterRequest;
	BiConsumer<? super HttpClientResponse, ? super Connection> doAfterResponseSuccess;
	BiConsumer<? super HttpClientResponse, ? super Connection> doOnRedirect;
	BiConsumer<? super HttpClientRequest, ? super Connection> doOnRequest;
	BiConsumer<? super HttpClientRequest, ? super Throwable> doOnRequestError;
	BiConsumer<? super HttpClientResponse, ? super Connection> doOnResponse;
	BiConsumer<? super HttpClientResponse, ? super Throwable> doOnResponseError;
	BiPredicate<HttpClientRequest, HttpClientResponse> followRedirectPredicate;
	HttpHeaders headers;
	Http2SettingsSpec http2Settings;
	HttpMethod method;
	HttpProtocol[] protocols;
	int _protocols;
	BiConsumer<HttpHeaders, HttpClientRequest> redirectRequestBiConsumer;
	Consumer<HttpClientRequest> redirectRequestConsumer;
	Duration responseTimeout;
	boolean retryDisabled;
	SslProvider sslProvider;
	URI uri;
	String uriStr;
	Function<String, String> uriTagValue;
	WebsocketClientSpec websocketClientSpec;

	HttpClientConfig(ConnectionProvider connectionProvider, Map<ChannelOption<?>, ?> options,
			Supplier<? extends SocketAddress> remoteAddress) {
		super(connectionProvider, options, remoteAddress);
		this.acceptGzip = false;
		this.cookieDecoder = ClientCookieDecoder.STRICT;
		this.cookieEncoder = ClientCookieEncoder.STRICT;
		this.decoder = new HttpResponseDecoderSpec();
		this.headers = new DefaultHttpHeaders();
		this.method = HttpMethod.GET;
		this.protocols = new HttpProtocol[]{HttpProtocol.HTTP11};
		this._protocols = h11;
		this.retryDisabled = false;
	}

	HttpClientConfig(HttpClientConfig parent) {
		super(parent);
		this.acceptGzip = parent.acceptGzip;
		this.baseUrl = parent.baseUrl;
		this.body = parent.body;
		this.connector = parent.connector;
		this.cookieDecoder = parent.cookieDecoder;
		this.cookieEncoder = parent.cookieEncoder;
		this.decoder = parent.decoder;
		this.deferredConf = parent.deferredConf;
		this.doAfterRequest = parent.doAfterRequest;
		this.doAfterResponseSuccess = parent.doAfterResponseSuccess;
		this.doOnRedirect = parent.doOnRedirect;
		this.doOnRequest = parent.doOnRequest;
		this.doOnRequestError = parent.doOnRequestError;
		this.doOnResponse = parent.doOnResponse;
		this.doOnResponseError = parent.doOnResponseError;
		this.followRedirectPredicate = parent.followRedirectPredicate;
		this.headers = parent.headers;
		this.http2Settings = parent.http2Settings;
		this.method = parent.method;
		this.protocols = parent.protocols;
		this._protocols = parent._protocols;
		this.redirectRequestBiConsumer = parent.redirectRequestBiConsumer;
		this.redirectRequestConsumer = parent.redirectRequestConsumer;
		this.responseTimeout = parent.responseTimeout;
		this.retryDisabled = parent.retryDisabled;
		this.sslProvider = parent.sslProvider;
		this.uri = parent.uri;
		this.uriStr = parent.uriStr;
		this.uriTagValue = parent.uriTagValue;
		this.websocketClientSpec = parent.websocketClientSpec;
	}

	/**
	 * Provides a global {@link AddressResolverGroup} from {@link HttpResources}
	 * that is shared amongst all HTTP clients. {@link AddressResolverGroup} uses the global
	 * {@link LoopResources} from {@link HttpResources}.
	 *
	 * @return the global {@link AddressResolverGroup}
	 */
	@Override
	public AddressResolverGroup<?> defaultAddressResolverGroup() {
		return HttpResources.get().getOrCreateDefaultResolver();
	}

	@Override
	protected ConnectionObserver defaultConnectionObserver() {
		if (doAfterRequest == null && doAfterResponseSuccess == null && doOnRedirect == null &&
					doOnRequest == null && doOnRequestError == null && doOnResponse == null && doOnResponseError == null) {
			return super.defaultConnectionObserver();
		}
		else {
			return super.defaultConnectionObserver()
			            .then(new HttpClientDoOn(doAfterRequest, doAfterResponseSuccess, doOnRedirect, doOnRequest,
			                doOnRequestError, doOnResponse, doOnResponseError));
		}
	}

	@Override
	protected LoggingHandler defaultLoggingHandler() {
		return LOGGING_HANDLER;
	}

	@Override
	protected LoopResources defaultLoopResources() {
		return HttpResources.get();
	}

	@Override
	protected ChannelMetricsRecorder defaultMetricsRecorder() {
		return MicrometerHttpClientMetricsRecorder.INSTANCE;
	}

	@Override
	protected ChannelPipelineConfigurer defaultOnChannelInit() {
		return super.defaultOnChannelInit()
		            .then(new HttpClientChannelInitializer(this));
	}

	@Override
	protected void loggingHandler(LoggingHandler loggingHandler) {
		super.loggingHandler(loggingHandler);
	}

	@Override
	protected void metricsRecorder(@Nullable Supplier<? extends ChannelMetricsRecorder> metricsRecorder) {
		super.metricsRecorder(metricsRecorder);
	}

	@Override
	protected void proxyProvider(ProxyProvider proxyProvider) {
		super.proxyProvider(proxyProvider);
	}

	@Override
	protected AddressResolverGroup<?> resolverInternal() {
		return super.resolverInternal();
	}

	void deferredConf(Function<HttpClientConfig, Mono<HttpClientConfig>> deferrer) {
		if (deferredConf != null) {
			deferredConf = deferredConf.andThen(deferredConf -> deferredConf.flatMap(deferrer));
		}
		else {
			deferredConf = deferredConf -> deferredConf.flatMap(deferrer);
		}
	}

	void protocols(HttpProtocol... protocols) {
		this.protocols = protocols;
		int _protocols = 0;

		for (HttpProtocol p : protocols) {
			if (p == HttpProtocol.HTTP11) {
				_protocols |= h11;
			}
			else if (p == HttpProtocol.H2) {
				_protocols |= h2;
			}
			else if (p == HttpProtocol.H2C) {
				_protocols |= h2c;
			}
		}

		this._protocols = _protocols;
	}

	boolean checkProtocol(int protocol) {
		return (_protocols & protocol) == protocol;
	}

	Http2Settings http2Settings() {
		Http2Settings settings = Http2Settings.defaultSettings();

		if (http2Settings != null) {
			Long headerTableSize = http2Settings.headerTableSize();
			if (headerTableSize != null) {
				settings.headerTableSize(headerTableSize);
			}

			Integer initialWindowSize = http2Settings.initialWindowSize();
			if (initialWindowSize != null) {
				settings.initialWindowSize(initialWindowSize);
			}

			Long maxConcurrentStreams = http2Settings.maxConcurrentStreams();
			if (maxConcurrentStreams != null) {
				settings.maxConcurrentStreams(maxConcurrentStreams);
			}

			Integer maxFrameSize = http2Settings.maxFrameSize();
			if (maxFrameSize != null) {
				settings.maxFrameSize(maxFrameSize);
			}

			settings.maxHeaderListSize(http2Settings.maxHeaderListSize());

			Boolean pushEnabled = http2Settings.pushEnabled();
			if (pushEnabled != null) {
				settings.pushEnabled(pushEnabled);
			}
		}

		return settings;
	}

	static void addStreamHandlers(
			Channel ch,
			ConnectionObserver obs,
			ChannelOperations.OnSetup opsFactory,
			boolean acceptGzip,
			@Nullable ChannelMetricsRecorder metricsRecorder,
			long responseTimeoutMillis,
			@Nullable Function<String, String> uriTagValue) {

		if (HttpClientOperations.log.isDebugEnabled()) {
			HttpClientOperations.log.debug(format(ch, "New HTTP/2 stream"));
		}

		ChannelPipeline pipeline = ch.pipeline();
		pipeline.addLast(NettyPipeline.H2ToHttp11Codec, HTTP2_STREAM_FRAME_TO_HTTP_OBJECT)
				.addLast(NettyPipeline.HttpTrafficHandler, HTTP_2_STREAM_BRIDGE_CLIENT_HANDLER);

		if (acceptGzip) {
			pipeline.addLast(NettyPipeline.HttpDecompressor, new HttpContentDecompressor());
		}

		ChannelOperations.addReactiveBridge(ch, opsFactory, obs);

		if (metricsRecorder != null) {
			if (metricsRecorder instanceof HttpClientMetricsRecorder) {
				ChannelHandler handler;
				Channel parent = ch.parent();
				ChannelHandler existingHandler = parent.pipeline().get(NettyPipeline.HttpMetricsHandler);
				if (existingHandler != null) {
					// This use case can happen only in HTTP/2 clear text connection upgrade
					parent.pipeline().remove(NettyPipeline.HttpMetricsHandler);
					if (metricsRecorder instanceof MicrometerHttpClientMetricsRecorder) {
						handler = new MicrometerHttpClientMetricsHandler((MicrometerHttpClientMetricsHandler) existingHandler);
					}
					else if (metricsRecorder instanceof ContextAwareHttpClientMetricsRecorder) {
						handler = new ContextAwareHttpClientMetricsHandler((ContextAwareHttpClientMetricsHandler) existingHandler);
					}
					else {
						handler = new HttpClientMetricsHandler((HttpClientMetricsHandler) existingHandler);
					}
				}
				else {
					if (metricsRecorder instanceof MicrometerHttpClientMetricsRecorder micrometerHttpClientMetricsRecorder) {
						handler = new MicrometerHttpClientMetricsHandler(micrometerHttpClientMetricsRecorder, uriTagValue);
					}
					else if (metricsRecorder instanceof ContextAwareHttpClientMetricsRecorder contextAwareHttpClientMetricsRecorder) {
						handler = new ContextAwareHttpClientMetricsHandler(contextAwareHttpClientMetricsRecorder, uriTagValue);
					}
					else {
						handler = new HttpClientMetricsHandler((HttpClientMetricsRecorder) metricsRecorder, uriTagValue);
					}
				}
				pipeline.addBefore(NettyPipeline.ReactiveBridge, NettyPipeline.HttpMetricsHandler, handler);
			}
		}

		if (responseTimeoutMillis > -1) {
			Connection.from(ch).addHandlerFirst(NettyPipeline.ResponseTimeoutHandler,
					new ReadTimeoutHandler(responseTimeoutMillis, TimeUnit.MILLISECONDS));
		}

		if (log.isDebugEnabled()) {
			log.debug(format(ch, "Initialized HTTP/2 stream pipeline {}"), ch.pipeline());
		}

		ChannelOperations<?, ?> ops = opsFactory.create(Connection.from(ch), obs, null);
		if (ops != null) {
			ops.bind();
		}
	}

	static void configureHttp2Pipeline(ChannelPipeline p, boolean acceptGzip, HttpResponseDecoderSpec decoder,
			Http2Settings http2Settings, ConnectionObserver observer) {
		Http2FrameCodecBuilder http2FrameCodecBuilder =
				Http2FrameCodecBuilder.forClient()
				                      .validateHeaders(decoder.validateHeaders())
				                      .initialSettings(http2Settings);

		if (p.get(NettyPipeline.LoggingHandler) != null) {
			http2FrameCodecBuilder.frameLogger(new Http2FrameLogger(LogLevel.DEBUG,
					"reactor.netty5.http.client.h2"));
		}

		p.addBefore(NettyPipeline.ReactiveBridge, NettyPipeline.H2Flush, new FlushConsolidationHandler(1024, true))
		 .addBefore(NettyPipeline.ReactiveBridge, NettyPipeline.HttpCodec, http2FrameCodecBuilder.build())
		 .addBefore(NettyPipeline.ReactiveBridge, NettyPipeline.H2MultiplexHandler, new Http2MultiplexHandler(new H2Codec(acceptGzip)))
		 .addBefore(NettyPipeline.ReactiveBridge, NettyPipeline.HttpTrafficHandler, new HttpTrafficHandler(observer));
	}

	static void configureHttp11OrH2CleartextPipeline(
			ChannelPipeline p,
			boolean acceptGzip,
			HttpResponseDecoderSpec decoder,
			Http2Settings http2Settings,
			@Nullable ChannelMetricsRecorder metricsRecorder,
			ConnectionObserver observer,
			ChannelOperations.OnSetup opsFactory,
			@Nullable Function<String, String> uriTagValue) {
		HttpClientCodec httpClientCodec =
				new HttpClientCodec(
						decoder.maxInitialLineLength(),
						decoder.maxHeaderSize(),
						decoder.failOnMissingResponse,
						decoder.validateHeaders(),
						decoder.initialBufferSize(),
						decoder.parseHttpAfterConnectRequest,
						decoder.allowDuplicateContentLengths());

		Http2FrameCodecBuilder http2FrameCodecBuilder =
				Http2FrameCodecBuilder.forClient()
						.validateHeaders(decoder.validateHeaders())
						.initialSettings(http2Settings);

		if (p.get(NettyPipeline.LoggingHandler) != null) {
			http2FrameCodecBuilder.frameLogger(new Http2FrameLogger(LogLevel.DEBUG,
					"reactor.netty5.http.client.h2"));
		}

		Http2FrameCodec http2FrameCodec = http2FrameCodecBuilder.build();

		Http2ClientUpgradeCodec upgradeCodec = new Http2ClientUpgradeCodec(http2FrameCodec,
				new H2CleartextCodec(http2FrameCodec, opsFactory, acceptGzip, metricsRecorder, uriTagValue));

		HttpClientUpgradeHandler upgradeHandler = new HttpClientUpgradeHandler(httpClientCodec, upgradeCodec, decoder.h2cMaxContentLength());

		p.addBefore(NettyPipeline.ReactiveBridge, null, httpClientCodec)
		 .addBefore(NettyPipeline.ReactiveBridge, NettyPipeline.H2CUpgradeHandler, upgradeHandler)
		 .addBefore(NettyPipeline.ReactiveBridge, NettyPipeline.HttpTrafficHandler, new HttpTrafficHandler(observer));

		if (acceptGzip) {
			p.addBefore(NettyPipeline.ReactiveBridge, NettyPipeline.HttpDecompressor, new HttpContentDecompressor());
		}

		if (metricsRecorder != null) {
			if (metricsRecorder instanceof HttpClientMetricsRecorder) {
				ChannelHandler handler;
				if (metricsRecorder instanceof MicrometerHttpClientMetricsRecorder micrometerHttpClientMetricsRecorder) {
					handler = new MicrometerHttpClientMetricsHandler(micrometerHttpClientMetricsRecorder, uriTagValue);
				}
				else if (metricsRecorder instanceof ContextAwareHttpClientMetricsRecorder contextAwareHttpClientMetricsRecorder) {
					handler = new ContextAwareHttpClientMetricsHandler(contextAwareHttpClientMetricsRecorder, uriTagValue);
				}
				else {
					handler = new HttpClientMetricsHandler((HttpClientMetricsRecorder) metricsRecorder, uriTagValue);
				}
				p.addBefore(NettyPipeline.ReactiveBridge, NettyPipeline.HttpMetricsHandler, handler);
			}
		}

	}

	static void configureHttp11Pipeline(ChannelPipeline p,
			boolean acceptGzip,
			HttpResponseDecoderSpec decoder,
			@Nullable ChannelMetricsRecorder metricsRecorder,
			@Nullable Function<String, String> uriTagValue) {
		p.addBefore(NettyPipeline.ReactiveBridge,
				NettyPipeline.HttpCodec,
				new HttpClientCodec(
						decoder.maxInitialLineLength(),
						decoder.maxHeaderSize(),
						decoder.failOnMissingResponse,
						decoder.validateHeaders(),
						decoder.initialBufferSize(),
						decoder.parseHttpAfterConnectRequest,
						decoder.allowDuplicateContentLengths()));

		if (acceptGzip) {
			p.addAfter(NettyPipeline.HttpCodec, NettyPipeline.HttpDecompressor, new HttpContentDecompressor());
		}

		if (metricsRecorder != null) {
			if (metricsRecorder instanceof HttpClientMetricsRecorder) {
				ChannelHandler handler;
				if (metricsRecorder instanceof MicrometerHttpClientMetricsRecorder micrometerHttpClientMetricsRecorder) {
					handler = new MicrometerHttpClientMetricsHandler(micrometerHttpClientMetricsRecorder, uriTagValue);
				}
				else if (metricsRecorder instanceof ContextAwareHttpClientMetricsRecorder contextAwareHttpClientMetricsRecorder) {
					handler = new ContextAwareHttpClientMetricsHandler(contextAwareHttpClientMetricsRecorder, uriTagValue);
				}
				else {
					handler = new HttpClientMetricsHandler((HttpClientMetricsRecorder) metricsRecorder, uriTagValue);
				}
				p.addBefore(NettyPipeline.ReactiveBridge, NettyPipeline.HttpMetricsHandler, handler);
			}
		}
	}

	static final Pattern FOLLOW_REDIRECT_CODES = Pattern.compile("30[12378]");

	static final BiPredicate<HttpClientRequest, HttpClientResponse> FOLLOW_REDIRECT_PREDICATE =
			(req, res) -> FOLLOW_REDIRECT_CODES.matcher(res.status()
			                                               .codeAsText())
			                                   .matches();

	static final int h2 = 0b010;

	static final int h2c = 0b001;

	static final int h11 = 0b100;

	static final int h11orH2 = h11 | h2;

	static final int h11orH2C = h11 | h2c;

	static final Http2StreamFrameToHttpObjectCodec HTTP2_STREAM_FRAME_TO_HTTP_OBJECT =
			new Http2StreamFrameToHttpObjectCodec(false);

	static final Http2StreamBridgeClientHandler HTTP_2_STREAM_BRIDGE_CLIENT_HANDLER =
			new Http2StreamBridgeClientHandler();

	static final Logger log = Loggers.getLogger(HttpClientConfig.class);

	static final LoggingHandler LOGGING_HANDLER =
			AdvancedByteBufFormat.HEX_DUMP
					.toLoggingHandler(HttpClient.class.getName(), LogLevel.DEBUG, Charset.defaultCharset());

	/**
	 * Default value whether the SSL debugging on the client side will be enabled/disabled,
	 * fallback to SSL debugging disabled
	 */
	static final boolean SSL_DEBUG = Boolean.parseBoolean(System.getProperty(ReactorNetty.SSL_CLIENT_DEBUG, "false"));

	static final class H2CleartextCodec extends ChannelHandlerAdapter {

		final boolean acceptGzip;
		final Http2FrameCodec http2FrameCodec;
		final ChannelMetricsRecorder metricsRecorder;
		final ChannelOperations.OnSetup opsFactory;
		final Function<String, String> uriTagValue;

		H2CleartextCodec(
				Http2FrameCodec http2FrameCodec,
				ChannelOperations.OnSetup opsFactory,
				boolean acceptGzip,
				@Nullable ChannelMetricsRecorder metricsRecorder,
				@Nullable Function<String, String> uriTagValue) {
			this.acceptGzip = acceptGzip;
			this.http2FrameCodec = http2FrameCodec;
			this.metricsRecorder = metricsRecorder;
			this.opsFactory = opsFactory;
			this.uriTagValue = uriTagValue;
		}

		@Override
		public void handlerAdded(ChannelHandlerContext ctx) {
			ChannelPipeline pipeline = ctx.pipeline();
			ReadTimeoutHandler responseTimeoutHandler =
					(ReadTimeoutHandler) pipeline.get(NettyPipeline.ResponseTimeoutHandler);
			Http2MultiplexHandler http2MultiplexHandler;
			ConnectionObserver channelOwner = ctx.channel().attr(OWNER).get();
			Http2ConnectionProvider.DisposableAcquire owner = null;
			ConnectionObserver obs = null;
			if (channelOwner instanceof Http2ConnectionProvider.DisposableAcquire) {
				owner = (Http2ConnectionProvider.DisposableAcquire) channelOwner;
				obs = owner.obs;
			}
			if (responseTimeoutHandler != null) {
				pipeline.remove(NettyPipeline.ResponseTimeoutHandler);
				http2MultiplexHandler = new Http2MultiplexHandler(new H2Codec(opsFactory, acceptGzip),
						new H2Codec(owner, obs, opsFactory, acceptGzip, metricsRecorder,
								responseTimeoutHandler.getReaderIdleTimeInMillis(), uriTagValue));
			}
			else {
				http2MultiplexHandler = new Http2MultiplexHandler(new H2Codec(opsFactory, acceptGzip),
						new H2Codec(owner, obs, opsFactory, acceptGzip, metricsRecorder, uriTagValue));
			}
			pipeline.addAfter(ctx.name(), NettyPipeline.HttpCodec, http2FrameCodec)
			        .addAfter(NettyPipeline.HttpCodec, NettyPipeline.H2MultiplexHandler, http2MultiplexHandler);
			if (pipeline.get(NettyPipeline.HttpDecompressor) != null) {
				pipeline.remove(NettyPipeline.HttpDecompressor);
			}
			pipeline.remove(NettyPipeline.ReactiveBridge);
			pipeline.remove(this);
		}
	}

	static final class H2Codec extends ChannelInitializer<Channel> {

		final boolean acceptGzip;
		final ChannelMetricsRecorder metricsRecorder;
		final ConnectionObserver observer;
		final ChannelOperations.OnSetup opsFactory;
		final Http2ConnectionProvider.DisposableAcquire owner;
		final long responseTimeoutMillis;
		final Function<String, String> uriTagValue;

		H2Codec(boolean acceptGzip) {
			// Handle inbound streams (server pushes)
			// TODO this is not supported
			this(null, null, null, acceptGzip, null, -1, null);
		}

		H2Codec(@Nullable ChannelOperations.OnSetup opsFactory, boolean acceptGzip) {
			// Handle inbound streams (server pushes)
			// TODO this is not supported
			this(null, null, opsFactory, acceptGzip, null, -1, null);
		}

		H2Codec(
				@Nullable Http2ConnectionProvider.DisposableAcquire owner,
				@Nullable ConnectionObserver observer,
				@Nullable ChannelOperations.OnSetup opsFactory,
				boolean acceptGzip,
				@Nullable ChannelMetricsRecorder metricsRecorder,
				@Nullable Function<String, String> uriTagValue) {
			// Handle outbound and upgrade streams
			this(owner, observer, opsFactory, acceptGzip, metricsRecorder, -1, uriTagValue);
		}

		H2Codec(
				@Nullable Http2ConnectionProvider.DisposableAcquire owner,
				@Nullable ConnectionObserver observer,
				@Nullable ChannelOperations.OnSetup opsFactory,
				boolean acceptGzip,
				@Nullable ChannelMetricsRecorder metricsRecorder,
				long responseTimeoutMillis,
				@Nullable Function<String, String> uriTagValue) {
			// Handle outbound and upgrade streams
			this.acceptGzip = acceptGzip;
			this.metricsRecorder = metricsRecorder;
			this.observer = observer;
			this.opsFactory = opsFactory;
			this.owner = owner;
			this.responseTimeoutMillis = responseTimeoutMillis;
			this.uriTagValue = uriTagValue;
		}

		@Override
		protected void initChannel(Channel ch) {
			if (observer != null && opsFactory != null && owner != null) {
				Http2ConnectionProvider.registerClose(ch, owner);
				addStreamHandlers(ch, observer.then(new StreamConnectionObserver(owner.currentContext())), opsFactory,
						acceptGzip, metricsRecorder, responseTimeoutMillis, uriTagValue);
			}
			else {
				// Handle server pushes (inbound streams)
				// TODO this is not supported
			}
		}
	}

	static final class H2OrHttp11Codec extends ChannelHandlerAdapter {
		final boolean                                    acceptGzip;
		final HttpResponseDecoderSpec                    decoder;
		final Http2Settings                              http2Settings;
		final ChannelMetricsRecorder                     metricsRecorder;
		final ConnectionObserver                         observer;
		final Function<String, String>                   uriTagValue;

		H2OrHttp11Codec(HttpClientChannelInitializer initializer, ConnectionObserver observer) {
			this.acceptGzip = initializer.acceptGzip;
			this.decoder = initializer.decoder;
			this.http2Settings = initializer.http2Settings;
			this.metricsRecorder = initializer.metricsRecorder;
			this.observer = observer;
			this.uriTagValue = initializer.uriTagValue;
		}

		@Override
		public void channelActive(ChannelHandlerContext ctx) {
			SslHandler sslHandler = ctx.pipeline().get(SslHandler.class);
			if (sslHandler == null) {
				throw new IllegalStateException("Cannot determine negotiated application-level protocol.");
			}
			String protocol = sslHandler.applicationProtocol() != null ? sslHandler.applicationProtocol() : ApplicationProtocolNames.HTTP_1_1;
			if (log.isDebugEnabled()) {
				log.debug(format(ctx.channel(), "Negotiated application-level protocol [" + protocol + "]"));
			}
			if (ApplicationProtocolNames.HTTP_2.equals(protocol)) {
				configureHttp2Pipeline(ctx.channel().pipeline(), acceptGzip, decoder, http2Settings, observer);
			}
			else if (ApplicationProtocolNames.HTTP_1_1.equals(protocol)) {
				configureHttp11Pipeline(ctx.channel().pipeline(), acceptGzip, decoder, metricsRecorder, uriTagValue);
			}
			else {
				throw new IllegalStateException("unknown protocol: " + protocol);
			}

			ctx.fireChannelActive();

			ctx.channel().pipeline().remove(this);
		}
	}

	static final class HttpClientChannelInitializer implements ChannelPipelineConfigurer {

		final boolean                                    acceptGzip;
		final HttpResponseDecoderSpec                    decoder;
		final Http2Settings                              http2Settings;
		final ChannelMetricsRecorder                     metricsRecorder;
		final ChannelOperations.OnSetup                  opsFactory;
		final int                                        protocols;
		final SslProvider                                sslProvider;
		final Function<String, String>                   uriTagValue;

		HttpClientChannelInitializer(HttpClientConfig config) {
			this.acceptGzip = config.acceptGzip;
			this.decoder = config.decoder;
			this.http2Settings = config.http2Settings();
			this.metricsRecorder = config.metricsRecorderInternal();
			this.opsFactory = config.channelOperationsProvider();
			this.protocols = config._protocols;
			this.sslProvider = config.sslProvider;
			this.uriTagValue = config.uriTagValue;
		}

		@Override
		public void onChannelInit(ConnectionObserver observer, Channel channel, @Nullable SocketAddress remoteAddress) {
			if (sslProvider != null) {
				sslProvider.addSslHandler(channel, remoteAddress, SSL_DEBUG);

				if ((protocols & h11orH2) == h11orH2) {
					channel.pipeline()
					       .addBefore(NettyPipeline.ReactiveBridge, NettyPipeline.H2OrHttp11Codec,
					               new H2OrHttp11Codec(this, observer));
				}
				else if ((protocols & h11) == h11) {
					configureHttp11Pipeline(channel.pipeline(), acceptGzip, decoder, metricsRecorder, uriTagValue);
				}
				else if ((protocols & h2) == h2) {
					configureHttp2Pipeline(channel.pipeline(), acceptGzip, decoder, http2Settings, observer);
				}
			}
			else {
				if ((protocols & h11orH2C) == h11orH2C) {
					configureHttp11OrH2CleartextPipeline(channel.pipeline(), acceptGzip, decoder, http2Settings, metricsRecorder, observer, opsFactory, uriTagValue);
				}
				else if ((protocols & h11) == h11) {
					configureHttp11Pipeline(channel.pipeline(), acceptGzip, decoder, metricsRecorder, uriTagValue);
				}
				else if ((protocols & h2c) == h2c) {
					configureHttp2Pipeline(channel.pipeline(), acceptGzip, decoder, http2Settings, observer);
				}
			}
		}
	}

	static final class HttpClientDoOn implements ConnectionObserver {

		final BiConsumer<? super HttpClientRequest, ? super Connection> doAfterRequest;
		final BiConsumer<? super HttpClientResponse, ? super Connection> doAfterResponseSuccess;
		final BiConsumer<? super HttpClientResponse, ? super Connection> doOnRedirect;
		final BiConsumer<? super HttpClientRequest, ? super Connection> doOnRequest;
		final BiConsumer<? super HttpClientRequest, ? super Throwable> doOnRequestError;
		final BiConsumer<? super HttpClientResponse, ? super Connection> doOnResponse;
		final BiConsumer<? super HttpClientResponse, ? super Throwable> doOnResponseError;

		HttpClientDoOn(@Nullable BiConsumer<? super HttpClientRequest,  ? super Connection> doAfterRequest,
				@Nullable BiConsumer<? super HttpClientResponse,  ? super Connection> doAfterResponseSuccess,
				@Nullable BiConsumer<? super HttpClientResponse, ? super Connection> doOnRedirect,
				@Nullable BiConsumer<? super HttpClientRequest,  ? super Connection> doOnRequest,
				@Nullable BiConsumer<? super HttpClientRequest,  ? super Throwable> doOnRequestError,
				@Nullable BiConsumer<? super HttpClientResponse, ? super Connection> doOnResponse,
				@Nullable BiConsumer<? super HttpClientResponse, ? super Throwable> doOnResponseError) {
			this.doAfterRequest = doAfterRequest;
			this.doAfterResponseSuccess = doAfterResponseSuccess;
			this.doOnRedirect = doOnRedirect;
			this.doOnRequest = doOnRequest;
			this.doOnRequestError = doOnRequestError;
			this.doOnResponse = doOnResponse;
			this.doOnResponseError = doOnResponseError;
		}

		@Override
		public void onStateChange(Connection connection, State newState) {
			if (doOnRequest != null && newState == HttpClientState.REQUEST_PREPARED) {
				doOnRequest.accept(connection.as(HttpClientOperations.class), connection);
				return;
			}
			if (doAfterResponseSuccess != null && newState == HttpClientState.RESPONSE_COMPLETED) {
				doAfterResponseSuccess.accept(connection.as(HttpClientOperations.class), connection);
				return;
			}
			if (doAfterRequest != null && newState == HttpClientState.REQUEST_SENT) {
				doAfterRequest.accept(connection.as(HttpClientOperations.class), connection);
				return;
			}
			if (doOnResponse != null && newState == HttpClientState.RESPONSE_RECEIVED) {
				doOnResponse.accept(connection.as(HttpClientOperations.class), connection);
				return;
			}
			if (doOnResponseError != null && newState == HttpClientState.RESPONSE_INCOMPLETE) {
				HttpClientOperations ops = connection.as(HttpClientOperations.class);
				if (ops == null) {
					return;
				}
				if (ops.responseState != null) {
					doOnResponseError.accept(ops, new PrematureCloseException("Connection prematurely closed DURING response"));
				}
			}
		}

		@Override
		public void onUncaughtException(Connection connection, Throwable error) {
			if (doOnRedirect != null && error instanceof RedirectClientException) {
				doOnRedirect.accept(connection.as(HttpClientOperations.class), connection);
				return;
			}
			HttpClientOperations ops = connection.as(HttpClientOperations.class);
			if (ops == null) {
				return;
			}
			if (doOnRequestError != null && ops.retrying && ops.responseState == null) {
				doOnRequestError.accept(connection.as(HttpClientOperations.class), error);
				return;
			}
			if (doOnResponseError != null && ops.responseState != null &&
					!(error instanceof RedirectClientException)) {
				doOnResponseError.accept(connection.as(HttpClientOperations.class), error);
			}
		}
	}

	static final class StreamConnectionObserver implements ConnectionObserver {

		final Context context;

		StreamConnectionObserver(Context context) {
			this.context = context;
		}

		@Override
		public Context currentContext() {
			return context;
		}

		@Override
		public void onStateChange(Connection connection, State state) {
			if (state == State.DISCONNECTING) {
				if (!connection.isPersistent() && connection.channel().isActive()) {
					// Will be released by closeFuture
					connection.channel().close();
				}
			}
		}
	}
}