/*
 * 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.jdo.tck.api.persistencemanager;

import java.util.List;
import javax.jdo.JDODataStoreException;
import javax.jdo.JDOFatalException;
import javax.jdo.JDOUnsupportedOptionException;
import javax.jdo.PersistenceManager;
import javax.jdo.Query;
import javax.jdo.Transaction;
import junit.framework.AssertionFailedError;
import org.apache.jdo.tck.JDO_Test;
import org.apache.jdo.tck.pc.mylib.PCPoint;
import org.apache.jdo.tck.pc.mylib.PCRect;
import org.apache.jdo.tck.util.BatchTestRunner;
import org.apache.jdo.tck.util.ThreadExceptionHandler;

/**
 * <B>Title:</B> DatastoreTimeout <br>
 * <B>Keywords:</B> datastore timeout <br>
 * <B>Assertion ID:</B> A12.6.9-1, A14.6.1-7 <br>
 * <B>Assertion Description: </B>
 */
public class DatastoreTimeout extends JDO_Test {

  /** */
  private static final String ASSERTION_FAILED =
      "Assertion A12.6.9-1, A14.6.1-7 (DatastoreTimeout) failed: ";

  /** Timeout value for datastore read */
  private static final Integer READ_TIMEOUT = Integer.valueOf(100);

  /** Timeout value for datastore write */
  private static final Integer WRITE_TIMEOUT = Integer.valueOf(100);

  /** Zero Timeout value */
  private static final Integer ZERO_TIMEOUT = Integer.valueOf(0);

  /** Time for the main thread to sleep after starting a parallel thread. */
  private static final int MAIN_SLEEP_MILLIS = 1000;

  /** Time for the parallel threads to sleep before commit. */
  private static final int THREAD_SLEEP_MILLIS_SHORT = 2500;

  /** Time for the parallel threads to sleep before commit. */
  private static final int THREAD_SLEEP_MILLIS_LONG = 7500;

  /** Oid of the PCRect instance created by localSetUp */
  private Object rectOid;

  /** Oid of the first PCPOint instance created by localSetUp */
  private Object point1Oid;

  /** Oid of the second PCPoint instance created by localSetUp */
  private Object point2Oid;

  /**
   * The <code>main</code> is called when the class is directly executed from the command line.
   *
   * @param args The arguments passed to the program.
   */
  public static void main(String[] args) {
    BatchTestRunner.run(DatastoreTimeout.class);
  }

  /**
   * Method testing DatastoreReadTimeout.
   *
   * @throws Exception exception
   */
  public void testDatastoreReadTimeout() throws Exception {

    if (debug) logger.debug("isDatastoreTimeoutSupported:" + isDatastoreTimeoutSupported());

    // Parallel thread writing the instances and causing them to be locked
    ThreadExceptionHandler group = new ThreadExceptionHandler();
    ParallelWriter runnable = new ParallelWriter(THREAD_SLEEP_MILLIS_LONG);
    Thread t = new Thread(group, runnable, "Parallel Writer");
    t.start();

    // Wait for a second such that the other thread can lock the instances
    Thread.sleep(MAIN_SLEEP_MILLIS);

    try {
      runQueryReadingPCPointInstances(READ_TIMEOUT);
      runGetObjectByIdReadingPCPointInstance(READ_TIMEOUT);
      runNavigationalReadPCPointInstance(READ_TIMEOUT);
    } finally {
      t.join();
    }
    Throwable problem = group.getUncaughtException(t);
    if (problem != null) {
      if (problem instanceof AssertionFailedError) throw (AssertionFailedError) problem;
      else throw new JDOFatalException("Thread " + t.getName() + " results in exception ", problem);
    }
  }

  /**
   * Method testing DatastoreWriteTimeout.
   *
   * @throws Exception exception
   */
  public void testDatastoreWriteTimeout() throws Exception {

    if (debug) logger.debug("isDatastoreTimeoutSupported:" + isDatastoreTimeoutSupported());

    // Parallel thread reading the instances and causing them to be locked
    ThreadExceptionHandler group = new ThreadExceptionHandler();
    ParallelReader runnable = new ParallelReader(THREAD_SLEEP_MILLIS_LONG);
    Thread t = new Thread(group, runnable, "Parallel Reader");
    t.start();

    // Wait for a second such that the other thread can lock the instances
    Thread.sleep(MAIN_SLEEP_MILLIS);

    try {
      runUpdatePCointInstance(WRITE_TIMEOUT);
      runDeletePCPointInstance(WRITE_TIMEOUT);
      runDeletePCPointInstancesByQuery(WRITE_TIMEOUT);
    } finally {
      t.join();
    }
    Throwable problem = group.getUncaughtException(t);
    if (problem != null) {
      if (problem instanceof AssertionFailedError) throw (AssertionFailedError) problem;
      else throw new JDOFatalException("Thread " + t.getName() + " results in exception ", problem);
    }
  }

