// Path: berserker-commons/src/main/java/io/smartcat/berserker/api/DataSource.java
// public interface DataSource<T> {
// 
//     /**
//      * Returns true if data source can provide next value, otherwise false.
//      *
//      * @param time Relative time in nanoseconds from starting load generator. Time can be used for implementing time
//      *            dependent data source which can then return data with time stamps and/or data influenced by time in
//      *            any other way.
//      * @return True if data source can provide next value, otherwise false.
//      */
//     boolean hasNext(long time);
// 
//     /**
//      * Returns next value from this data source.
//      *
//      * @param time Relative time in nanoseconds from starting load generator. Time can be used for implementing time
//      *            dependent data source which can then return data with time stamps and/or data influenced by time in
//      *            any other way.
//      * @return Next value from this data source.
//      */
//     T getNext(long time);
// }


import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.smartcat.berserker.api.DataSource;
package io.smartcat.berserker.datasource;




/**
 * Data source buffer which polls from specified delegate data source from different thread. Blocks polling at the
 * beginning until buffer is filled to <code>initialBufferFullness</code>.
 * <p>
 * <b>Note:</b>Only time independent {@link DataSource} implementations should be used with this implementation.
 * For more info read the next paragraph.
 * </p>
 * <p>
 * Buffered data source breaks {@link DataSource} interface since <code>delegate</code> data source is called
 * with time value of <code>0</code>. This is due to buffer implementation which fills buffer from another thread.
 * Since <code>delegate.getNext(...)</code> is invoked from another thread, it does not have information on time
 * component, hence cannot invoke <code>delegate.getNext(...)</code> with proper value. If you are using
 * time dependent {@link DataSource} implementation, it will not work together with this implementation.
 * </p>
 *
 * @param <T> Type of the data provided by data source implementation.
 */
public class BufferedDataSource<T> implements DataSource<T>, AutoCloseable {

    private static final int DEFAULT_BUFFER_SIZE = 1024;
    private static final double DEFAULT_INITIAL_BUFFER_FULLNESS = 0.75;

    private static final Logger LOGGER = LoggerFactory.getLogger(BufferedDataSource.class);

    private final Object bufferFilled = new Object();
    private final DataSource<T> delegate;
    private final int bufferSize;
    private final Queue<T> buffer;
    private final double initialBufferFullness;

    private AtomicBoolean terminateThread = new AtomicBoolean(false);
    private Thread thread;

    /**
     * Constructs a buffered data source with specified <code>delegate</code> data source. <code>bufferSize</code> is
     * set to 1024, <code>initialBufferFullness</code> is set to 75%. Blocks polling at the beginning until buffer is
     * filled to <code>initialBufferFullness</code>.
     *
     * @param delegate Data source which will be used to fill buffer.
     */
    public BufferedDataSource(DataSource<T> delegate) {
        this(delegate, DEFAULT_BUFFER_SIZE);
    }

    /**
     * Constructs a buffered data source with specified <code>delegate</code> data source and <code>bufferSize</code>.
     * <code>initialBufferFullness</code> is set to 75%. Blocks polling at the beginning until buffer is filled to
     * <code>initialBufferFullness</code>.
     *
     * @param delegate Data source which will be used to fill buffer.
     * @param bufferSize Size of the buffer.
     */
    public BufferedDataSource(DataSource<T> delegate, int bufferSize) {
        this(delegate, bufferSize, DEFAULT_INITIAL_BUFFER_FULLNESS);
    }

    /**
     * Constructs a buffered data source with specified <code>delegate</code> data source, <code>bufferSize</code> and
     * <code>initialBufferFullness</code>. Blocks polling at the beginning until buffer is filled to
     * <code>initialBufferFullness</code>.
     *
     * @param delegate Data source which will be used to fill buffer.
     * @param bufferSize Size of the buffer.
     * @param initialBufferFullness Initial fullness of buffer.
     */
    public BufferedDataSource(DataSource<T> delegate, int bufferSize, double initialBufferFullness) {
        this.delegate = delegate;
        this.bufferSize = bufferSize;
        this.buffer = new LinkedBlockingQueue<>(bufferSize);
        this.initialBufferFullness = initialBufferFullness;
        fillBuffer();
    }

