package org.influxdb.impl;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.influxdb.InfluxDB;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;

import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;

/**
 * A BatchProcessor can be attached to a InfluxDB Instance to collect single point writes and
 * aggregates them to BatchPoints to get a better write performance.
 *
 * @author stefan.majer [at] gmail.com
 *
 */
public class BatchProcessor {

  private static final Logger LOG = Logger.getLogger(BatchProcessor.class.getName());
  protected final BlockingQueue<BatchEntry> queue = new LinkedBlockingQueue<>();
  private final ScheduledExecutorService scheduler;
  final InfluxDBImpl influxDB;
  final int actions;
  private final TimeUnit flushIntervalUnit;
  private final int flushInterval;

  /**
   * The Builder to create a BatchProcessor instance.
   */
  public static final class Builder {
    private final InfluxDBImpl influxDB;
    private ThreadFactory threadFactory = Executors.defaultThreadFactory();
    private int actions;
    private TimeUnit flushIntervalUnit;
    private int flushInterval;

    /**
     * @param threadFactory
     *            is optional.
     */
    public Builder threadFactory(final ThreadFactory threadFactory) {
      this.threadFactory = threadFactory;
      return this;
    }

    /**
     * @param influxDB
     *            is mandatory.
     */
    public Builder(final InfluxDB influxDB) {
      this.influxDB = (InfluxDBImpl) influxDB;
    }

    /**
     * The number of actions after which a batchwrite must be performed.
     *
     * @param maxActions
     *            number of Points written after which a write must happen.
     * @return this Builder to use it fluent
     */
    public Builder actions(final int maxActions) {
      this.actions = maxActions;
      return this;
    }

    /**
     * The interval at which at least should issued a write.
     *
     * @param interval
     *            the interval
     * @param unit
     *            the TimeUnit of the interval
     *
     * @return this Builder to use it fluent
     */
    public Builder interval(final int interval, final TimeUnit unit) {
      this.flushInterval = interval;
      this.flushIntervalUnit = unit;
      return this;
    }

    /**
     * Create the BatchProcessor.
     *
     * @return the BatchProcessor instance.
     */
    public BatchProcessor build() {
      Preconditions.checkNotNull(this.influxDB, "influxDB may not be null");
      Preconditions.checkArgument(this.actions > 0, "actions should > 0");
      Preconditions.checkArgument(this.flushInterval > 0, "flushInterval should > 0");
      Preconditions.checkNotNull(this.flushIntervalUnit, "flushIntervalUnit may not be null");
      Preconditions.checkNotNull(this.threadFactory, "threadFactory may not be null");
      return new BatchProcessor(this.influxDB, this.threadFactory, this.actions, this.flushIntervalUnit,
                                this.flushInterval);
    }
  }

  static class BatchEntry {
    private final Point point;
    private final String db;
    private final String rp;

    public BatchEntry(final Point point, final String db, final String rp) {
      super();
      this.point = point;
      this.db = db;
      this.rp = rp;
    }

    public Point getPoint() {
      return this.point;
    }

    public String getDb() {
      return this.db;
    }

    public String getRp() {
      return this.rp;
    }
  }

  /**
   * Static method to create the Builder for this BatchProcessor.
   *
   * @param influxDB
   *            the influxdb database handle.
   * @return the Builder to create the BatchProcessor.
   */
  public static Builder builder(final InfluxDB influxDB) {
    return new Builder(influxDB);
  }

  BatchProcessor(final InfluxDBImpl influxDB, final ThreadFactory threadFactory, final int actions,
                 final TimeUnit flushIntervalUnit, final int flushInterval) {
    super();
    this.influxDB = influxDB;
    this.actions = actions;
    this.flushIntervalUnit = flushIntervalUnit;
    this.flushInterval = flushInterval;
    this.scheduler = Executors.newSingleThreadScheduledExecutor(threadFactory);
    // Flush at specified Rate
    this.scheduler.scheduleAtFixedRate(new Runnable() {
      @Override
      public void run() {
        write();
      }
    }, this.flushInterval, this.flushInterval, this.flushIntervalUnit);

  }

  void write() {
    try {
      if (this.queue.isEmpty()) {
        return;
      }

      Map<String, BatchPoints> databaseToBatchPoints = Maps.newHashMap();
      List<BatchEntry> batchEntries = new ArrayList<>(this.queue.size());
      this.queue.drainTo(batchEntries);

      for (BatchEntry batchEntry : batchEntries) {
        String dbName = batchEntry.getDb();
        if (!databaseToBatchPoints.containsKey(dbName)) {
          BatchPoints batchPoints = BatchPoints.database(dbName)
                                               .retentionPolicy(batchEntry.getRp()).build();
          databaseToBatchPoints.put(dbName, batchPoints);
        }
        Point point = batchEntry.getPoint();
        databaseToBatchPoints.get(dbName).point(point);
      }

      for (BatchPoints batchPoints : databaseToBatchPoints.values()) {
        BatchProcessor.this.influxDB.write(batchPoints);
      }
    } catch (Throwable t) {
      // any exception wouldn't stop the scheduler
      LOG.log(Level.SEVERE, "Batch could not be sent. Data will be lost", t);
    }
  }

  /**
   * Put a single BatchEntry to the cache for later processing.
   *
   * @param batchEntry
   *            the batchEntry to write to the cache.
   */
  void put(final BatchEntry batchEntry) {
    this.queue.add(batchEntry);
    if (this.queue.size() >= this.actions) {
      write();
    }
  }

  /**
   * Flush the current open writes to influxdb and end stop the reaper thread. This should only be
   * called if no batch processing is needed anymore.
   *
   */
  void flush() {
    this.write();
    this.scheduler.shutdown();
  }

}

@@ -6178,30 +6178,145 @@
 ns) %7B%0A      
-write(
+this.scheduler.submit(new Runnable() %7B%0A        @Override%0A        public void run() %7B%0A          write();%0A        %7D%0A      %7D
 );%0A    %7D%0A  %7D

Put should always be async if batch is enabled, closes #147