/*
 * 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.kafka.server.purgatory;

import org.apache.kafka.server.util.timer.TimerTask;

import java.util.concurrent.locks.ReentrantLock;

/**
 * An operation whose processing needs to be delayed for at most the given delayMs. For example
 * a delayed produce operation could be waiting for specified number of acks; or
 * a delayed fetch operation could be waiting for a given number of bytes to accumulate.
 * <br/>
 * The logic upon completing a delayed operation is defined in onComplete() and will be called exactly once.
 * Once an operation is completed, isCompleted() will return true. onComplete() is called from forceComplete(),
 * which is triggered by either expiration, if the operation is not completed after delayMs; or tryComplete(),
 * if the operation can be completed now.
 * <br/>
 * A subclass of DelayedOperation needs to provide an implementation of both onComplete() and tryComplete().
 * <br/>
 * Noted that if you add a future delayed operation that calls ReplicaManager.appendRecords() in onComplete()
 * like DelayedJoin, you must be aware that this operation's onExpiration() needs to call actionQueue.tryCompleteAction().
 */
public abstract class DelayedOperation extends TimerTask {

    private volatile boolean completed = false;

    protected final ReentrantLock lock = new ReentrantLock();

    public DelayedOperation(long delayMs) {
        super(delayMs);
    }

    /*
     * Force completing the delayed operation, if not already completed.
     * This function can be triggered when
     *
     * 1. The operation has been verified to be completable inside tryComplete()
     * 2. The operation has expired and hence needs to be completed right now
     *
     * Return true iff the operation is completed by the caller: note that
     * concurrent threads can try to complete the same operation, but only
     * the first thread will succeed in completing the operation and return
     * true, others will still return false.
     */
    public boolean forceComplete() {
        // Do not proceed if the operation is already completed.
        if (completed) {
            return false;
        }
        // Attain lock prior completing the request.
        lock.lock();
        try {
            // Re-check, if the operation is already completed by some other thread.
            if (!completed) {
                completed = true;
                // cancel the timeout timer
                cancel();
                onComplete();
                return true;
            } else {
                return false;
            }
        } finally {
            lock.unlock();
        }
    }

    /**
     * Check if the delayed operation is already completed
     */
    public boolean isCompleted() {
        return completed;
    }

    /**
     * Call-back to execute when a delayed operation gets expired and hence forced to complete.
     */
    public abstract void onExpiration();

    /**
     * Process for completing an operation; This function needs to be defined
     * in subclasses and will be called exactly once in forceComplete()
     */
    public abstract void onComplete();

    /**
     * Try to complete the delayed operation by first checking if the operation
     * can be completed by now. If yes execute the completion logic by calling
     * forceComplete() and return true iff forceComplete returns true; otherwise return false
     * <br/>
     * This function needs to be defined in subclasses
     */
    public abstract boolean tryComplete();

    /**
     * Thread-safe variant of tryComplete() and call extra function if first tryComplete returns false
     * @param action else function to be executed after first tryComplete returns false
     * @return result of tryComplete
     */
    boolean safeTryCompleteOrElse(Action action) {
        lock.lock();
        try {
            if (tryComplete()) return true;
            else {
                action.apply();
                // last completion check
                return tryComplete();
            }
        } finally {
            lock.unlock();
        }
    }

    /**
     * Thread-safe variant of tryComplete()
     */
    boolean safeTryComplete() {
        lock.lock();
        try {
            if (isCompleted()) return false;
            else return tryComplete();
        } finally {
            lock.unlock();
        }
    }

    /**
     * run() method defines a task that is executed on timeout
     */
    @Override
    public void run() {
        if (forceComplete())
            onExpiration();
    }

    @FunctionalInterface
    public interface Action {
        void apply();
    }
}
