ReactorNetty.java

/*
 * Copyright (c) 2011-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;

import java.net.SocketAddress;
import java.nio.channels.FileChannel;
import java.nio.file.Path;
import java.time.ZoneId;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;

import io.netty5.buffer.BufferUtil;
import io.netty5.buffer.Buffer;
import io.netty5.buffer.BufferAllocator;
import io.netty5.buffer.BufferHolder;
import io.netty5.buffer.MemoryManager;
import io.netty5.channel.nio.AbstractNioChannel;
import io.netty5.util.Resource;
import io.netty5.channel.Channel;
import io.netty5.channel.ChannelHandler;
import io.netty5.channel.ChannelHandlerAdapter;
import io.netty5.channel.ChannelHandlerContext;
import io.netty5.channel.ChannelPipeline;
import io.netty5.handler.ssl.SslHandler;
import io.netty5.handler.stream.ChunkedWriteHandler;
import io.netty5.handler.timeout.IdleState;
import io.netty5.handler.timeout.IdleStateEvent;
import io.netty5.handler.timeout.IdleStateHandler;
import io.netty5.util.AttributeKey;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscription;
import reactor.core.CorePublisher;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.SignalType;
import reactor.pool.AllocationStrategy;
import reactor.pool.PoolBuilder;
import reactor.pool.introspection.SamplingAllocationStrategy;
import reactor.util.Logger;
import reactor.util.Loggers;
import reactor.util.annotation.Nullable;
import reactor.util.context.Context;

import static java.util.Objects.requireNonNull;

/**
 * Internal helpers for reactor-netty contracts
 *
 * @author Stephane Maldini
 */
public final class ReactorNetty {

	// System properties names


	/**
	 * Specifies whether the channel ID will be prepended to the log message when possible.
	 * By default, it will be prepended.
	 */
	static final boolean LOG_CHANNEL_INFO =
			Boolean.parseBoolean(System.getProperty("reactor.netty5.logChannelInfo", "true"));

	/**
	 * Default worker thread count, fallback to available processor
	 * (but with a minimum value of 4)
	 */
	public static final String IO_WORKER_COUNT = "reactor.netty5.ioWorkerCount";
	/**
	 * Default selector thread count, fallback to -1 (no selector thread)
	 */
	public static final String IO_SELECT_COUNT = "reactor.netty5.ioSelectCount";
	/**
	 * Default worker thread count for UDP, fallback to available processor
	 * (but with a minimum value of 4)
	 */
	public static final String UDP_IO_THREAD_COUNT = "reactor.netty5.udp.ioThreadCount";
	/**
	 * Default quiet period that guarantees that the disposal of the underlying LoopResources
	 * will not happen, fallback to 2 seconds.
	 */
	public static final String SHUTDOWN_QUIET_PERIOD = "reactor.netty5.ioShutdownQuietPeriod";
	/**
	 * Default maximum amount of time to wait until the disposal of the underlying LoopResources
	 * regardless if a task was submitted during the quiet period, fallback to 15 seconds.
	 */
	public static final String SHUTDOWN_TIMEOUT = "reactor.netty5.ioShutdownTimeout";

	/**
	 * Default value whether the native transport (epoll, kqueue) will be preferred,
	 * fallback it will be preferred when available
	 */
	public static final String NATIVE = "reactor.netty5.native";


