/*
 * Copyright Debezium Authors.
 *
 * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
 */
package io.debezium.connector.mysql;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import org.apache.kafka.connect.source.SourceRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.debezium.annotation.ThreadSafe;

/**
 * A {@link Reader} implementation that runs one or more other {@link Reader}s in a consistently, completely, and sequentially.
 * This reader ensures that all records generated by one of its contained {@link Reader}s are all passed through to callers
 * via {@link #poll() polling} before the next reader is started. And, when this reader is {@link #stop() stopped}, this
 * class ensures that current reader is stopped and that no additional readers will be started.
 * 
 * @author Randall Hauch
 */
@ThreadSafe
public final class ChainedReader implements Reader {

    private final Logger logger = LoggerFactory.getLogger(getClass());
    private final List<Reader> readers = new ArrayList<>();
    private final LinkedList<Reader> remainingReaders = new LinkedList<>();
    private final AtomicBoolean running = new AtomicBoolean();
    private final AtomicBoolean completed = new AtomicBoolean(true);
    private final AtomicReference<Reader> currentReader = new AtomicReference<>();
    private final AtomicReference<Runnable> uponCompletion = new AtomicReference<>();
    private final AtomicReference<String> completionMessage = new AtomicReference<>();

    /**
     * Create a new chained reader.
     */
    public ChainedReader() {
    }

    /**
     * Set the message that should be logged when all of the readers have completed their work.
     * 
     * @param msg the message; may be null
     */
    public void uponCompletion(String msg) {
        completionMessage.set(msg);
    }

    @Override
    public void uponCompletion(Runnable handler) {
        uponCompletion.set(handler);
    }

    protected ChainedReader add(Reader reader) {
        assert reader != null;
        reader.uponCompletion(this::readerCompletedPolling);
        readers.add(reader);
        return this;
    }

    @Override
    public synchronized void start() {
        if (running.compareAndSet(false, true)) {
            completed.set(false);

            // Build up the list of readers that need to be called ...
            remainingReaders.clear();
            readers.forEach(remainingReaders::add);

            // Start the first reader, if there is one ...
            if (!startNextReader()) {
                // We couldn't start it ...
                running.set(false);
                completed.set(true);
            }
        }
    }

    @Override
    public synchronized void stop() {
        if (running.compareAndSet(true, false)) {
            // First, remove all readers that have not yet been started, which ensures the next one is is not started
            // while we're trying to stop the previous one ...
            remainingReaders.clear();
            // Then stop the currently-running reader but do not remove it as it will be removed when it completes ...
            Reader current = currentReader.get();
            if (current != null) {
                try {
                    logger.info("Stopping the {} reader", current.name());
                    current.stop();
                } catch (Throwable t) {
                    logger.error("Unexpected error stopping the {} reader", current.name(), t);
                }
            }
        }
    }

    @Override
    public State state() {
        if (running.get()) {
            return State.RUNNING;
        }
        return completed.get() ? State.STOPPED : State.STOPPING;
    }

    @Override
    public List<SourceRecord> poll() throws InterruptedException {
        // We have to be prepared for the current reader to be in the middle of stopping and this chain transitioning
        // to the next reader. In this case, the reader that is stopping will return null or empty, and we can ignore
        // those until the next reader has started ...
        while (running.get() || !completed.get()) {
            Reader reader = currentReader.get();
            if (reader != null) {
                List<SourceRecord> records = reader.poll();
                if (records != null && !records.isEmpty()) return records;
                // otherwise, we'll go ahead until the next reader is ready or until we're no longer running ...
            }
        }
        return null;
    }

    /**
     * Called when the previously-started reader has returned all of its records via {@link #poll() polling}.
     * Only when this method is called is the now-completed reader removed as the current reader, and this is what
     * guarantees that all records produced by the now-completed reader have been polled.
     */
    private synchronized void readerCompletedPolling() {
        if (!startNextReader()) {
            // We've finished with the last reader ...
            try {
                if (running.get() || !completed.get()) {
                    // Notify the handler ...
                    Runnable handler = uponCompletion.get();
                    if (handler != null) {
                        handler.run();
                    }
                    // and output our message ...
                    String msg = completionMessage.get();
                    if (msg != null) {
                        logger.info(msg);
                    }
                }
            } finally {
                // And since this is the last reader, make sure this chain is also stopped ...
                completed.set(true);
                running.set(false);
            }
        }
    }

    /**
     * Start the next reader.
     * 
     * @return {@code true} if the next reader was started, or {@code false} if there are no more readers
     */
    private boolean startNextReader() {
        Reader reader = remainingReaders.isEmpty() ? null : remainingReaders.pop();
        if (reader == null) {
            // There are no readers, so nothing to do ...
            Reader lastReader = currentReader.getAndSet(null);
            if (lastReader != null) {
                // Make sure it has indeed stopped ...
                lastReader.stop();
            }
            return false;
        }

        // There is at least one more reader, so start it ...
        Reader lastReader = currentReader.getAndSet(null);
        if (lastReader != null) {
            logger.debug("Transitioning from the {} reader to the {} reader", lastReader.name(), reader.name());
        } else {
            logger.debug("Starting the {} reader", reader.name());
        }
        reader.start();
        currentReader.set(reader);
        return true;
    }

    @Override
    public String name() {
        Reader reader = currentReader.get();
        return reader != null ? reader.name() : "chained";
    }
}
