/*
 * 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 com.datastax.driver.core;

import static com.datastax.driver.core.SchemaElement.KEYSPACE;

import com.datastax.driver.core.exceptions.ConnectionException;
import com.datastax.driver.core.exceptions.DriverInternalError;
import com.datastax.driver.core.exceptions.NoHostAvailableException;
import com.datastax.driver.core.exceptions.OperationTimedOutException;
import com.datastax.driver.core.exceptions.QueryExecutionException;
import com.datastax.driver.core.exceptions.QueryValidationException;
import com.google.common.util.concurrent.AbstractFuture;
import com.google.common.util.concurrent.Uninterruptibles;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/** Internal implementation of ResultSetFuture. */
class DefaultResultSetFuture extends AbstractFuture<ResultSet>
    implements ResultSetFuture, RequestHandler.Callback {

  private static final Logger logger = LoggerFactory.getLogger(ResultSetFuture.class);

  private final SessionManager session;
  private final ProtocolVersion protocolVersion;
  private final Message.Request request;
  private volatile RequestHandler handler;

  DefaultResultSetFuture(
      SessionManager session, ProtocolVersion protocolVersion, Message.Request request) {
    this.session = session;
    this.protocolVersion = protocolVersion;
    this.request = request;
  }

  @Override
  public void register(RequestHandler handler) {
    this.handler = handler;
  }

  @Override
  public Message.Request request() {
    return request;
  }

  @Override
  public void onSet(
      Connection connection,
      Message.Response response,
      ExecutionInfo info,
      Statement statement,
      long latency) {
    try {
      switch (response.type) {
        case RESULT:
          Responses.Result rm = (Responses.Result) response;
          switch (rm.kind) {
            case SET_KEYSPACE:
              // propagate the keyspace change to other connections
              session.poolsState.setKeyspace(((Responses.Result.SetKeyspace) rm).keyspace);
              set(ArrayBackedResultSet.fromMessage(rm, session, protocolVersion, info, statement));
              break;
            case SCHEMA_CHANGE:
              ResultSet rs =
                  ArrayBackedResultSet.fromMessage(rm, session, protocolVersion, info, statement);
              final Cluster.Manager cluster = session.cluster.manager;
              if (!cluster.configuration.getQueryOptions().isMetadataEnabled()) {
                cluster.waitForSchemaAgreementAndSignal(connection, this, rs);
              } else {
                Responses.Result.SchemaChange scc = (Responses.Result.SchemaChange) rm;
                switch (scc.change) {
                  case CREATED:
                  case UPDATED:
                    cluster.refreshSchemaAndSignal(
                        connection,
                        this,
                        rs,
                        scc.targetType,
                        scc.targetKeyspace,
                        scc.targetName,
                        scc.targetSignature);
                    break;
                  case DROPPED:
                    if (scc.targetType == KEYSPACE) {
                      // If that the one keyspace we are logged in, reset to null (it shouldn't
                      // really happen but ...)
                      // Note: Actually, Cassandra doesn't do that so we don't either as this could
                      // confuse prepared statements.
                      // We'll add it back if CASSANDRA-5358 changes that behavior
                      // if (scc.keyspace.equals(session.poolsState.keyspace))
                      //    session.poolsState.setKeyspace(null);
                      final KeyspaceMetadata removedKeyspace =
                          cluster.metadata.removeKeyspace(scc.targetKeyspace);
                      if (removedKeyspace != null) {
                        cluster.executor.submit(
                            new Runnable() {
                              @Override
                              public void run() {
                                cluster.metadata.triggerOnKeyspaceRemoved(removedKeyspace);
                              }
                            });
                      }
                    } else {
                      KeyspaceMetadata keyspace =
                          session.cluster.manager.metadata.keyspaces.get(scc.targetKeyspace);
                      if (keyspace == null) {
                        logger.warn(
                            "Received a DROPPED notification for {} {}.{}, but this keyspace is unknown in our metadata",
                            scc.targetType,
                            scc.targetKeyspace,
                            scc.targetName);
                      } else {
                        switch (scc.targetType) {
                          case TABLE:
                            // we can't tell whether it's a table or a view,
                            // but since two objects cannot have the same name,
                            // try removing both
                            final TableMetadata removedTable = keyspace.removeTable(scc.targetName);
                            if (removedTable != null) {
                              cluster.executor.submit(
                                  new Runnable() {
                                    @Override
                                    public void run() {
                                      cluster.metadata.triggerOnTableRemoved(removedTable);
                                    }
                                  });
                            } else {
                              final MaterializedViewMetadata removedView =
                                  keyspace.removeMaterializedView(scc.targetName);
                              if (removedView != null) {
                                cluster.executor.submit(
                                    new Runnable() {
                                      @Override
                                      public void run() {
                                        cluster.metadata.triggerOnMaterializedViewRemoved(
                                            removedView);
                                      }
                                    });
                              }
                            }
                            break;
                          case TYPE:
                            final UserType removedType = keyspace.removeUserType(scc.targetName);
                            if (removedType != null) {
                              cluster.executor.submit(
                                  new Runnable() {
                                    @Override
                                    public void run() {
                                      cluster.metadata.triggerOnUserTypeRemoved(removedType);
                                    }
                                  });
                            }
                            break;
                          case FUNCTION:
                            final FunctionMetadata removedFunction =
                                keyspace.removeFunction(
                                    Metadata.fullFunctionName(scc.targetName, scc.targetSignature));
                            if (removedFunction != null) {
                              cluster.executor.submit(
                                  new Runnable() {
                                    @Override
                                    public void run() {
                                      cluster.metadata.triggerOnFunctionRemoved(removedFunction);
                                    }
                                  });
                            }
                            break;
                          case AGGREGATE:
                            final AggregateMetadata removedAggregate =
                                keyspace.removeAggregate(
                                    Metadata.fullFunctionName(scc.targetName, scc.targetSignature));
                            if (removedAggregate != null) {
                              cluster.executor.submit(
                                  new Runnable() {
                                    @Override
                                    public void run() {
                                      cluster.metadata.triggerOnAggregateRemoved(removedAggregate);
                                    }
                                  });
                            }
                            break;
                        }
                      }
                    }
                    session.cluster.manager.waitForSchemaAgreementAndSignal(connection, this, rs);
                    break;
                  default:
                    logger.info("Ignoring unknown schema change result");
                    break;
                }
              }
              break;
            default:
              set(ArrayBackedResultSet.fromMessage(rm, session, protocolVersion, info, statement));
              break;
          }
          break;
        case ERROR:
          setException(((Responses.Error) response).asException(connection.endPoint));
          break;
        default:
          // This mean we have probably have a bad node, so defunct the connection
          connection.defunct(
              new ConnectionException(
                  connection.endPoint, String.format("Got unexpected %s response", response.type)));
          setException(
              new DriverInternalError(
                  String.format(
                      "Got unexpected %s response from %s", response.type, connection.endPoint)));
          break;
      }
    } catch (Throwable e) {
      // If we get a bug here, the client will not get it, so better forwarding the error
      setException(
          new DriverInternalError(
              "Unexpected error while processing response from " + connection.endPoint, e));
    }
  }

  @Override
  public void onSet(
      Connection connection, Message.Response response, long latency, int retryCount) {
    // This is only called for internal calls (i.e, when the callback is not wrapped in
    // ResponseHandler),
    // so don't bother with ExecutionInfo.
    onSet(connection, response, null, null, latency);
  }

  @Override
  public void onException(
      Connection connection, Exception exception, long latency, int retryCount) {
    setException(exception);
  }

  @Override
  public boolean onTimeout(Connection connection, long latency, int retryCount) {
    // This is only called for internal calls (i.e, when the future is not wrapped in
    // RequestHandler).
    // So just set an exception for the final result, which should be handled correctly by said
    // internal call.
    setException(new OperationTimedOutException(connection.endPoint));
    return true;
  }

  // We sometimes need (in the driver) to set the future from outside this class,
  // but AbstractFuture#set is protected so this method. We don't want it public
  // however, no particular reason to give users rope to hang themselves.
  void setResult(ResultSet rs) {
    set(rs);
  }

  /**
   * Waits for the query to return and return its result.
   *
   * <p>This method is usually more convenient than {@link #get} because it:
   *
   * <ul>
   *   <li>Waits for the result uninterruptibly, and so doesn't throw {@link InterruptedException}.
   *   <li>Returns meaningful exceptions, instead of having to deal with ExecutionException.
   * </ul>
   *
   * As such, it is the preferred way to get the future result.
   *
   * @throws NoHostAvailableException if no host in the cluster can be contacted successfully to
   *     execute this query.
   * @throws QueryExecutionException if the query triggered an execution exception, that is an
   *     exception thrown by Cassandra when it cannot execute the query with the requested
   *     consistency level successfully.
   * @throws QueryValidationException if the query is invalid (syntax error, unauthorized or any
   *     other validation problem).
   */
  @Override
  public ResultSet getUninterruptibly() {
    try {
      return Uninterruptibles.getUninterruptibly(this);
    } catch (ExecutionException e) {
      throw DriverThrowables.propagateCause(e);
    }
  }

  /**
   * Waits for the provided time for the query to return and return its result if available.
   *
   * <p>This method is usually more convenient than {@link #get} because it:
   *
   * <ul>
   *   <li>Waits for the result uninterruptibly, and so doesn't throw {@link InterruptedException}.
   *   <li>Returns meaningful exceptions, instead of having to deal with ExecutionException.
   * </ul>
   *
   * As such, it is the preferred way to get the future result.
   *
   * @throws NoHostAvailableException if no host in the cluster can be contacted successfully to
   *     execute this query.
   * @throws QueryExecutionException if the query triggered an execution exception, that is an
   *     exception thrown by Cassandra when it cannot execute the query with the requested
   *     consistency level successfully.
   * @throws QueryValidationException if the query if invalid (syntax error, unauthorized or any
   *     other validation problem).
   * @throws TimeoutException if the wait timed out (Note that this is different from a Cassandra
   *     timeout, which is a {@code QueryExecutionException}).
   */
  @Override
  public ResultSet getUninterruptibly(long timeout, TimeUnit unit) throws TimeoutException {
    try {
      return Uninterruptibles.getUninterruptibly(this, timeout, unit);
    } catch (ExecutionException e) {
      throw DriverThrowables.propagateCause(e);
    }
  }

  /**
   * Attempts to cancel the execution of the request corresponding to this
   * future. This attempt will fail if the request has already returned.
   * <p/>
   * Please note that this only cancels the request driver side, but nothing
   * is done to interrupt the execution of the request Cassandra side (and that even
   * if {@code mayInterruptIfRunning} is true) since  Cassandra does not
   * support such interruption.
   * <p/>
   * This method can be used to ensure no more work is performed driver side
   * (which, while it doesn't include stopping a request already submitted
   * to a Cassandra node, may include not retrying another Cassandra host on
   * failure/timeout) if the ResultSet is not going to be retried. Typically,
   * the code to wait for a request result for a maximum of 1 second could
   * look like:
   * <pre>
   *   ResultSetFuture future = session.executeAsync(...some query...);
   *   try {
   *       ResultSet result = future.get(1, TimeUnit.SECONDS);
   *       ... process result ...
   *   } catch (TimeoutException e) {
   *       future.cancel(true); // Ensure any resource used by this query driver
   *                            // side is released immediately
   *       ... handle timeout ...
   *   }
   * <pre>
   *
   * @param mayInterruptIfRunning the value of this parameter is currently
   *                              ignored.
   * @return {@code false} if the future could not be cancelled (it has already
   * completed normally); {@code true} otherwise.
   */
  @Override
  public boolean cancel(boolean mayInterruptIfRunning) {
    if (!super.cancel(mayInterruptIfRunning)) return false;

    if (handler != null) {
      handler.cancel();
    }
    return true;
  }

  @Override
  public int retryCount() {
    // This is only called for internal calls (i.e, when the future is not wrapped in
    // RequestHandler).
    // There is no retry logic in that case, so the value does not really matter.
    return 0;
  }
}
