<commit_before>package org.influxdb.impl;


import com.squareup.moshi.JsonAdapter;
import com.squareup.moshi.Moshi;
import okhttp3.Headers;
import okhttp3.HttpUrl;
import okhttp3.MediaType;
import okhttp3.OkHttpClient;
import okhttp3.RequestBody;
import okhttp3.ResponseBody;
import okhttp3.logging.HttpLoggingInterceptor;
import okhttp3.logging.HttpLoggingInterceptor.Level;
import okio.BufferedSource;

import org.influxdb.BatchOptions;
import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBException;
import org.influxdb.InfluxDBIOException;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.BoundParameterQuery;
import org.influxdb.dto.Point;
import org.influxdb.dto.Pong;
import org.influxdb.dto.Query;
import org.influxdb.dto.QueryResult;
import org.influxdb.impl.BatchProcessor.HttpBatchEntry;
import org.influxdb.impl.BatchProcessor.UdpBatchEntry;
import retrofit2.Call;
import retrofit2.Callback;
import retrofit2.Response;
import retrofit2.Retrofit;
import retrofit2.converter.moshi.MoshiConverterFactory;

import java.io.EOFException;
import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

/**
 * Implementation of a InluxDB API.
 *
 * @author stefan.majer [at] gmail.com
 */
public class InfluxDBImpl implements InfluxDB {

  static final okhttp3.MediaType MEDIA_TYPE_STRING = MediaType.parse("text/plain");

  private static final String SHOW_DATABASE_COMMAND_ENCODED = Query.encode("SHOW DATABASES");

  /**
   * This static constant holds the http logging log level expected in DEBUG mode
   * It is set by System property {@code org.influxdb.InfluxDB.logLevel}.
   *
   * @see org.influxdb.impl.LOG_LEVEL_PROPERTY
   */
  private static final LogLevel LOG_LEVEL = LogLevel.parseLogLevel(System.getProperty(LOG_LEVEL_PROPERTY));

  private final InetAddress hostAddress;
  private final String username;
  private final String password;
  private String version = "";
  private final Retrofit retrofit;
  private final InfluxDBService influxDBService;
  private BatchProcessor batchProcessor;
  private final AtomicBoolean batchEnabled = new AtomicBoolean(false);
  private final LongAdder writeCount = new LongAdder();
  private final LongAdder unBatchedCount = new LongAdder();
  private final LongAdder batchedCount = new LongAdder();
  private volatile DatagramSocket datagramSocket;
  private final HttpLoggingInterceptor loggingInterceptor;
  private final GzipRequestInterceptor gzipRequestInterceptor;
  private LogLevel logLevel = LogLevel.NONE;
  private JsonAdapter<QueryResult> adapter;
  private String database;
  private String retentionPolicy = "autogen";
  private ConsistencyLevel consistency = ConsistencyLevel.ONE;

  public InfluxDBImpl(final String url, final String username, final String password,
      final OkHttpClient.Builder client) {
    super();
    Moshi moshi = new Moshi.Builder().build();
    this.hostAddress = parseHostAddress(url);
    this.username = username;
    this.password = password;

    this.loggingInterceptor = new HttpLoggingInterceptor();
    setLogLevel(LOG_LEVEL);

    this.gzipRequestInterceptor = new GzipRequestInterceptor();
    this.retrofit = new Retrofit.Builder()
        .baseUrl(url)
        .client(client.addInterceptor(loggingInterceptor).addInterceptor(gzipRequestInterceptor).build())
        .addConverterFactory(MoshiConverterFactory.create())
        .build();
    this.influxDBService = this.retrofit.create(InfluxDBService.class);
    this.adapter = moshi.adapter(QueryResult.class);
  }

    InfluxDBImpl(final String url, final String username, final String password, final OkHttpClient.Builder client,
            final InfluxDBService influxDBService, final JsonAdapter<QueryResult> adapter) {
        super();
        this.hostAddress = parseHostAddress(url);
        this.username = username;
        this.password = password;

        this.loggingInterceptor = new HttpLoggingInterceptor();
        setLogLevel(LOG_LEVEL);

        this.gzipRequestInterceptor = new GzipRequestInterceptor();
        this.retrofit = new Retrofit.Builder()
                .baseUrl(url)
                .client(client.addInterceptor(loggingInterceptor).addInterceptor(gzipRequestInterceptor).build())
                .addConverterFactory(MoshiConverterFactory.create())
                .build();
        this.influxDBService = influxDBService;
        this.adapter = adapter;
    }