	/**
	 * Default max connections. Fallback to
	 * 2 * available number of processors (but with a minimum value of 16)
	 */
	public static final String POOL_MAX_CONNECTIONS = "reactor.netty5.pool.maxConnections";
	/**
	 * Default acquisition timeout (milliseconds) before error. If -1 will never wait to
	 * acquire before opening a new
	 * connection in an unbounded fashion. Fallback 45 seconds
	 */
	public static final String POOL_ACQUIRE_TIMEOUT = "reactor.netty5.pool.acquireTimeout";
	/**
	 * Default max idle time, fallback - max idle time is not specified.
	 * <p><strong>Note:</strong> This configuration is not applicable for {@link reactor.netty5.tcp.TcpClient}.
	 * A TCP connection is always closed and never returned to the pool.
	 */
	public static final String POOL_MAX_IDLE_TIME = "reactor.netty5.pool.maxIdleTime";
	/**
	 * Default max life time, fallback - max life time is not specified.
	 * <p><strong>Note:</strong> This configuration is not applicable for {@link reactor.netty5.tcp.TcpClient}.
	 * A TCP connection is always closed and never returned to the pool.
	 */
	public static final String POOL_MAX_LIFE_TIME = "reactor.netty5.pool.maxLifeTime";
	/**
	 * Default leasing strategy (fifo, lifo), fallback to fifo.
	 * <ul>
	 *     <li>fifo - The connection selection is first in, first out</li>
	 *     <li>lifo - The connection selection is last in, first out</li>
	 * </ul>
	 * <p><strong>Note:</strong> This configuration is not applicable for {@link reactor.netty5.tcp.TcpClient}.
	 * A TCP connection is always closed and never returned to the pool.
	 */
	public static final String POOL_LEASING_STRATEGY = "reactor.netty5.pool.leasingStrategy";
	/**
	 * Default {@code getPermitsSamplingRate} (between 0d and 1d (percentage))
	 * to be used with a {@link SamplingAllocationStrategy}.
	 * This strategy wraps a {@link PoolBuilder#sizeBetween(int, int) sizeBetween} {@link AllocationStrategy}
	 * and samples calls to {@link AllocationStrategy#getPermits(int)}.
	 * Fallback - sampling is not enabled.
	 */
	public static final String POOL_GET_PERMITS_SAMPLING_RATE = "reactor.netty5.pool.getPermitsSamplingRate";
	/**
	 * Default {@code returnPermitsSamplingRate} (between 0d and 1d (percentage))
	 * to be used with a {@link SamplingAllocationStrategy}.
	 * This strategy wraps a {@link PoolBuilder#sizeBetween(int, int) sizeBetween} {@link AllocationStrategy}
	 * and samples calls to {@link AllocationStrategy#returnPermits(int)}.
	 * Fallback - sampling is not enabled.
	 */
	public static final String POOL_RETURN_PERMITS_SAMPLING_RATE = "reactor.netty5.pool.returnPermitsSamplingRate";


	/**
	 * Default SSL handshake timeout (milliseconds), fallback to 10 seconds
	 */
	public static final String SSL_HANDSHAKE_TIMEOUT = "reactor.netty5.tcp.sslHandshakeTimeout";
	/**
	 * Default value whether the SSL debugging on the client side will be enabled/disabled,
	 * fallback to SSL debugging disabled
	 */
	public static final String SSL_CLIENT_DEBUG = "reactor.netty5.tcp.ssl.client.debug";
	/**
	 * Default value whether the SSL debugging on the server side will be enabled/disabled,
	 * fallback to SSL debugging disabled
	 */
	public static final String SSL_SERVER_DEBUG = "reactor.netty5.tcp.ssl.server.debug";


	/**
	 * Specifies whether the Http Server access log will be enabled.
	 * By default, it is disabled.
	 */
	public static final String ACCESS_LOG_ENABLED = "reactor.netty5.http.server.accessLogEnabled";

	/**
	 *  Specifies the zone id used by the access log
	 */
	public static final ZoneId ZONE_ID_SYSTEM = ZoneId.systemDefault();

	/**
	 * Try to call {@link Resource#close()} if the specified message implements {@link Resource}.
	 * If the specified message doesn't implement {@link Resource} or it is already closed,
	 * this method does nothing.
	 */
	public static void safeRelease(Object msg) {
		if (msg instanceof Resource<?> resource && resource.isAccessible()) {
			resource.close();
		}
	}

