<commit_before>package com.github.msemys.esjc.http;

import com.github.msemys.esjc.UserCredentials;
import com.github.msemys.esjc.http.handler.HttpResponseHandler;
import com.github.msemys.esjc.util.concurrent.ResettableLatch;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.DefaultThreadFactory;
import io.netty.util.concurrent.Future;

import java.net.InetSocketAddress;
import java.time.Duration;
import java.util.Base64;
import java.util.Queue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;

import static com.github.msemys.esjc.util.Numbers.isPositive;
import static com.github.msemys.esjc.util.Preconditions.*;
import static com.github.msemys.esjc.util.Strings.*;
import static io.netty.buffer.Unpooled.copiedBuffer;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.concurrent.Executors.newSingleThreadExecutor;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;

/**
 * HTTP client without pipelining support
 */
public class HttpClient implements AutoCloseable {
    private final EventLoopGroup group = new NioEventLoopGroup(0, new DefaultThreadFactory("es-http"));
    private final Bootstrap bootstrap;
    private final String host;
    private final boolean acceptGzip;
    private final long operationTimeoutMillis;

    private final ExecutorService queueExecutor = newSingleThreadExecutor(new DefaultThreadFactory("es-http-queue"));
    private final Queue<HttpOperation> queue = new ConcurrentLinkedQueue<>();
    private final AtomicBoolean isProcessing = new AtomicBoolean();
    private final ResettableLatch received = new ResettableLatch(false);

    private volatile Channel channel;

    private HttpClient(Builder builder) {
        host = builder.address.getHostString();
        acceptGzip = builder.acceptGzip;
        operationTimeoutMillis = builder.operationTimeout.toMillis();

        bootstrap = new Bootstrap()
            .remoteAddress(builder.address)
            .option(ChannelOption.TCP_NODELAY, true)
            .option(ChannelOption.SO_REUSEADDR, false)
            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, (int) builder.connectTimeout.toMillis())
            .group(group)
            .channel(NioSocketChannel.class)
            .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline pipeline = ch.pipeline();