  /**
   * Method testing zero DatastoreReadTimeout.
   *
   * @throws Exception exception
   */
  public void testZeroDatastoreReadTimeout() throws Exception {

    // Parallel thread writing the instances and causing them to be locked
    ThreadExceptionHandler group = new ThreadExceptionHandler();
    ParallelWriter runnable = new ParallelWriter(THREAD_SLEEP_MILLIS_SHORT);
    Thread t = new Thread(group, runnable, "Parallel Writer");
    t.start();

    // Wait for a second such that the other thread can lock the instances
    Thread.sleep(MAIN_SLEEP_MILLIS);

    try {
      runQueryReadingPCPointInstances(ZERO_TIMEOUT);
    } finally {
      t.join();
    }
    Throwable problem = group.getUncaughtException(t);
    if (problem != null) {
      if (problem instanceof AssertionFailedError) throw (AssertionFailedError) problem;
      else throw new JDOFatalException("Thread " + t.getName() + " results in exception ", problem);
    }
  }

  /**
   * Method testing zero DatastoreWriteTimeout.
   *
   * @throws Exception exception
   */
  public void testZeroDatastoreWriteTimeout() throws Exception {

    // Parallel thread reading the instances and causing them to be locked
    ThreadExceptionHandler group = new ThreadExceptionHandler();
    ParallelReader runnable = new ParallelReader(THREAD_SLEEP_MILLIS_SHORT);
    Thread t = new Thread(group, runnable, "Parallel Reader");
    t.start();

    // Wait for a second such that the other thread can lock the instances
    Thread.sleep(MAIN_SLEEP_MILLIS);

    try {
      runUpdatePCointInstance(ZERO_TIMEOUT);
    } finally {
      t.join();
    }
    Throwable problem = group.getUncaughtException(t);
    if (problem != null) {
      if (problem instanceof AssertionFailedError) throw (AssertionFailedError) problem;
      else throw new JDOFatalException("Thread " + t.getName() + " results in exception ", problem);
    }
  }

  /**
   * @see org.apache.jdo.tck.JDO_Test#localSetUp()
   */
  @Override
  protected void localSetUp() {
    addTearDownClass(PCRect.class);
    addTearDownClass(PCPoint.class);

    getPM();
    try {
      Transaction tx = pm.currentTransaction();
      tx.begin();
      PCPoint p1 = new PCPoint(110, 120);
      PCPoint p2 = new PCPoint(130, 140);
      PCRect rect = new PCRect(0, p1, p2);
      pm.makePersistent(rect);
      tx.commit();
      rectOid = pm.getObjectId(rect);
      point1Oid = pm.getObjectId(p1);
      point2Oid = pm.getObjectId(p2);
    } finally {
      cleanupPM();
    }
  }

  // ----------------- Helper methods ---------------------------

  /** DatastoreReadTimeout helper method: query reading PCPoint instance. */
  private void runQueryReadingPCPointInstances(Integer timeout) {
    getPM();
    try {
      pm.currentTransaction().begin();
      Query<PCPoint> q = pm.newQuery(PCPoint.class);
      q.setDatastoreReadTimeoutMillis(timeout);
      @SuppressWarnings("unused")
      List<PCPoint> result = q.executeList();
      pm.currentTransaction().commit();
      if (!isDatastoreTimeoutSupported()) {
        fail(
            ASSERTION_FAILED,
            "Query.setDatastoreReadTimeoutMillis should throw a "
                + "JDOUnsupportedOptionException, if datastore timeout is not supported ");
      }
    } catch (JDOUnsupportedOptionException ex) {
      if (isDatastoreTimeoutSupported()) {
        fail(
            ASSERTION_FAILED,
            "Query.setDatastoreReadTimeoutMillis should not result "
                + "in a JDOUnsupportedOptionException, if datastore timeout is supported ");
      }
    } catch (JDODataStoreException ex) {
      if (!isDatastoreTimeoutSupported()) {
        fail(
            ASSERTION_FAILED,
            "Query.execute should not result in a "
                + "JDODataStoreException, if datastore timeout is not supported.");
      } else if (timeout == ZERO_TIMEOUT) {
        fail(
            ASSERTION_FAILED,
            "Query.execute should not result in a "
                + "JDODataStoreException, if zero timeout is specified (meaning no timeout).");
      }
    } finally {
      cleanupPM();
    }
  }