	/**
	 * Append channel ID to a log message for correlated traces
	 * @param channel current channel associated with the msg
	 * @param msg the log msg
	 * @return a formatted msg
	 */
	public static String format(Channel channel, String msg) {
		Objects.requireNonNull(channel, "channel");
		Objects.requireNonNull(msg, "msg");
		if (LOG_CHANNEL_INFO) {
			String channelStr;
			StringBuilder result;
			Connection connection = Connection.from(channel);
			if (connection instanceof ChannelOperationsId channelOperationsId) {
				channelStr = channelOperationsId.asLongText();
				if (channelStr.charAt(0) != TRACE_ID_PREFIX) {
					result = new StringBuilder(1 + channelStr.length() + 2 + msg.length())
							.append(CHANNEL_ID_PREFIX)
							.append(channelStr)
							.append(CHANNEL_ID_SUFFIX_1);
				}
				else {
					result = new StringBuilder(channelStr.length() + 1 + msg.length())
							.append(channelStr)
							.append(CHANNEL_ID_SUFFIX_2);
				}
				return result.append(msg)
						.toString();
			}
			else {
				channelStr = channel.toString();
				if (channelStr.charAt(0) == CHANNEL_ID_PREFIX) {
					channelStr = channelStr.substring(ORIGINAL_CHANNEL_ID_PREFIX_LENGTH);
					result = new StringBuilder(1 + channelStr.length() + 1 + msg.length())
							.append(CHANNEL_ID_PREFIX)
							.append(channelStr);
				}
				else {
					int ind = channelStr.indexOf(ORIGINAL_CHANNEL_ID_PREFIX);
					result = new StringBuilder(1 + (channelStr.length() - ORIGINAL_CHANNEL_ID_PREFIX_LENGTH) + 1 + msg.length())
							.append(channelStr.substring(0, ind))
							.append(CHANNEL_ID_PREFIX)
							.append(channelStr.substring(ind + ORIGINAL_CHANNEL_ID_PREFIX_LENGTH));
				}
				return result.append(CHANNEL_ID_SUFFIX_2)
						.append(msg)
						.toString();
			}
		}
		else {
			return msg;
		}
	}

	/**
	 * Pretty hex dump will be returned when the object is {@link Buffer}
	 */
	public static String toPrettyHexDump(Object msg) {
		Objects.requireNonNull(msg, "msg");
		String result;
		if (msg instanceof Buffer buffer) {
			int length = buffer.readableBytes();
			int rows = length / 16 + ((length & 15) == 0 ? 0 : 1) + 4;
			StringBuilder buf = new StringBuilder(rows * 80);
			BufferUtil.appendPrettyHexDump(buf, buffer, buffer.readerOffset(), length);
			result = "\n" + buf;
		}
		else {
			result = msg.toString();
		}
		return result;
	}

	/**
	 * Wrap possibly fatal or singleton exception into a new exception instance in order to propagate in reactor flows without side effect.
	 *
	 * @return a wrapped {@link RuntimeException}
	 */
	public static RuntimeException wrapException(Throwable throwable) {
		return new InternalNettyException(Objects.requireNonNull(throwable));
	}

	static void addChunkedWriter(Connection c) {
		if (c.channel()
		     .pipeline()
		     .get(ChunkedWriteHandler.class) == null) {
			c.addHandlerLast(NettyPipeline.ChunkedWriter, new ChunkedWriteHandler());
		}
	}

	/**
	 * A common implementation for the {@link Connection#addHandlerLast(String, ChannelHandler)}
	 * method that can be reused by other implementors.
	 * <p>
	 * This implementation will look for reactor added handlers on the right hand side of
	 * the pipeline, provided they are identified with the {@link NettyPipeline#RIGHT}
	 * prefix, and add the handler just before the first of these.
	 *
	 * @param context the {@link Connection} on which to add the decoder.
	 * @param name the name of the decoder.
	 * @param handler the decoder to add before the final reactor-specific handlers.
	 * @see Connection#addHandlerLast(String, ChannelHandler)
	 */
	static void addHandlerBeforeReactorEndHandlers(Connection context, String
			name,	ChannelHandler handler) {
		Objects.requireNonNull(name, "name");
		Objects.requireNonNull(handler, "handler");

		Channel channel = context.channel();
		boolean exists = channel.pipeline().get(name) != null;

		if (exists) {
			if (log.isDebugEnabled()) {
				log.debug(format(channel, "Handler [{}] already exists in the pipeline, decoder has been skipped"),
						name);
			}
			return;
		}

		//we need to find the correct position
		String before = null;
		for (String s : channel.pipeline().names()) {
			if (s.startsWith(NettyPipeline.RIGHT)) {
				before = s;
				break;
			}
		}

		if (before == null) {
			channel.pipeline().addLast(name, handler);
		}
		else {
			channel.pipeline().addBefore(before, name, handler);
		}

		registerForClose(context.isPersistent(),  name, context);

		if (log.isDebugEnabled()) {
			log.debug(format(channel, "Added decoder [{}] at the end of the user pipeline, full pipeline: {}"),
					name,
					channel.pipeline().names());
		}
	}