    @Override
    public boolean hasNext(long time) {
        return !buffer.isEmpty() || delegate.hasNext(time);
    }

// Path: berserker-commons/src/main/java/io/smartcat/berserker/api/DataSource.java
// public interface DataSource<T> {
// 
//     /**
//      * Returns true if data source can provide next value, otherwise false.
//      *
//      * @param time Relative time in nanoseconds from starting load generator. Time can be used for implementing time
//      *            dependent data source which can then return data with time stamps and/or data influenced by time in
//      *            any other way.
//      * @return True if data source can provide next value, otherwise false.
//      */
//     boolean hasNext(long time);
// 
//     /**
//      * Returns next value from this data source.
//      *
//      * @param time Relative time in nanoseconds from starting load generator. Time can be used for implementing time
//      *            dependent data source which can then return data with time stamps and/or data influenced by time in
//      *            any other way.
//      * @return Next value from this data source.
//      */
//     T getNext(long time);
// }


// Path: berserker-core/src/main/java/io/smartcat/berserker/datasource/BufferedDataSource.java
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.smartcat.berserker.api.DataSource;

package io.smartcat.berserker.datasource;




/**
 * Data source buffer which polls from specified delegate data source from different thread. Blocks polling at the
 * beginning until buffer is filled to <code>initialBufferFullness</code>.
 * <p>
 * <b>Note:</b>Only time independent {@link DataSource} implementations should be used with this implementation.
 * For more info read the next paragraph.
 * </p>
 * <p>
 * Buffered data source breaks {@link DataSource} interface since <code>delegate</code> data source is called
 * with time value of <code>0</code>. This is due to buffer implementation which fills buffer from another thread.
 * Since <code>delegate.getNext(...)</code> is invoked from another thread, it does not have information on time
 * component, hence cannot invoke <code>delegate.getNext(...)</code> with proper value. If you are using
 * time dependent {@link DataSource} implementation, it will not work together with this implementation.
 * </p>
 *
 * @param <T> Type of the data provided by data source implementation.
 */
public class BufferedDataSource<T> implements DataSource<T>, AutoCloseable {

    private static final int DEFAULT_BUFFER_SIZE = 1024;
    private static final double DEFAULT_INITIAL_BUFFER_FULLNESS = 0.75;

    private static final Logger LOGGER = LoggerFactory.getLogger(BufferedDataSource.class);

    private final Object bufferFilled = new Object();
    private final DataSource<T> delegate;
    private final int bufferSize;
    private final Queue<T> buffer;
    private final double initialBufferFullness;

    private AtomicBoolean terminateThread = new AtomicBoolean(false);
    private Thread thread;

    /**
     * Constructs a buffered data source with specified <code>delegate</code> data source. <code>bufferSize</code> is
     * set to 1024, <code>initialBufferFullness</code> is set to 75%. Blocks polling at the beginning until buffer is
     * filled to <code>initialBufferFullness</code>.
     *
     * @param delegate Data source which will be used to fill buffer.
     */
    public BufferedDataSource(DataSource<T> delegate) {
        this(delegate, DEFAULT_BUFFER_SIZE);
    }

    /**
     * Constructs a buffered data source with specified <code>delegate</code> data source and <code>bufferSize</code>.
     * <code>initialBufferFullness</code> is set to 75%. Blocks polling at the beginning until buffer is filled to
     * <code>initialBufferFullness</code>.
     *
     * @param delegate Data source which will be used to fill buffer.
     * @param bufferSize Size of the buffer.
     */
    public BufferedDataSource(DataSource<T> delegate, int bufferSize) {
        this(delegate, bufferSize, DEFAULT_INITIAL_BUFFER_FULLNESS);
    }

    /**
     * Constructs a buffered data source with specified <code>delegate</code> data source, <code>bufferSize</code> and
     * <code>initialBufferFullness</code>. Blocks polling at the beginning until buffer is filled to
     * <code>initialBufferFullness</code>.
     *
     * @param delegate Data source which will be used to fill buffer.
     * @param bufferSize Size of the buffer.
     * @param initialBufferFullness Initial fullness of buffer.
     */
    public BufferedDataSource(DataSource<T> delegate, int bufferSize, double initialBufferFullness) {
        this.delegate = delegate;
        this.bufferSize = bufferSize;
        this.buffer = new LinkedBlockingQueue<>(bufferSize);
        this.initialBufferFullness = initialBufferFullness;
        fillBuffer();
    }

    @Override
    public boolean hasNext(long time) {
        return !buffer.isEmpty() || delegate.hasNext(time);
    }

    @Override