  /** DatastoreReadTimeout helper method: getObjectById reading PCPoint instance. */
  private void runGetObjectByIdReadingPCPointInstance(Integer timeout) {
    getPM();
    try {
      pm.currentTransaction().begin();
      pm.setDatastoreReadTimeoutMillis(timeout);
      @SuppressWarnings("unused")
      PCPoint point1 = (PCPoint) pm.getObjectById(point1Oid, true);
      pm.currentTransaction().commit();
      if (!isDatastoreTimeoutSupported()) {
        fail(
            ASSERTION_FAILED,
            "PM.setDatastoreReadTimeoutMillis should throw a "
                + "JDOUnsupportedOptionException, if datastore timeout is not supported ");
      }
    } catch (JDOUnsupportedOptionException ex) {
      if (isDatastoreTimeoutSupported()) {
        fail(
            ASSERTION_FAILED,
            "PM.setDatastoreReadTimeoutMillis should not result "
                + "in a JDOUnsupportedOptionException, if datastore timeout is supported ");
      }
    } catch (JDODataStoreException ex) {
      if (!isDatastoreTimeoutSupported()) {
        fail(
            ASSERTION_FAILED,
            "PM.getObjectById should not result in a "
                + "JDODataStoreException, if datastore timeout is not supported.");
      } else if (timeout == ZERO_TIMEOUT) {
        fail(
            ASSERTION_FAILED,
            "PM.getObjectById should not result in a "
                + "JDODataStoreException, if zero timeout is specified (meaning no timeout).");
      }
    } finally {
      cleanupPM();
    }
  }

  /** DatastoreReadTimeout helper method: navigation reading PCPoint instance. */
  private void runNavigationalReadPCPointInstance(Integer timeout) {
    getPM();
    try {
      pm.currentTransaction().begin();
      pm.setDatastoreReadTimeoutMillis(timeout);
      PCRect rect = (PCRect) pm.getObjectById(rectOid, true);
      PCPoint p1 = rect.getUpperLeft();
      p1.name();
      pm.currentTransaction().commit();
      if (!isDatastoreTimeoutSupported()) {
        fail(
            ASSERTION_FAILED,
            "PM.setDatastoreReadTimeoutMillis should throw a "
                + "JDOUnsupportedOptionException, if datastore timeout is not supported ");
      }
    } catch (JDOUnsupportedOptionException ex) {
      if (isDatastoreTimeoutSupported()) {
        fail(
            ASSERTION_FAILED,
            "PM.setDatastoreReadTimeoutMillis should not result "
                + "in a JDOUnsupportedOptionException, if datastore timeout is supported ");
      }
    } catch (JDODataStoreException ex) {
      if (!isDatastoreTimeoutSupported()) {
        fail(
            ASSERTION_FAILED,
            "Navigational access should not result in a "
                + "JDODataStoreException, if datastore timeout is not supported.");
      } else if (timeout == ZERO_TIMEOUT) {
        fail(
            ASSERTION_FAILED,
            "Navigational access should not result in a "
                + "JDODataStoreException, if zero timeout is specified (meaning no timeout).");
      }
    } finally {
      cleanupPM();
    }
  }

  /** DatastoreWriteTimeout helper method: update PCPoint instance. */
  private void runUpdatePCointInstance(Integer timeout) {
    getPM();
    try {
      pm.currentTransaction().begin();
      pm.setDatastoreWriteTimeoutMillis(timeout);
      PCPoint point1 = (PCPoint) pm.getObjectById(point1Oid, true);
      point1.setX(500);
      pm.currentTransaction().commit();
      if (!isDatastoreTimeoutSupported()) {
        fail(
            ASSERTION_FAILED,
            "PM.setDatastoreWriteTimeoutMillis should throw a "
                + "JDOUnsupportedOptionException, if datastore timeout is not supported ");
      }
    } catch (JDOUnsupportedOptionException ex) {
      if (isDatastoreTimeoutSupported()) {
        fail(
            ASSERTION_FAILED,
            "PM.setDatastoreWriteTimeoutMillis should not result "
                + "in a JDOUnsupportedOptionException, if datastore timeout is supported ");
      }
    } catch (JDODataStoreException ex) {
      if (!isDatastoreTimeoutSupported()) {
        fail(
            ASSERTION_FAILED,
            "PM.getObjectById should not result in a "
                + "JDODataStoreException, if datastore timeout is not supported.");
      } else if (timeout == ZERO_TIMEOUT) {
        fail(
            ASSERTION_FAILED,
            "PM.getObjectById should not result in a "
                + "JDODataStoreException, if zero timeout is specified (meaning no timeout).");
      }
    } finally {
      cleanupPM();
    }
  }