	/**
	 * A common implementation for the {@link Connection#addHandlerFirst(String, ChannelHandler)}
	 * method that can be reused by other implementors.
	 * <p>
	 * This implementation will look for reactor added handlers on the left hand side of
	 * the pipeline, provided they are identified with the {@link NettyPipeline#LEFT}
	 * prefix, and add the handler just after the last of these.
	 *
	 * @param context the {@link Connection} on which to add the decoder.
	 * @param name the name of the encoder.
	 * @param handler the encoder to add after the initial reactor-specific handlers.
	 * @see Connection#addHandlerFirst(String, ChannelHandler)
	 */
	static void addHandlerAfterReactorCodecs(Connection context, String
			name,
			ChannelHandler handler) {
		Objects.requireNonNull(name, "name");
		Objects.requireNonNull(handler, "handler");

		Channel channel = context.channel();
		boolean exists = channel.pipeline().get(name) != null;

		if (exists) {
			if (log.isDebugEnabled()) {
				log.debug(format(channel, "Handler [{}] already exists in the pipeline, encoder has been skipped"),
						name);
			}
			return;
		}

		//we need to find the correct position
		String after = null;
		for (String s : channel.pipeline().names()) {
			if (s.startsWith(NettyPipeline.LEFT)) {
				after = s;
			}
		}

		if (after == null) {
			channel.pipeline().addFirst(name, handler);
		}
		else {
			channel.pipeline().addAfter(after, name, handler);
		}

		registerForClose(context.isPersistent(), name, context);

		if (log.isDebugEnabled()) {
			log.debug(format(channel, "Added encoder [{}] at the beginning of the user pipeline, full pipeline: {}"),
					name,
					channel.pipeline().names());
		}
	}

	static boolean mustChunkFileTransfer(Connection c, Path file) {
		// if channel multiplexing a parent channel as an http2 stream
		if (c.channel().parent() != null && c.channel().parent().pipeline().get(NettyPipeline.H2MultiplexHandler) != null) {
			return true;
		}
		ChannelPipeline p = c.channel().pipeline();
		return p.get(SslHandler.class) != null  ||
				p.get(NettyPipeline.CompressionHandler) != null ||
				(!c.channel().executor().isCompatible(AbstractNioChannel.class) &&
						!"file".equals(file.toUri().getScheme()));
	}

	static void registerForClose(boolean shouldCleanupOnClose,
			String name,
			Connection context) {
		if (!shouldCleanupOnClose) {
			return;
		}

		context.onTerminate().subscribe(null, null, () -> context.removeHandler(name));
	}

	static void removeHandler(Channel channel, String name) {
		if (channel.isActive() && channel.pipeline()
		                                 .context(name) != null) {
			channel.pipeline()
			       .remove(name);
			if (log.isDebugEnabled()) {
				log.debug(format(channel, "Removed handler: {}, pipeline: {}"),
						name,
						channel.pipeline());
			}
		}
		else if (log.isDebugEnabled()) {
			log.debug(format(channel, "Non Removed handler: {}, context: {}, pipeline: {}"),
					name,
					channel.pipeline()
					       .context(name),
					channel.pipeline());
		}
	}

