/*
 * Copyright (c) 2013 DataTorrent, Inc. ALL Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.datatorrent.contrib.couchdb;

import com.datatorrent.api.annotation.ShipContainingJars;
import com.datatorrent.lib.db.AbstractStoreInputOperator;
import com.google.common.base.Throwables;
import org.ektorp.ViewQuery;
import org.ektorp.ViewResult;

import javax.validation.constraints.Min;
import java.io.IOException;
import java.util.List;

/**
 * Base class for CouchDb input adaptor.<br/>
 * <p>
 * CouchDb filters documents in the database using stored views. Views are referred as design documents.
 * This operator queries the view and emits the view result.
 * </p>
 *
 * <p>
 * Subclasses  of this operator provide the ViewQuery which corresponds to a database view.<br/>
 * In this base implementation, if the ViewQuery doesn't change, then the same view results are emitted
 * at the end of every streaming window.
 * </p>
 *
 * <p>
 * The operator can emit paged results as well. The page size is configured using {@link #pageSize}.<br/>
 * The operator assumes that the ViewQuery implementation by sub-classes does not depend on data outside the document
 * (like the current date) because that will break the caching of a view's result in CouchDb.
 * Also the {@link #getViewQuery()} method should return the same view stored in CouchDb every time.<br/>
 * </p>
 *
 * @param <T>Type of tuples which are generated</T>
 * @since 0.3.5
 */
@ShipContainingJars(classes = {ViewQuery.class})
public abstract class AbstractCouchDBInputOperator<T> extends AbstractStoreInputOperator<T, CouchDbStore>
{

  @Min(0)
  private int pageSize;

  private String nextPageKey = null;
  private boolean started = false;

  @Override
  public void emitTuples()
  {
    if (pageSize == 0) {
      ViewQuery viewQuery = getViewQuery();
      ViewResult result = store.queryStore(viewQuery);
      try {
        for (ViewResult.Row row : result.getRows()) {
          T tuple = getTuple(row);
          outputPort.emit(tuple);
        }
      }
      catch (Throwable cause) {
        Throwables.propagate(cause);
      }
    }
    else {
      if (!started || nextPageKey != null) {
        started = true;
        ViewQuery query = getViewQuery().limit(pageSize + 1);

        if (nextPageKey != null) {
          query.startKey(nextPageKey);
        }
        ViewResult result = store.queryStore(query);
        List<ViewResult.Row> rows = result.getRows();
        List<ViewResult.Row> rowsToEmit = rows;
        if (rows.size() > pageSize) {
          //More pages to fetch. We don't emit the last row as it is a link to next page.
          nextPageKey = rows.get(rows.size() - 1).getKey();
          rowsToEmit = rows.subList(0, rows.size() - 1);
        }
        else {
          //No next page so emit all the rows.
          nextPageKey = null;
        }
        try {
          for (ViewResult.Row row : rowsToEmit) {
            T tuple = getTuple(row);
            outputPort.emit(tuple);
          }
        }
        catch (Throwable cause) {
          Throwables.propagate(cause);
        }
      }
    }
  }

  /**
   * @return view-query that specifies the couch-db view whose results will be fetched.
   */

  public abstract ViewQuery getViewQuery();

  /**
   * This operator fetches result of a view in {@link ViewResult}. Sub-classes should provide the
   * implementation to convert a row of ViewResult to emitted tuple type.
   *
   * @param value a row of ViewResult that should be converted to a tuple.
   * @return emitted tuple.
   */
  public abstract T getTuple(ViewResult.Row value) throws IOException;

  /**
   * Sets the no. of rows in a page.
   *
   * @param pageSize size of a page
   */
  public void setPageSize(int pageSize)
  {
    this.pageSize = pageSize;
  }
}

@@ -670,117 +670,95 @@
 ort 
-com.datatorrent.api.annotation.ShipContainingJars;%0Aimport com.datatorrent.lib.db.AbstractStoreInputOperator
+java.io.IOException;%0Aimport java.util.List;%0A%0Aimport javax.validation.constraints.Min
 ;%0A
+%0A
 impo
@@ -795,16 +795,17 @@
 wables;%0A
+%0A
 import o
@@ -867,91 +867,116 @@
 ort 
-javax.validation.constraints.Min;%0Aimport java.io.IOException;%0Aimport java.util.List
+com.datatorrent.lib.db.AbstractStoreInputOperator;%0A%0Aimport com.datatorrent.api.annotation.ShipContainingJars
 ;%0A%0A/
@@ -1229,17 +1229,16 @@
 classes 
- 
 of this 
@@ -1922,16 +1922,17 @@
 aram %3CT%3E
+ 
 Type of 
@@ -2139,17 +2139,16 @@
 tore%3E%0A%7B%0A
-%0A
   @Min(0
@@ -2173,17 +2173,16 @@
 geSize;%0A
-%0A
   privat
@@ -3749,17 +3749,16 @@
 .%0A   */%0A
-%0A
   public
@@ -4076,24 +4076,50 @@
 tted tuple.%0A
+   * @throws IOException %0A
    */%0A  publ
@@ -4348,14 +4348,15 @@
 geSize;%0A
-
   %7D%0A
+%0A
 %7D%0A

address minor warnings.