  public InfluxDBImpl(final String url, final String username, final String password,
                      final OkHttpClient.Builder client, final String database,
                      final String retentionPolicy, final ConsistencyLevel consistency) {
    this(url, username, password, client);

    setConsistency(consistency);
    setDatabase(database);
    setRetentionPolicy(retentionPolicy);
  }

  private InetAddress parseHostAddress(final String url) {
      HttpUrl httpUrl = HttpUrl.parse(url);

      if (httpUrl == null) {
          throw new IllegalArgumentException("Unable to parse url: " + url);
      }

      try {
          return InetAddress.getByName(httpUrl.host());
      } catch (UnknownHostException e) {
          throw new InfluxDBIOException(e);
      }
  }

  @Override
  public InfluxDB setLogLevel(final LogLevel logLevel) {
    switch (logLevel) {
    case NONE:
      this.loggingInterceptor.setLevel(Level.NONE);
      break;
    case BASIC:
      this.loggingInterceptor.setLevel(Level.BASIC);
      break;
    case HEADERS:
      this.loggingInterceptor.setLevel(Level.HEADERS);
      break;
    case FULL:
      this.loggingInterceptor.setLevel(Level.BODY);
      break;
    default:
      break;
    }
    this.logLevel = logLevel;
    return this;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public InfluxDB enableGzip() {
    this.gzipRequestInterceptor.enable();
    return this;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public InfluxDB disableGzip() {
    this.gzipRequestInterceptor.disable();
    return this;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public boolean isGzipEnabled() {
    return this.gzipRequestInterceptor.isEnabled();
  }

  @Override
  public InfluxDB enableBatch() {
    enableBatch(BatchOptions.DEFAULTS);
    return this;
  }

  @Override
  public InfluxDB enableBatch(final BatchOptions batchOptions) {

    if (this.batchEnabled.get()) {
      throw new IllegalStateException("BatchProcessing is already enabled.");
    }
    this.batchProcessor = BatchProcessor
            .builder(this)
            .actions(batchOptions.getActions())
            .exceptionHandler(batchOptions.getExceptionHandler())
            .interval(batchOptions.getFlushDuration(), batchOptions.getJitterDuration(), TimeUnit.MILLISECONDS)
            .threadFactory(batchOptions.getThreadFactory())
            .bufferLimit(batchOptions.getBufferLimit())
            .consistencyLevel(batchOptions.getConsistency())
            .build();
    this.batchEnabled.set(true);
    return this;
  }

  @Override
  public InfluxDB enableBatch(final int actions, final int flushDuration,
                              final TimeUnit flushDurationTimeUnit) {
    enableBatch(actions, flushDuration, flushDurationTimeUnit, Executors.defaultThreadFactory());
    return this;
  }

  @Override
  public InfluxDB enableBatch(final int actions, final int flushDuration,
                              final TimeUnit flushDurationTimeUnit, final ThreadFactory threadFactory) {
    enableBatch(actions, flushDuration, flushDurationTimeUnit, threadFactory, (points, throwable) -> { });
    return this;
  }

  @Override
  public InfluxDB enableBatch(final int actions, final int flushDuration, final TimeUnit flushDurationTimeUnit,
                              final ThreadFactory threadFactory,
                              final BiConsumer<Iterable<Point>, Throwable> exceptionHandler,
                              final ConsistencyLevel consistency) {
    enableBatch(actions, flushDuration, flushDurationTimeUnit, threadFactory, exceptionHandler)
        .setConsistency(consistency);
    return this;
  }

  @Override
  public InfluxDB enableBatch(final int actions, final int flushDuration, final TimeUnit flushDurationTimeUnit,
                              final ThreadFactory threadFactory,
                              final BiConsumer<Iterable<Point>, Throwable> exceptionHandler) {
    enableBatch(actions, flushDuration, 0, flushDurationTimeUnit, threadFactory, exceptionHandler);
    return this;
  }

  private InfluxDB enableBatch(final int actions, final int flushDuration, final int jitterDuration,
                               final TimeUnit durationTimeUnit, final ThreadFactory threadFactory,
                               final BiConsumer<Iterable<Point>, Throwable> exceptionHandler) {
    if (this.batchEnabled.get()) {
      throw new IllegalStateException("BatchProcessing is already enabled.");
    }
    this.batchProcessor = BatchProcessor
            .builder(this)
            .actions(actions)
            .exceptionHandler(exceptionHandler)
            .interval(flushDuration, jitterDuration, durationTimeUnit)
            .threadFactory(threadFactory)
            .consistencyLevel(consistency)
            .build();
    this.batchEnabled.set(true);
    return this;
  }

  @Override
  public void disableBatch() {
    this.batchEnabled.set(false);
    if (this.batchProcessor != null) {
      this.batchProcessor.flushAndShutdown();
    }
  }

  @Override
  public boolean isBatchEnabled() {
    return this.batchEnabled.get();
  }

  @Override
  public Pong ping() {
    final long started = System.currentTimeMillis();
    Call<ResponseBody> call = this.influxDBService.ping();
    try {
      Response<ResponseBody> response = call.execute();
      Headers headers = response.headers();
      String version = "unknown";
      for (String name : headers.toMultimap().keySet()) {
        if (null != name && "X-Influxdb-Version".equalsIgnoreCase(name)) {
          version = headers.get(name);
          break;
        }
      }
      Pong pong = new Pong();
      pong.setVersion(version);
      pong.setResponseTime(System.currentTimeMillis() - started);
      return pong;
    } catch (IOException e) {
      throw new InfluxDBIOException(e);
    }
  }

  @Override
  public String version() {
    if (version == "") {
      this.version = ping().getVersion();
    }
    return this.version;
  }

  @Override
  public void write(final Point point) {<commit_msg>check for null instead of empty string<commit_after>package org.influxdb.impl;


import com.squareup.moshi.JsonAdapter;
import com.squareup.moshi.Moshi;
import okhttp3.Headers;
import okhttp3.HttpUrl;
import okhttp3.MediaType;
import okhttp3.OkHttpClient;
import okhttp3.RequestBody;
import okhttp3.ResponseBody;
import okhttp3.logging.HttpLoggingInterceptor;
import okhttp3.logging.HttpLoggingInterceptor.Level;
import okio.BufferedSource;

import org.influxdb.BatchOptions;
import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBException;
import org.influxdb.InfluxDBIOException;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.BoundParameterQuery;
import org.influxdb.dto.Point;
import org.influxdb.dto.Pong;
import org.influxdb.dto.Query;
import org.influxdb.dto.QueryResult;
import org.influxdb.impl.BatchProcessor.HttpBatchEntry;
import org.influxdb.impl.BatchProcessor.UdpBatchEntry;
import retrofit2.Call;
import retrofit2.Callback;
import retrofit2.Response;
import retrofit2.Retrofit;
import retrofit2.converter.moshi.MoshiConverterFactory;

import java.io.EOFException;
import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

/**
 * Implementation of a InluxDB API.
 *
 * @author stefan.majer [at] gmail.com
 */
public class InfluxDBImpl implements InfluxDB {

  static final okhttp3.MediaType MEDIA_TYPE_STRING = MediaType.parse("text/plain");

  private static final String SHOW_DATABASE_COMMAND_ENCODED = Query.encode("SHOW DATABASES");

  /**
   * This static constant holds the http logging log level expected in DEBUG mode
   * It is set by System property {@code org.influxdb.InfluxDB.logLevel}.
   *
   * @see org.influxdb.impl.LOG_LEVEL_PROPERTY
   */
  private static final LogLevel LOG_LEVEL = LogLevel.parseLogLevel(System.getProperty(LOG_LEVEL_PROPERTY));

  private final InetAddress hostAddress;
  private final String username;
  private final String password;
  private String version;
  private final Retrofit retrofit;
  private final InfluxDBService influxDBService;
  private BatchProcessor batchProcessor;
  private final AtomicBoolean batchEnabled = new AtomicBoolean(false);
  private final LongAdder writeCount = new LongAdder();
  private final LongAdder unBatchedCount = new LongAdder();
  private final LongAdder batchedCount = new LongAdder();
  private volatile DatagramSocket datagramSocket;
  private final HttpLoggingInterceptor loggingInterceptor;
  private final GzipRequestInterceptor gzipRequestInterceptor;
  private LogLevel logLevel = LogLevel.NONE;
  private JsonAdapter<QueryResult> adapter;
  private String database;
  private String retentionPolicy = "autogen";
  private ConsistencyLevel consistency = ConsistencyLevel.ONE;

  public InfluxDBImpl(final String url, final String username, final String password,
      final OkHttpClient.Builder client) {
    super();
    Moshi moshi = new Moshi.Builder().build();
    this.hostAddress = parseHostAddress(url);
    this.username = username;
    this.password = password;

    this.loggingInterceptor = new HttpLoggingInterceptor();
    setLogLevel(LOG_LEVEL);

    this.gzipRequestInterceptor = new GzipRequestInterceptor();
    this.retrofit = new Retrofit.Builder()
        .baseUrl(url)
        .client(client.addInterceptor(loggingInterceptor).addInterceptor(gzipRequestInterceptor).build())
        .addConverterFactory(MoshiConverterFactory.create())
        .build();
    this.influxDBService = this.retrofit.create(InfluxDBService.class);
    this.adapter = moshi.adapter(QueryResult.class);
  }

    InfluxDBImpl(final String url, final String username, final String password, final OkHttpClient.Builder client,
            final InfluxDBService influxDBService, final JsonAdapter<QueryResult> adapter) {
        super();
        this.hostAddress = parseHostAddress(url);
        this.username = username;
        this.password = password;

        this.loggingInterceptor = new HttpLoggingInterceptor();
        setLogLevel(LOG_LEVEL);

        this.gzipRequestInterceptor = new GzipRequestInterceptor();
        this.retrofit = new Retrofit.Builder()
                .baseUrl(url)
                .client(client.addInterceptor(loggingInterceptor).addInterceptor(gzipRequestInterceptor).build())
                .addConverterFactory(MoshiConverterFactory.create())
                .build();
        this.influxDBService = influxDBService;
        this.adapter = adapter;
    }

  public InfluxDBImpl(final String url, final String username, final String password,
                      final OkHttpClient.Builder client, final String database,
                      final String retentionPolicy, final ConsistencyLevel consistency) {
    this(url, username, password, client);

    setConsistency(consistency);
    setDatabase(database);
    setRetentionPolicy(retentionPolicy);
  }

  private InetAddress parseHostAddress(final String url) {
      HttpUrl httpUrl = HttpUrl.parse(url);

      if (httpUrl == null) {
          throw new IllegalArgumentException("Unable to parse url: " + url);
      }

      try {
          return InetAddress.getByName(httpUrl.host());
      } catch (UnknownHostException e) {
          throw new InfluxDBIOException(e);
      }
  }

  @Override
  public InfluxDB setLogLevel(final LogLevel logLevel) {
    switch (logLevel) {
    case NONE:
      this.loggingInterceptor.setLevel(Level.NONE);
      break;
    case BASIC:
      this.loggingInterceptor.setLevel(Level.BASIC);
      break;
    case HEADERS:
      this.loggingInterceptor.setLevel(Level.HEADERS);
      break;
    case FULL:
      this.loggingInterceptor.setLevel(Level.BODY);
      break;
    default:
      break;
    }
    this.logLevel = logLevel;
    return this;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public InfluxDB enableGzip() {
    this.gzipRequestInterceptor.enable();
    return this;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public InfluxDB disableGzip() {
    this.gzipRequestInterceptor.disable();
    return this;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public boolean isGzipEnabled() {
    return this.gzipRequestInterceptor.isEnabled();
  }

  @Override
  public InfluxDB enableBatch() {
    enableBatch(BatchOptions.DEFAULTS);
    return this;
  }

  @Override
  public InfluxDB enableBatch(final BatchOptions batchOptions) {

    if (this.batchEnabled.get()) {
      throw new IllegalStateException("BatchProcessing is already enabled.");
    }
    this.batchProcessor = BatchProcessor
            .builder(this)
            .actions(batchOptions.getActions())
            .exceptionHandler(batchOptions.getExceptionHandler())
            .interval(batchOptions.getFlushDuration(), batchOptions.getJitterDuration(), TimeUnit.MILLISECONDS)
            .threadFactory(batchOptions.getThreadFactory())
            .bufferLimit(batchOptions.getBufferLimit())
            .consistencyLevel(batchOptions.getConsistency())
            .build();
    this.batchEnabled.set(true);
    return this;
  }

  @Override
  public InfluxDB enableBatch(final int actions, final int flushDuration,
                              final TimeUnit flushDurationTimeUnit) {
    enableBatch(actions, flushDuration, flushDurationTimeUnit, Executors.defaultThreadFactory());
    return this;
  }

  @Override
  public InfluxDB enableBatch(final int actions, final int flushDuration,
                              final TimeUnit flushDurationTimeUnit, final ThreadFactory threadFactory) {
    enableBatch(actions, flushDuration, flushDurationTimeUnit, threadFactory, (points, throwable) -> { });
    return this;
  }

  @Override
  public InfluxDB enableBatch(final int actions, final int flushDuration, final TimeUnit flushDurationTimeUnit,
                              final ThreadFactory threadFactory,
                              final BiConsumer<Iterable<Point>, Throwable> exceptionHandler,
                              final ConsistencyLevel consistency) {
    enableBatch(actions, flushDuration, flushDurationTimeUnit, threadFactory, exceptionHandler)
        .setConsistency(consistency);
    return this;
  }

  @Override
  public InfluxDB enableBatch(final int actions, final int flushDuration, final TimeUnit flushDurationTimeUnit,
                              final ThreadFactory threadFactory,
                              final BiConsumer<Iterable<Point>, Throwable> exceptionHandler) {
    enableBatch(actions, flushDuration, 0, flushDurationTimeUnit, threadFactory, exceptionHandler);
    return this;
  }

  private InfluxDB enableBatch(final int actions, final int flushDuration, final int jitterDuration,
                               final TimeUnit durationTimeUnit, final ThreadFactory threadFactory,
                               final BiConsumer<Iterable<Point>, Throwable> exceptionHandler) {
    if (this.batchEnabled.get()) {
      throw new IllegalStateException("BatchProcessing is already enabled.");
    }
    this.batchProcessor = BatchProcessor
            .builder(this)
            .actions(actions)
            .exceptionHandler(exceptionHandler)
            .interval(flushDuration, jitterDuration, durationTimeUnit)
            .threadFactory(threadFactory)
            .consistencyLevel(consistency)
            .build();
    this.batchEnabled.set(true);
    return this;
  }

  @Override
  public void disableBatch() {
    this.batchEnabled.set(false);
    if (this.batchProcessor != null) {
      this.batchProcessor.flushAndShutdown();
    }
  }

  @Override
  public boolean isBatchEnabled() {
    return this.batchEnabled.get();
  }

  @Override
  public Pong ping() {
    final long started = System.currentTimeMillis();
    Call<ResponseBody> call = this.influxDBService.ping();
    try {
      Response<ResponseBody> response = call.execute();
      Headers headers = response.headers();
      String version = "unknown";
      for (String name : headers.toMultimap().keySet()) {
        if (null != name && "X-Influxdb-Version".equalsIgnoreCase(name)) {
          version = headers.get(name);
          break;
        }
      }
      Pong pong = new Pong();
      pong.setVersion(version);
      pong.setResponseTime(System.currentTimeMillis() - started);
      return pong;
    } catch (IOException e) {
      throw new InfluxDBIOException(e);
    }
  }

  @Override
  public String version() {
    if (version == null) {
      this.version = ping().getVersion();
    }
    return this.version;
  }

  @Override
  public void write(final Point point) {
{'fraction_non_alphanumeric': 0.08862662899845074, 'fraction_numerical': 0.001366991706916978, 'mean_word_length': 3.3920352211326796, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 25, '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': '4663187', 'n_tokens_mistral': 6143, 'n_tokens_neox': 6006, 'n_words': 1359}