聊聊debezium的BlockingReader

public interface Reader {
​
    public static enum State {
        /**
         * The reader is stopped and static.
         */
        STOPPED,
​
        /**
         * The reader is running and generated records.
         */
        RUNNING,
​
        /**
         * The reader has completed its work or been explicitly stopped, but not all of the generated records have been
         * consumed via {@link Reader#poll() polling}.
         */
        STOPPING;
    }
​
    public String name();
​
    public State state();
​
    public void uponCompletion(Runnable handler);
​
    public default void initialize() {
        // do nothing
    }
​
    public default void destroy() {
        // do nothing
    }
​
    public void start();
​
    public void stop();
​
    public List<SourceRecord> poll() throws InterruptedException;
}

public class BlockingReader implements Reader {
​
    protected final Logger logger = LoggerFactory.getLogger(getClass());
    private final AtomicReference<Runnable> uponCompletion = new AtomicReference<>();
    private final AtomicReference<State> state = new AtomicReference<>();
    private final Metronome metronome;
​
    private final String name;
    private final String runningLogMessage;
​
    public BlockingReader(String name, String runningLogMessage) {
        this.name = name;
        this.metronome = Metronome.parker(ConfigurationDefaults.RETURN_CONTROL_INTERVAL, Clock.SYSTEM);
        this.runningLogMessage = runningLogMessage;
​
    }
​
    /**
     * Does nothing until the connector task is shut down, but regularly returns control back to Connect in order for being paused if requested.
     */
    @Override
    public List<SourceRecord> poll() throws InterruptedException {
        if (state.get() == State.STOPPED) {
            return null;
        }
​
        metronome.pause();
        state.compareAndSet(State.RUNNING, State.STOPPING);
​
        return null;
    }
​
    @Override
    public State state() {
        return state.get();
    }
​
    @Override
    public void uponCompletion(Runnable handler) {
        assert this.uponCompletion.get() == null;
        this.uponCompletion.set(handler);
    }
​
    @Override
    public void start() {
        state.set(State.RUNNING);
        logger.info(runningLogMessage);
    }
​
    @Override
    public void stop() {
        try {
            state.set(State.STOPPED);
​
            // Cleanup Resources
            Runnable completionHandler = uponCompletion.getAndSet(null); // set to null so that we call it only once
            if (completionHandler != null) {
                completionHandler.run();
            }
​
        }
        finally {
            logger.info("Blocking Reader has completed.");
        }
    }
​
    @Override
    public String name() {
        return name;
    }
​
}

public class TimedBlockingReader extends BlockingReader {
​
    protected final Logger logger = LoggerFactory.getLogger(getClass());
​
    private final Duration timeout;
    private volatile Timer timer;
​
    /**
     * @param name Name of the reader
     * @param timeout Duration of time until this TimedBlockingReader should stop
     */
    public TimedBlockingReader(String name, Duration timeout) {
        super(name, "The connector will wait for " + timeout.toMillis() + " ms before proceeding");
        this.timeout = timeout;
    }
​
    @Override
    public void start() {
        super.start();
        this.timer = Threads.timer(Clock.SYSTEM, timeout);
    }
​
    @Override
    public List<SourceRecord> poll() throws InterruptedException {
        super.poll();
​
        // Stop when we've reached the timeout threshold
        if (timer != null && timer.expired()) {
            stop();
        }
​
        return null;
    }
}