<commit_before>
package org.xbib.elasticsearch.support.client.ingest.index;

import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.util.concurrent.TimeUnit;

import org.elasticsearch.ElasticsearchIllegalStateException;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthStatus;
import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.Requests;
import org.elasticsearch.common.logging.ESLogger;
import org.elasticsearch.common.logging.ESLoggerFactory;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;

import org.xbib.elasticsearch.action.ingest.IngestItemFailure;
import org.xbib.elasticsearch.action.ingest.IngestResponse;
import org.xbib.elasticsearch.action.ingest.index.IngestIndexProcessor;
import org.xbib.elasticsearch.action.ingest.index.IngestIndexRequest;
import org.xbib.elasticsearch.support.client.AbstractIngestTransportClient;
import org.xbib.elasticsearch.support.client.ClientHelper;
import org.xbib.elasticsearch.support.client.State;

/**
 * Ingest index client
 */
public class IngestIndexTransportClient extends AbstractIngestTransportClient {

    private final static ESLogger logger = ESLoggerFactory.getLogger(IngestIndexTransportClient.class.getName());
    /**
     * The default size of a request
     */
    private int maxBulkActions = 100;

    /**
     * The default number of maximum concurrent requests
     */
    private int maxConcurrentBulkRequests = Runtime.getRuntime().availableProcessors() * 4;
    /**
     * The maximum volume
     */
    private ByteSizeValue maxVolume = new ByteSizeValue(10, ByteSizeUnit.MB);

    /**
     * The maximum wait time for responses when shutting down
     */
    private TimeValue maxWaitTime = new TimeValue(60, TimeUnit.SECONDS);

    /**
     * The processor
     */
    private IngestIndexProcessor ingestProcessor;

    private State state;

    /**
     * The last exception if any
     */
    private Throwable throwable;

    private volatile boolean closed = false;

    @Override
    public IngestIndexTransportClient maxActionsPerBulkRequest(int maxBulkActions) {
        this.maxBulkActions = maxBulkActions;
        return this;
    }

    @Override
    public IngestIndexTransportClient maxConcurrentBulkRequests(int maxConcurrentBulkRequests) {
        this.maxConcurrentBulkRequests = maxConcurrentBulkRequests;
        return this;
    }

    @Override
    public IngestIndexTransportClient maxVolumePerBulkRequest(ByteSizeValue maxVolume) {
        this.maxVolume = maxVolume;
        return this;
    }

    @Override
    public IngestIndexTransportClient maxRequestWait(TimeValue timeout) {
        this.maxWaitTime = timeout;
        return this;
    }

    /**
     * Create a new client for this indexer
     *
     * @return this indexer
     */
    public IngestIndexTransportClient newClient() {
        return this.newClient(findURI());
    }

    public IngestIndexTransportClient newClient(Client client) {
        return this.newClient(findURI());
    }

    /**
     * Create new client with concurrent ingestProcessor processor.
     * <p/>
     * The URI describes host and port of the node the client should connect to,
     * with the parameter <tt>es.cluster.name</tt> for the cluster name.
     *
     * @param uri the cluster URI<commit_msg>javadoc<commit_after>
package org.xbib.elasticsearch.support.client.ingest.index;

import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.util.concurrent.TimeUnit;

import org.elasticsearch.ElasticsearchIllegalStateException;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthStatus;
import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.Requests;
import org.elasticsearch.common.logging.ESLogger;
import org.elasticsearch.common.logging.ESLoggerFactory;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;

import org.xbib.elasticsearch.action.ingest.IngestItemFailure;
import org.xbib.elasticsearch.action.ingest.IngestResponse;
import org.xbib.elasticsearch.action.ingest.index.IngestIndexProcessor;
import org.xbib.elasticsearch.action.ingest.index.IngestIndexRequest;
import org.xbib.elasticsearch.support.client.AbstractIngestTransportClient;
import org.xbib.elasticsearch.support.client.ClientHelper;
import org.xbib.elasticsearch.support.client.State;

/**
 * Ingest index client
 */
public class IngestIndexTransportClient extends AbstractIngestTransportClient {

    private final static ESLogger logger = ESLoggerFactory.getLogger(IngestIndexTransportClient.class.getName());
    /**
     * The default size of a request
     */
    private int maxBulkActions = 100;

    /**
     * The default number of maximum concurrent requests
     */
    private int maxConcurrentBulkRequests = Runtime.getRuntime().availableProcessors() * 4;
    /**
     * The maximum volume
     */
    private ByteSizeValue maxVolume = new ByteSizeValue(10, ByteSizeUnit.MB);

    /**
     * The maximum wait time for responses when shutting down
     */
    private TimeValue maxWaitTime = new TimeValue(60, TimeUnit.SECONDS);

    /**
     * The processor
     */
    private IngestIndexProcessor ingestProcessor;

    private State state;

    /**
     * The last exception if any
     */
    private Throwable throwable;

    private volatile boolean closed = false;

    @Override
    public IngestIndexTransportClient maxActionsPerBulkRequest(int maxBulkActions) {
        this.maxBulkActions = maxBulkActions;
        return this;
    }

    @Override
    public IngestIndexTransportClient maxConcurrentBulkRequests(int maxConcurrentBulkRequests) {
        this.maxConcurrentBulkRequests = maxConcurrentBulkRequests;
        return this;
    }

    @Override
    public IngestIndexTransportClient maxVolumePerBulkRequest(ByteSizeValue maxVolume) {
        this.maxVolume = maxVolume;
        return this;
    }

    @Override
    public IngestIndexTransportClient maxRequestWait(TimeValue timeout) {
        this.maxWaitTime = timeout;
        return this;
    }

    /**
     * Create a new client for this indexer
     *
     * @return this indexer
     */
    public IngestIndexTransportClient newClient() {
        return this.newClient(findURI());
    }

    public IngestIndexTransportClient newClient(Client client) {
        return this.newClient(findURI());
    }

    /**
     * Create new client with concurrent ingest processor.
     *
     * The URI describes host and port of the node the client should connect to,
     * with the parameter <tt>es.cluster.name</tt> for the cluster name.
     *
     * @param uri the cluster URI
{'fraction_non_alphanumeric': 0.08198297171106839, 'fraction_numerical': 0.0021971985718209283, 'mean_word_length': 3.7820091923834536, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 8, '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': '6277594', 'n_tokens_mistral': 1963, 'n_tokens_neox': 1971, 'n_words': 446}