/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 org.apache.hadoop.hbase.client;

import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.coprocessor.Batch;
import org.apache.yetus.audience.InterfaceAudience;
import org.apache.yetus.audience.InterfaceStability;

/**
 * Contains the attributes of a task which will be executed by
 * {@link org.apache.hadoop.hbase.client.AsyncProcess}. The attributes will be validated by
 * AsyncProcess. It's intended for advanced client applications.
 * @param <T> The type of response from server-side
 */
@InterfaceAudience.Private
@InterfaceStability.Evolving
public class AsyncProcessTask<T> {
  /**
   * The number of processed rows. The AsyncProcess has traffic control which may reject some rows.
   */
  public enum SubmittedRows {
    ALL,
    AT_LEAST_ONE,
    NORMAL
  }

  public static <T> Builder<T> newBuilder(final Batch.Callback<T> callback) {
    return new Builder<>(callback);
  }

  public static Builder newBuilder() {
    return new Builder();
  }

  public static class Builder<T> {

    private ExecutorService pool;
    private TableName tableName;
    private RowAccess<? extends Row> rows;
    private SubmittedRows submittedRows = SubmittedRows.ALL;
    private Batch.Callback<T> callback;
    private boolean needResults;
    private int rpcTimeout;
    private int operationTimeout;
    private CancellableRegionServerCallable callable;
    private Object[] results;
    private Map<String, byte[]> requestAttributes = Collections.emptyMap();

    private Builder() {
    }

    private Builder(Batch.Callback<T> callback) {
      this.callback = callback;
    }

    Builder<T> setResults(Object[] results) {
      this.results = results;
      if (results != null && results.length != 0) {
        setNeedResults(true);
      }
      return this;
    }

    public Builder<T> setPool(ExecutorService pool) {
      this.pool = pool;
      return this;
    }

    public Builder<T> setRpcTimeout(int rpcTimeout) {
      this.rpcTimeout = rpcTimeout;
      return this;
    }

    public Builder<T> setOperationTimeout(int operationTimeout) {
      this.operationTimeout = operationTimeout;
      return this;
    }

    public Builder<T> setTableName(TableName tableName) {
      this.tableName = tableName;
      return this;
    }

    public Builder<T> setRowAccess(List<? extends Row> rows) {
      this.rows = new ListRowAccess<>(rows);
      return this;
    }

    public Builder<T> setRowAccess(RowAccess<? extends Row> rows) {
      this.rows = rows;
      return this;
    }

    public Builder<T> setSubmittedRows(SubmittedRows submittedRows) {
      this.submittedRows = submittedRows;
      return this;
    }

    public Builder<T> setNeedResults(boolean needResults) {
      this.needResults = needResults;
      return this;
    }

    Builder<T> setCallable(CancellableRegionServerCallable callable) {
      this.callable = callable;
      return this;
    }

    Builder<T> setRequestAttributes(Map<String, byte[]> requestAttributes) {
      this.requestAttributes = requestAttributes;
      return this;
    }

    public AsyncProcessTask<T> build() {
      return new AsyncProcessTask<>(pool, tableName, rows, submittedRows, callback, callable,
        needResults, rpcTimeout, operationTimeout, results, requestAttributes);
    }
  }

  private final ExecutorService pool;
  private final TableName tableName;
  private final RowAccess<? extends Row> rows;
  private final SubmittedRows submittedRows;
  private final Batch.Callback<T> callback;
  private final CancellableRegionServerCallable callable;
  private final boolean needResults;
  private final int rpcTimeout;
  private final int operationTimeout;
  private final Object[] results;
  private final Map<String, byte[]> requestAttributes;

  AsyncProcessTask(AsyncProcessTask<T> task) {
    this(task.getPool(), task.getTableName(), task.getRowAccess(), task.getSubmittedRows(),
      task.getCallback(), task.getCallable(), task.getNeedResults(), task.getRpcTimeout(),
      task.getOperationTimeout(), task.getResults(), task.getRequestAttributes());
  }

  AsyncProcessTask(ExecutorService pool, TableName tableName, RowAccess<? extends Row> rows,
    SubmittedRows size, Batch.Callback<T> callback, CancellableRegionServerCallable callable,
    boolean needResults, int rpcTimeout, int operationTimeout, Object[] results,
    Map<String, byte[]> requestAttributes) {
    this.pool = pool;
    this.tableName = tableName;
    this.rows = rows;
    this.submittedRows = size;
    this.callback = callback;
    this.callable = callable;
    this.needResults = needResults;
    this.rpcTimeout = rpcTimeout;
    this.operationTimeout = operationTimeout;
    this.results = results;
    this.requestAttributes = requestAttributes;
  }

  public int getOperationTimeout() {
    return operationTimeout;
  }

  public ExecutorService getPool() {
    return pool;
  }

  public TableName getTableName() {
    return tableName;
  }

  public RowAccess<? extends Row> getRowAccess() {
    return rows;
  }

  public SubmittedRows getSubmittedRows() {
    return submittedRows;
  }

  public Batch.Callback<T> getCallback() {
    return callback;
  }

  CancellableRegionServerCallable getCallable() {
    return callable;
  }

  public Map<String, byte[]> getRequestAttributes() {
    return requestAttributes;
  }

  Object[] getResults() {
    return results;
  }

  public boolean getNeedResults() {
    return needResults;
  }

  public int getRpcTimeout() {
    return rpcTimeout;
  }

  static class ListRowAccess<T> implements RowAccess<T> {

    private final List<T> data;

    ListRowAccess(final List<T> data) {
      this.data = data;
    }

    @Override
    public int size() {
      return data.size();
    }

    @Override
    public boolean isEmpty() {
      return data.isEmpty();
    }

    @Override
    public Iterator<T> iterator() {
      return data.iterator();
    }
  }
}