                    pipeline.addLast("http-codec", new HttpClientCodec());
                    if (acceptGzip) {
                        pipeline.addLast("content-decompressor", new HttpContentDecompressor());
                    }
                    pipeline.addLast("object-aggregator", new HttpObjectAggregator(builder.maxContentLength));
                    pipeline.addLast("logger", new LoggingHandler(HttpClient.class, LogLevel.TRACE));
                    pipeline.addLast("response-handler", new HttpResponseHandler());
                }
            });
    }

    public CompletableFuture<FullHttpResponse> send(HttpRequest request) {
        checkNotNull(request, "request is null");
        checkState(isRunning(), "HTTP client is closed");

        request.headers().set(HttpHeaders.Names.HOST, host);

        if (acceptGzip) {
            request.headers().set(HttpHeaders.Names.ACCEPT_ENCODING, HttpHeaders.Values.GZIP);
        }

        CompletableFuture<FullHttpResponse> response = new CompletableFuture<>();
        enqueue(new HttpOperation(request, response));

        return response;
    }

    private void enqueue(HttpOperation operation) {
        checkNotNull(operation, "operation is null");

        queue.offer(operation);

        if (isProcessing.compareAndSet(false, true)) {
            queueExecutor.execute(this::processQueue);
        }
    }

    private void processQueue() {
        do {
            HttpOperation operation;

            while ((operation = queue.poll()) != null) {
                if (channel == null || !channel.isActive()) {
                    try {
                        channel = bootstrap.connect().syncUninterruptibly().channel();
                    } catch (Exception e) {
                        operation.response.completeExceptionally(e);

                        if (isRunning()) {
                            continue;
                        } else {
                            break;
                        }
                    }
                }

                write(operation);
            }

            isProcessing.set(false);
        } while (isRunning() && !queue.isEmpty() && isProcessing.compareAndSet(false, true));
    }

    private void write(HttpOperation operation) {
        received.reset();

        operation.response.whenComplete((r, t) -> received.release());

        HttpResponseHandler responseHandler = channel.pipeline().get(HttpResponseHandler.class);
        responseHandler.pendingResponse = operation.response;

        try {
            channel.writeAndFlush(operation.request).await();

            if (!received.await(operationTimeoutMillis, MILLISECONDS)) {
                channel.close().awaitUninterruptibly();

                String message = String.format("%s %s request never got response from server.",
                    operation.request.getMethod().name(),
                    operation.request.getUri());

                operation.response.completeExceptionally(new TimeoutException(message));
            }
        } catch (InterruptedException e) {
            operation.response.completeExceptionally(e);
        } finally {
            responseHandler.pendingResponse = null;
        }
    }

    private boolean isRunning() {
        return !group.isShuttingDown();
    }

    @Override
    public void close() {
        Future shutdownFuture = group.shutdownGracefully(0, 15, SECONDS);
        queueExecutor.shutdown();

        HttpOperation operation;<commit_msg>maintenance: complete exceptionally with cause exception, when error occurs on write and flush<commit_after>package com.github.msemys.esjc.http;

import com.github.msemys.esjc.UserCredentials;
import com.github.msemys.esjc.http.handler.HttpResponseHandler;
import com.github.msemys.esjc.util.concurrent.ResettableLatch;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.DefaultThreadFactory;
import io.netty.util.concurrent.Future;

import java.net.InetSocketAddress;
import java.time.Duration;
import java.util.Base64;
import java.util.Queue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;

import static com.github.msemys.esjc.util.Numbers.isPositive;
import static com.github.msemys.esjc.util.Preconditions.*;
import static com.github.msemys.esjc.util.Strings.*;
import static io.netty.buffer.Unpooled.copiedBuffer;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.concurrent.Executors.newSingleThreadExecutor;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;

/**
 * HTTP client without pipelining support
 */
public class HttpClient implements AutoCloseable {
    private final EventLoopGroup group = new NioEventLoopGroup(0, new DefaultThreadFactory("es-http"));
    private final Bootstrap bootstrap;
    private final String host;
    private final boolean acceptGzip;
    private final long operationTimeoutMillis;

    private final ExecutorService queueExecutor = newSingleThreadExecutor(new DefaultThreadFactory("es-http-queue"));
    private final Queue<HttpOperation> queue = new ConcurrentLinkedQueue<>();
    private final AtomicBoolean isProcessing = new AtomicBoolean();
    private final ResettableLatch received = new ResettableLatch(false);

    private volatile Channel channel;

    private HttpClient(Builder builder) {
        host = builder.address.getHostString();
        acceptGzip = builder.acceptGzip;
        operationTimeoutMillis = builder.operationTimeout.toMillis();

        bootstrap = new Bootstrap()
            .remoteAddress(builder.address)
            .option(ChannelOption.TCP_NODELAY, true)
            .option(ChannelOption.SO_REUSEADDR, false)
            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, (int) builder.connectTimeout.toMillis())
            .group(group)
            .channel(NioSocketChannel.class)
            .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline pipeline = ch.pipeline();

                    pipeline.addLast("http-codec", new HttpClientCodec());
                    if (acceptGzip) {
                        pipeline.addLast("content-decompressor", new HttpContentDecompressor());
                    }
                    pipeline.addLast("object-aggregator", new HttpObjectAggregator(builder.maxContentLength));
                    pipeline.addLast("logger", new LoggingHandler(HttpClient.class, LogLevel.TRACE));
                    pipeline.addLast("response-handler", new HttpResponseHandler());
                }
            });
    }

    public CompletableFuture<FullHttpResponse> send(HttpRequest request) {
        checkNotNull(request, "request is null");
        checkState(isRunning(), "HTTP client is closed");

        request.headers().set(HttpHeaders.Names.HOST, host);

        if (acceptGzip) {
            request.headers().set(HttpHeaders.Names.ACCEPT_ENCODING, HttpHeaders.Values.GZIP);
        }

        CompletableFuture<FullHttpResponse> response = new CompletableFuture<>();
        enqueue(new HttpOperation(request, response));

        return response;
    }

    private void enqueue(HttpOperation operation) {
        checkNotNull(operation, "operation is null");

        queue.offer(operation);

        if (isProcessing.compareAndSet(false, true)) {
            queueExecutor.execute(this::processQueue);
        }
    }

    private void processQueue() {
        do {
            HttpOperation operation;

            while ((operation = queue.poll()) != null) {
                if (channel == null || !channel.isActive()) {
                    try {
                        channel = bootstrap.connect().syncUninterruptibly().channel();
                    } catch (Exception e) {
                        operation.response.completeExceptionally(e);

                        if (isRunning()) {
                            continue;
                        } else {
                            break;
                        }
                    }
                }

                write(operation);
            }

            isProcessing.set(false);
        } while (isRunning() && !queue.isEmpty() && isProcessing.compareAndSet(false, true));
    }

    private void write(HttpOperation operation) {
        received.reset();

        operation.response.whenComplete((r, t) -> received.release());

        HttpResponseHandler responseHandler = channel.pipeline().get(HttpResponseHandler.class);
        responseHandler.pendingResponse = operation.response;

        try {
            channel.writeAndFlush(operation.request).sync();

            if (!received.await(operationTimeoutMillis, MILLISECONDS)) {
                channel.close().awaitUninterruptibly();

                String message = String.format("%s %s request never got response from server.",
                    operation.request.getMethod().name(),
                    operation.request.getUri());

                operation.response.completeExceptionally(new TimeoutException(message));
            }
        } catch (Exception e) {
            operation.response.completeExceptionally(e);
        } finally {
            responseHandler.pendingResponse = null;
        }
    }

    private boolean isRunning() {
        return !group.isShuttingDown();
    }

    @Override
    public void close() {
        Future shutdownFuture = group.shutdownGracefully(0, 15, SECONDS);
        queueExecutor.shutdown();

        HttpOperation operation;
{'fraction_non_alphanumeric': 0.09866970733561384, 'fraction_numerical': 0.001064234131508932, 'mean_word_length': 2.6676888764984668, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2139564', 'n_tokens_mistral': 3403, 'n_tokens_neox': 3276, 'n_words': 645}