	static void replaceHandler(Channel channel, String name, ChannelHandler handler) {
		if (channel.isActive() && channel.pipeline()
		                                 .context(name) != null) {
			channel.pipeline()
			       .replace(name, name, handler);
			if (log.isDebugEnabled()) {
				log.debug(format(channel, "Replaced handler: {}, pipeline: {}"),
						name,
						channel.pipeline());
			}
		}
		else if (log.isDebugEnabled()) {
			log.debug(format(channel, "Non Replaced handler: {}, context: {}, pipeline: {}"),
					name,
					channel.pipeline()
					       .context(name),
					channel.pipeline());
		}
	}

	static ConnectionObserver compositeConnectionObserver(ConnectionObserver observer,
			ConnectionObserver other) {

		if (observer == ConnectionObserver.emptyListener()) {
			return other;
		}

		if (other == ConnectionObserver.emptyListener()) {
			return observer;
		}

		final ConnectionObserver[] newObservers;
		final ConnectionObserver[] thizObservers;
		final ConnectionObserver[] otherObservers;
		int length = 2;

		if (observer instanceof CompositeConnectionObserver compositeConnectionObserver) {
			thizObservers = compositeConnectionObserver.observers;
			length += thizObservers.length - 1;
		}
		else {
			thizObservers = null;
		}

		if (other instanceof CompositeConnectionObserver compositeConnectionObserver) {
			otherObservers = compositeConnectionObserver.observers;
			length += otherObservers.length - 1;
		}
		else {
			otherObservers = null;
		}

		newObservers = new ConnectionObserver[length];

		int pos;
		if (thizObservers != null) {
			pos = thizObservers.length;
			System.arraycopy(thizObservers, 0,
					newObservers, 0,
					pos);
		}
		else {
			pos = 1;
			newObservers[0] = observer;
		}

		if (otherObservers != null) {
			System.arraycopy(otherObservers, 0,
					newObservers, pos,
					otherObservers.length);
		}
		else {
			newObservers[pos] = other;
		}

		return new CompositeConnectionObserver(newObservers);
	}


	static <T, V> CorePublisher<V> publisherOrScalarMap(Publisher<T> publisher,
			Function<? super T, ? extends V> mapper) {

		if (publisher instanceof Callable) {
			return Mono.fromCallable(new ScalarMap<>(publisher, mapper));
		}
		else if (publisher instanceof Mono) {
			return ((Mono<T>) publisher).map(mapper);
		}

		return Flux.from(publisher)
		           .map(mapper);
	}

	static <T, V> CorePublisher<V> publisherOrScalarMap(
			Publisher<T> publisher,
			Function<? super T, ? extends V> monoMapper,
			Function<? super List<T>, ? extends V> fluxMapper) {

		if (publisher instanceof Callable) {
			return Mono.fromCallable(new ScalarMap<>(publisher, monoMapper));
		}
		else if (publisher instanceof Mono) {
			return ((Mono<T>) publisher).map(monoMapper);
		}

		return Flux.from(publisher)
		           .collectList()
		           .map(fluxMapper);
	}

	ReactorNetty() {
	}

	static final class ScalarMap<T, V> implements Callable<V> {

		final Callable<T>                      source;
		final Function<? super T, ? extends V> mapper;

		@SuppressWarnings("unchecked")
		ScalarMap(Publisher<T> source, Function<? super T, ? extends V> mapper) {
			this.source = (Callable<T>) source;
			this.mapper = mapper;
		}

		@Override
		public V call() throws Exception {
			T called = source.call();
			if (called == null) {
				return null;
			}
			return mapper.apply(called);
		}
	}

	static final class CompositeChannelPipelineConfigurer implements ChannelPipelineConfigurer {

		final ChannelPipelineConfigurer[] configurers;

		CompositeChannelPipelineConfigurer(ChannelPipelineConfigurer[] configurers) {
			this.configurers = configurers;
		}

		@Override
		public void onChannelInit(ConnectionObserver connectionObserver, Channel channel, @Nullable SocketAddress remoteAddress) {
			for (ChannelPipelineConfigurer configurer : configurers) {
				configurer.onChannelInit(connectionObserver, channel, remoteAddress);
			}
		}