  /** DatastoreWriteTimeout helper method: delete PCPoint instance. */
  private void runDeletePCPointInstance(Integer timeout) {
    getPM();
    try {
      pm.currentTransaction().begin();
      pm.setDatastoreWriteTimeoutMillis(timeout);
      PCPoint point1 = (PCPoint) pm.getObjectById(point1Oid, true);
      pm.deletePersistent(point1);
      pm.currentTransaction().commit();
      if (!isDatastoreTimeoutSupported()) {
        fail(
            ASSERTION_FAILED,
            "PM.setDatastoreWriteTimeoutMillis should throw a "
                + "JDOUnsupportedOptionException, if datastore timeout is not supported ");
      }
    } catch (JDOUnsupportedOptionException ex) {
      if (isDatastoreTimeoutSupported()) {
        fail(
            ASSERTION_FAILED,
            "PM.setDatastoreWriteTimeoutMillis should not result "
                + "in a JDOUnsupportedOptionException, if datastore timeout is supported ");
      }
    } catch (JDODataStoreException ex) {
      if (!isDatastoreTimeoutSupported()) {
        fail(
            ASSERTION_FAILED,
            "PM.getObjectById should not result in a "
                + "JDODataStoreException, if datastore timeout is not supported.");
      } else if (timeout == ZERO_TIMEOUT) {
        fail(
            ASSERTION_FAILED,
            "PM.getObjectById should not result in a "
                + "JDODataStoreException, if zero timeout is specified (meaning no timeout).");
      }
    } finally {
      cleanupPM();
    }
  }

  /** DatastoreWriteTimeout helper method: delete by query. */
  private void runDeletePCPointInstancesByQuery(Integer timeout) {
    getPM();
    try {
      pm.currentTransaction().begin();
      Query<PCPoint> q = pm.newQuery(PCPoint.class);
      q.setDatastoreWriteTimeoutMillis(timeout);
      q.deletePersistentAll();
      pm.currentTransaction().commit();
      if (!isDatastoreTimeoutSupported()) {
        fail(
            ASSERTION_FAILED,
            "Query.setDatastoreWriteTimeoutMillis should throw a "
                + "JDOUnsupportedOptionException, if datastore timeout is not supported ");
      }
    } catch (JDOUnsupportedOptionException ex) {
      if (isDatastoreTimeoutSupported()) {
        fail(
            ASSERTION_FAILED,
            "Query.setDatastoreWriteTimeoutMillis should not result "
                + "in a JDOUnsupportedOptionException, if datastore timeout is supported ");
      }
    } catch (JDODataStoreException ex) {
      if (!isDatastoreTimeoutSupported()) {
        fail(
            ASSERTION_FAILED,
            "Query.deletePersistentAll should not result in a "
                + "JDODataStoreException, if datastore timeout is not supported.");
      } else if (timeout == ZERO_TIMEOUT) {
        fail(
            ASSERTION_FAILED,
            "Query.deletePersistentAll should not result in a "
                + "JDODataStoreException, if zero timeout is specified (meaning no timeout).");
      }
    } finally {
      cleanupPM();
    }
  }

  // ------------- Helper classes -------------------------------

  /** Runnable class updating instances and causing them to be read locked. */
  class ParallelReader implements Runnable {

    private final int sleepMillis;

    ParallelReader(int sleepMillis) {
      this.sleepMillis = sleepMillis;
    }

    public void run() {
      PersistenceManager parallelPM = getPMF().getPersistenceManager();
      Transaction tx = parallelPM.currentTransaction();
      // Run datastore transaction
      tx.setOptimistic(false);
      try {
        tx.begin();
        // read PCPoint instances
        PCPoint p1 = (PCPoint) parallelPM.getObjectById(point1Oid, true);
        p1.name();
        PCPoint p2 = (PCPoint) parallelPM.getObjectById(point2Oid, true);
        p2.name();
        // Give the main thread a chance to try to write the instances
        Thread.sleep(sleepMillis);
        tx.commit();
      } catch (InterruptedException ex) {
        throw new RuntimeException(ex);
      } finally {
        cleanupPM(parallelPM);
      }
    }
  }

  /** Runnable class updating instances and causing them to be write locked. */
  class ParallelWriter implements Runnable {

    private final int sleepMillis;

    ParallelWriter(int sleepMillis) {
      this.sleepMillis = sleepMillis;
    }

    public void run() {
      PersistenceManager parallelPM = getPMF().getPersistenceManager();
      Transaction tx = parallelPM.currentTransaction();
      // Run datastore transaction
      tx.setOptimistic(false);
      try {
        tx.begin();
        PCPoint p1 = (PCPoint) parallelPM.getObjectById(point1Oid);
        p1.setX(210);
        p1.setY(220);
        PCPoint p2 = (PCPoint) parallelPM.getObjectById(point2Oid);
        p2.setX(230);
        p2.setY(240);
        // Flush will set a write lock on the instances
        parallelPM.flush();
        // Give the main thread a chance to try to read the instances
        Thread.sleep(sleepMillis);
        tx.commit();
      } catch (InterruptedException ex) {
        throw new RuntimeException(ex);
      } finally {
        cleanupPM(parallelPM);
      }
    }
  }
}