package io.vertx.ext.asyncsql.impl;

import com.github.mauricio.async.db.QueryResult;
import com.github.mauricio.async.db.ResultSet;
import com.github.mauricio.async.db.RowData;
import io.vertx.core.AsyncResult;
import io.vertx.core.Future;
import io.vertx.core.Handler;
import io.vertx.core.json.JsonArray;
import io.vertx.ext.sql.SQLRowStream;
import scala.Option;
import scala.collection.Iterator;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

class AsyncSQLRowStream implements SQLRowStream {

  private final ResultSet rs;
  private final Iterator<RowData> cursor;
  private List<String> columns;

  private final AtomicBoolean paused = new AtomicBoolean(false);
  private final AtomicBoolean ended = new AtomicBoolean(false);

  private Handler<JsonArray> handler;
  private Handler<Void> endHandler;
  private Handler<Void> rsClosedHandler;

  AsyncSQLRowStream(QueryResult qr) {
    final Option<ResultSet> rows = qr.rows();
    if (rows.isDefined()) {
      rs = rows.get();
      cursor = rs.iterator();
    } else {
      rs = null;
      cursor = null;
    }

    paused.set(true);
  }

  @Override
  public int column(String name) {
    if (rs == null) {
      throw new IndexOutOfBoundsException("'" + name + "' not found");
    }

    // the columns value will be cached
    return columns().indexOf(name);
  }

  @Override
  public List<String> columns() {
    // populate the cache
    if (columns == null) {
      // quick escape
      if (rs == null) {
        return Collections.emptyList();
      }
      // this list is always read only
      columns = Collections.unmodifiableList(ScalaUtils.toJavaList(rs.columnNames().toList()));
    }
    return columns;
  }

  @Override
  public SQLRowStream exceptionHandler(Handler<Throwable> handler) {
    return this;
  }

  @Override
  public SQLRowStream handler(Handler<JsonArray> handler) {
    this.handler = handler;
    // start pumping data once the handler is set
    resume();
    return this;
  }

  @Override
  public SQLRowStream pause() {
    paused.compareAndSet(false, true);
    return this;
  }

  @Override
  public SQLRowStream resume() {
    if (paused.compareAndSet(true, false)) {
      nextRow();
    }
    return this;
  }

  private void nextRow() {
    while (!paused.get()) {
      if (cursor.hasNext()) {
        handler.handle(ScalaUtils.rowToJsonArray(cursor.next()));
      } else {
        // mark as ended if the handler was registered too late
        ended.set(true);
        // automatically close resources
        if (rsClosedHandler != null) {
          // only notify (since the rs is closed by the underlying driver)
          rsClosedHandler.handle(null);
        } else {
          // default behavior notify that everything ended
          close(c -> {
            if (endHandler != null) {
              endHandler.handle(null);
            }
          });
        }
        break;
      }
    }
  }

  @Override
  public SQLRowStream endHandler(Handler<Void> handler) {
    this.endHandler = handler;
    // registration was late but we're already ended, notify
    if (ended.compareAndSet(true, false)) {
      // only notify once
      endHandler.handle(null);
    }
    return this;
  }

  @Override
  public SQLRowStream resultSetClosedHandler(Handler<Void> handler) {
    this.rsClosedHandler = handler;
    // registration was late but we're already ended, notify
    if (ended.compareAndSet(true, false)) {
      // only notify once
      rsClosedHandler.handle(null);
    }
    return this;
  }

  @Override
  public void moreResults() {
    // NO-OP since the underlying driver only returns 1 ResultSet
  }

  @Override
  public void close() {
    close(null);
  }

  @Override
  public void close(Handler<AsyncResult<Void>> handler) {
    // make sure we stop pumping data
    pause();
    // call the provided handler
    if (handler != null) {
      handler.handle(Future.succeededFuture());
    }
  }
}

@@ -669,61 +669,24 @@
 ate 
-final AtomicBoolean paused = new AtomicBoolean(false)
+long demand = 0L
 ;%0A  
@@ -1089,31 +1089,8 @@
    %7D
-%0A%0A    paused.set(true);
 %0A  %7D
@@ -2019,41 +2019,19 @@
     
-paused.compareAndSet(false, true)
+demand = 0L
 ;%0A  
@@ -2075,115 +2075,278 @@
 lic 
-SQLRowStream resume() %7B%0A    if (paused.compareAndSet(true, false)) %7B%0A      nextRow();%0A    %7D%0A    return this
+synchronized SQLRowStream fetch(long amount) %7B%0A    if (amount %3E 0L) %7B%0A      if ((demand += amount) %3C 0L) %7B%0A        demand = Long.MAX_VALUE;%0A      %7D%0A      nextRow();%0A    %7D%0A    return this;%0A  %7D%0A%0A  @Override%0A  public SQLRowStream resume() %7B%0A    return fetch(Long.MAX_VALUE)
 ;%0A  
@@ -2390,21 +2390,19 @@
 le (
-!paused.get()
+demand %3E 0L
 ) %7B%0A
@@ -2427,24 +2427,94 @@
 asNext()) %7B%0A
+        if (demand != Long.MAX_VALUE) %7B%0A          demand--;%0A        %7D%0A
         hand

Implement read stream fetch