		static ChannelPipelineConfigurer compositeChannelPipelineConfigurer(
				ChannelPipelineConfigurer configurer, ChannelPipelineConfigurer other) {

			if (configurer == ChannelPipelineConfigurer.emptyConfigurer()) {
				return other;
			}

			if (other == ChannelPipelineConfigurer.emptyConfigurer()) {
				return configurer;
			}

			final ChannelPipelineConfigurer[] newConfigurers;
			final ChannelPipelineConfigurer[] thizConfigurers;
			final ChannelPipelineConfigurer[] otherConfigurers;
			int length = 2;

			if (configurer instanceof CompositeChannelPipelineConfigurer compositeChannelPipelineConfigurer) {
				thizConfigurers = compositeChannelPipelineConfigurer.configurers;
				length += thizConfigurers.length - 1;
			}
			else {
				thizConfigurers = null;
			}

			if (other instanceof CompositeChannelPipelineConfigurer compositeChannelPipelineConfigurer) {
				otherConfigurers = compositeChannelPipelineConfigurer.configurers;
				length += otherConfigurers.length - 1;
			}
			else {
				otherConfigurers = null;
			}

			newConfigurers = new ChannelPipelineConfigurer[length];

			int pos;
			if (thizConfigurers != null) {
				pos = thizConfigurers.length;
				System.arraycopy(thizConfigurers, 0, newConfigurers, 0, pos);
			}
			else {
				pos = 1;
				newConfigurers[0] = configurer;
			}

			if (otherConfigurers != null) {
				System.arraycopy(otherConfigurers, 0, newConfigurers, pos, otherConfigurers.length);
			}
			else {
				newConfigurers[pos] = other;
			}

			return new CompositeChannelPipelineConfigurer(newConfigurers);
		}
	}

	static final class CompositeConnectionObserver implements ConnectionObserver {

		final ConnectionObserver[] observers;

		CompositeConnectionObserver(ConnectionObserver[] observers) {
			this.observers = observers;
		}

		@Override
		public Context currentContext() {
			return observers[observers.length - 1].currentContext();
		}

		@Override
		public void onUncaughtException(Connection connection, Throwable error) {
			for (ConnectionObserver observer : observers) {
				observer.onUncaughtException(connection, error);
			}
		}

		@Override
		public void onStateChange(Connection connection, State newState) {
			for (ConnectionObserver observer : observers) {
				observer.onStateChange(connection, newState);
			}
		}
	}

	/**
	 * An appending write that delegates to its origin context and append the passed
	 * publisher after the origin success if any.
	 */
	static final class OutboundThen implements NettyOutbound {

		final NettyOutbound source;
		final Mono<Void> thenMono;

		static final Runnable EMPTY_CLEANUP = () -> {};


		OutboundThen(NettyOutbound source, Publisher<Void> thenPublisher) {
			this(source, thenPublisher, EMPTY_CLEANUP);
		}

		OutboundThen(NettyOutbound source, Publisher<Void> thenPublisher, Runnable onCleanup) {
			this.source = source;
			Objects.requireNonNull(onCleanup, "onCleanup");

			Mono<Void> parentMono = source.then();

			if (parentMono == Mono.<Void>empty()) {
				if (onCleanup == EMPTY_CLEANUP) {
					this.thenMono = Mono.from(thenPublisher);
				}
				else {
					this.thenMono = Mono.from(thenPublisher)
					                    .doOnCancel(onCleanup)
					                    .doOnError(t -> onCleanup.run());
				}
			}
			else {
				if (onCleanup == EMPTY_CLEANUP) {
					this.thenMono = parentMono.thenEmpty(thenPublisher);
				}
				else {
					this.thenMono = parentMono.thenEmpty(thenPublisher)
					                          .doOnCancel(onCleanup)
					                          .doOnError(t -> onCleanup.run());
				}
			}
		}

		@Override
		public <S> NettyOutbound sendUsing(Callable<? extends S> sourceInput,
				BiFunction<? super Connection, ? super S, ?> mappedInput,
				Consumer<? super S> sourceCleanup) {
			return then(source.sendUsing(sourceInput, mappedInput, sourceCleanup));
		}

		@Override
		public BufferAllocator alloc() {
			return source.alloc();
		}

		@Override
		public NettyOutbound withConnection(Consumer<? super Connection> withConnection) {
			return source.withConnection(withConnection);
		}

		@Override
		public NettyOutbound send(Publisher<? extends Buffer> dataStream, Predicate<Buffer> predicate) {
			return then(source.send(dataStream, predicate));
		}

		@Override
		public NettyOutbound sendObject(Publisher<?> dataStream, Predicate<Object> predicate) {
			return then(source.sendObject(dataStream, predicate));
		}

		@Override
		public NettyOutbound sendObject(Object message) {
			return then(source.sendObject(message),
					() -> ReactorNetty.safeRelease(message));
		}

		@Override
		public Mono<Void> then() {
			return thenMono;
		}
	}

	final static class OutboundIdleStateHandler extends IdleStateHandler {

		final Runnable onWriteIdle;

		OutboundIdleStateHandler(long idleTimeout, Runnable onWriteIdle) {
			super(0, idleTimeout, 0, TimeUnit.MILLISECONDS);
			this.onWriteIdle = requireNonNull(onWriteIdle, "onWriteIdle");
		}

		@Override
		protected void channelIdle(ChannelHandlerContext ctx,
				IdleStateEvent evt) throws Exception {
			if (evt.state() == IdleState.WRITER_IDLE) {
				onWriteIdle.run();
			}
			super.channelIdle(ctx, evt);
		}
	}

	final static class InboundIdleStateHandler extends IdleStateHandler {

		final Runnable onReadIdle;

		InboundIdleStateHandler(long idleTimeout, Runnable onReadIdle) {
			super(idleTimeout, 0, 0, TimeUnit.MILLISECONDS);
			this.onReadIdle = requireNonNull(onReadIdle, "onReadIdle");
		}

		@Override
		protected void channelIdle(ChannelHandlerContext ctx,
				IdleStateEvent evt) throws Exception {
			if (evt.state() == IdleState.READER_IDLE) {
				onReadIdle.run();
			}
			super.channelIdle(ctx, evt);
		}
	}

	final static ConnectionObserver.State CONNECTED = new ConnectionObserver.State() {
		@Override
		public String toString() {
			return "[connected]";
		}
	};

	final static ConnectionObserver.State ACQUIRED = new ConnectionObserver.State() {
		@Override
		public String toString() {
			return "[acquired]";
		}
	};

	final static ConnectionObserver.State CONFIGURED = new ConnectionObserver.State() {
		@Override
		public String toString() {
			return "[configured]";
		}
	};

	final static ConnectionObserver.State RELEASED = new ConnectionObserver.State() {
		@Override
		public String toString() {
			return "[released]";
		}
	};

	final static ConnectionObserver.State DISCONNECTING = new ConnectionObserver.State() {
		@Override
		public String toString() {
			return "[disconnecting]";
		}
	};

	/**
	 * A handler that can be used to extract {@link Buffer} out of {@link BufferHolder},
	 * optionally also outputting additional messages
	 *
	 * @author Stephane Maldini
	 * @author Simon Baslé
	 */
	static final class ExtractorHandler extends ChannelHandlerAdapter {


		final BiConsumer<? super ChannelHandlerContext, Object> extractor;
		ExtractorHandler(BiConsumer<? super ChannelHandlerContext, Object> extractor) {
			this.extractor = Objects.requireNonNull(extractor, "extractor");
		}

		@Override
		public void channelRead(ChannelHandlerContext ctx, Object msg) {
			extractor.accept(ctx, msg);
		}

		@Override
		public boolean isSharable() {
			return true;
		}
	}

	static final class ChannelDisposer extends BaseSubscriber<Void> {

		final DisposableChannel channelDisposable;
		ChannelDisposer(DisposableChannel channelDisposable) {
			this.channelDisposable = channelDisposable;
		}

		@Override
		protected void hookOnSubscribe(Subscription subscription) {
			request(Long.MAX_VALUE);
			channelDisposable.onDispose(this);
		}

		@Override
		protected void hookFinally(SignalType type) {
			if (type != SignalType.CANCEL) {
				channelDisposable.dispose();
			}
		}
	}

	static final class SimpleConnection extends AtomicLong implements Connection {

		final Channel channel;

		SimpleConnection(Channel channel) {
			this.channel = Objects.requireNonNull(channel, "channel");
		}

		@Override
		public Channel channel() {
			return channel;
		}

		@Override
		public String toString() {
			return "SimpleConnection{" + "channel=" + channel + '}';
		}
	}

	static NettyInbound unavailableInbound(Connection c) {
		return new NettyInbound() {
			@Override
			public BufferFlux receive() {
				return BufferFlux.fromInbound(Mono.error(new IllegalStateException("Receiver Unavailable")));
			}

			@Override
			public Flux<?> receiveObject() {
				return Flux.error(new IllegalStateException("Receiver Unavailable"));
			}

			@Override
			public NettyInbound withConnection(Consumer<? super Connection> withConnection) {
				withConnection.accept(c);
				return this;
			}
		};
	}

	static NettyOutbound unavailableOutbound(Connection c) {
		return new NettyOutbound() {

			@Override
			public BufferAllocator alloc() {
				return c.channel().bufferAllocator();
			}

			@Override
			public NettyOutbound send(Publisher<? extends Buffer> dataStream, Predicate<Buffer> predicate) {
				return this;
			}

			@Override
			public NettyOutbound sendObject(Publisher<?> dataStream, Predicate<Object> predicate) {
				return this;
			}

			@Override
			public NettyOutbound sendObject(Object message) {
				return this;
			}

			@Override
			public <S> NettyOutbound sendUsing(Callable<? extends S> sourceInput,
					BiFunction<? super Connection, ? super S, ?> mappedInput,
					Consumer<? super S> sourceCleanup) {
				return this;
			}

			@Override
			public NettyOutbound withConnection(Consumer<? super Connection> withConnection) {
				withConnection.accept(c);
				return this;
			}

			@Override
			public Mono<Void> then() {
				return Mono.error(new IllegalStateException("Sender Unavailable"));
			}
		};
	}

	static final class InternalNettyException extends RuntimeException {

		InternalNettyException(Throwable cause) {
			super(cause);
		}

		@Override
		public synchronized Throwable fillInStackTrace() {
			return this;
		}

		private static final long serialVersionUID = 6643227207055930902L;
	}

	static final ChannelPipelineConfigurer NOOP_CONFIGURER = (observer, ch, address) -> {};

	static final ConnectionObserver NOOP_LISTENER = (connection, newState) -> {};

	static final Logger log                               = Loggers.getLogger(ReactorNetty.class);
	static final AttributeKey<Boolean> PERSISTENT_CHANNEL = AttributeKey.valueOf("$PERSISTENT_CHANNEL");

	static final AttributeKey<Connection> CONNECTION = AttributeKey.valueOf("$CONNECTION");

	static final Consumer<? super FileChannel> fileCloser = fc -> {
		try {
			fc.close();
		}
		catch (Throwable e) {
			if (log.isTraceEnabled()) {
				log.trace("", e);
			}
		}
	};


	static final Predicate<Buffer>         PREDICATE_BUFFER_FLUSH = b -> false;

	static final Predicate<Object>         PREDICATE_FLUSH        = o -> false;

	static final Buffer                    BOUNDARY               = MemoryManager.unsafeWrap(new byte[0]).makeReadOnly();

	static final char CHANNEL_ID_PREFIX = '[';
	static final String CHANNEL_ID_SUFFIX_1 = "] ";
	static final char CHANNEL_ID_SUFFIX_2 = ' ';
	static final String ORIGINAL_CHANNEL_ID_PREFIX = "[id: 0x";
	static final int ORIGINAL_CHANNEL_ID_PREFIX_LENGTH = ORIGINAL_CHANNEL_ID_PREFIX.length();
	static final char TRACE_ID_PREFIX = '(';

	@SuppressWarnings("ReferenceEquality")
	//Design to use reference comparison here
	public static final Predicate<Buffer> PREDICATE_GROUP_FLUSH = b -> b == BOUNDARY;

}