<commit_before>package eu.qualimaster.dataManagement.sinks.replay;

import eu.qualimaster.dataManagement.DataManager;
import eu.qualimaster.dataManagement.common.replay.Tuple;
import eu.qualimaster.dataManagement.serialization.ISerializer;
import eu.qualimaster.dataManagement.serialization.SerializerRegistry;
import eu.qualimaster.dataManagement.storage.AbstractStorageTable;
import eu.qualimaster.dataManagement.storage.support.IStorageSupport;
import eu.qualimaster.dataManagement.strategies.IStorageStrategyDescriptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * Simulate the results from the Replay Store (currently HBase) as stream items
 * @author tuan
 * @since 03/06/16.
 */
public class ReplayStreamer<T> {

    private static final Logger LOG = LoggerFactory.getLogger(ReplayStreamer.class);

    /** Speed factor */
    private float speedFactor = -1;
    private Date startDate, endDate;

    /** the query as passed from the client. So far we only process
     * key query (point or range) */
    private String query;
    private String prefix;
    
    /** Control the thread of fetching data from HBase */
    private volatile boolean stopFetchingDataThread = false;

    /**
     * Timeout 50 miliseconds for stop querying the database to have
     * a pseudo-non-blocking call to the getData()
     */
    private static final int TIMEOUT = 50;
    private int counter = 0;
    
    private LinkedBlockingQueue<T> buffer;

    private ISerializer<T> serializer;
    private ReplayDataInput resultWrapper;
    private ExecutorService fetcherThread;

    public ReplayStreamer(Class<T> cls, Tuple schema, String location, IStorageStrategyDescriptor d) {
        AbstractStorageTable table = DataManager.REPLAY_STORAGE_MANAGER.getTable(location, schema.getName(), d);
        IStorageSupport storage = table.getStorageSupport();
        resultWrapper = new ReplayDataInput(schema, storage);
        this.serializer = SerializerRegistry.getSerializer(cls.getSimpleName(), cls);
        buffer = new LinkedBlockingQueue<>(100);
        fetcherThread = Executors.newSingleThreadExecutor();
        fetcherThread.submit(new DataFetcher());
    }

    /* Need to synchronize the speed set with other thread as well ? */
    public void setSpeed(float speed) {
        stopFetchingDataThread = true;
        buffer.clear();
        speedFactor = speed;
        updateQuery();
        stopFetchingDataThread = false;
    }

    public void setStart(Date date) {
    	stopFetchingDataThread = true;

        // the internal fullyLock() method is called inside BlockingQueue's
        // clear() method code so no race condition here
        buffer.clear();
        startDate = date;
        updateQuery();
        stopFetchingDataThread = false;
    }

    public void setEnd(Date date) {
    	stopFetchingDataThread = true;
        buffer.clear();
        endDate = date;
        updateQuery();
        stopFetchingDataThread = false;
    }

    public void setQuery(String query) {
    	stopFetchingDataThread = true;
        buffer.clear();
        this.query = query;
        updateQuery();
        stopFetchingDataThread = false;
    }

    /** Note: Mock the test of serializer */
    public void setSerializer(ISerializer s) {
        this.serializer = s;
    }

    /**
     * Get the data of the current parameters. Null value means two things:
     * - some error happens (server error, parsing of schema, timeout...),
     *  the call still moves on
     * - the current stream is all finished<commit_msg>temporarily comment the buffer reset<commit_after>package eu.qualimaster.dataManagement.sinks.replay;

import eu.qualimaster.dataManagement.DataManager;
import eu.qualimaster.dataManagement.common.replay.Tuple;
import eu.qualimaster.dataManagement.serialization.ISerializer;
import eu.qualimaster.dataManagement.serialization.SerializerRegistry;
import eu.qualimaster.dataManagement.storage.AbstractStorageTable;
import eu.qualimaster.dataManagement.storage.support.IStorageSupport;
import eu.qualimaster.dataManagement.strategies.IStorageStrategyDescriptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * Simulate the results from the Replay Store (currently HBase) as stream items
 * @author tuan
 * @since 03/06/16.
 */
public class ReplayStreamer<T> {

    private static final Logger LOG = LoggerFactory.getLogger(ReplayStreamer.class);

    /** Speed factor */
    private float speedFactor = -1;
    private Date startDate, endDate;

    /** the query as passed from the client. So far we only process
     * key query (point or range) */
    private String query;
    private String prefix;
    
    /** Control the thread of fetching data from HBase */
    private volatile boolean stopFetchingDataThread = false;

    /**
     * Timeout 50 miliseconds for stop querying the database to have
     * a pseudo-non-blocking call to the getData()
     */
    private static final int TIMEOUT = 50;
    private int counter = 0;
    
    private LinkedBlockingQueue<T> buffer;

    private ISerializer<T> serializer;
    private ReplayDataInput resultWrapper;
    private ExecutorService fetcherThread;

    public ReplayStreamer(Class<T> cls, Tuple schema, String location, IStorageStrategyDescriptor d) {
        AbstractStorageTable table = DataManager.REPLAY_STORAGE_MANAGER.getTable(location, schema.getName(), d);
        IStorageSupport storage = table.getStorageSupport();
        resultWrapper = new ReplayDataInput(schema, storage);
        this.serializer = SerializerRegistry.getSerializer(cls.getSimpleName(), cls);
        buffer = new LinkedBlockingQueue<>(100);
        fetcherThread = Executors.newSingleThreadExecutor();
        fetcherThread.submit(new DataFetcher());
    }

    /* Need to synchronize the speed set with other thread as well ? */
    public void setSpeed(float speed) {
        stopFetchingDataThread = true;
        // buffer.clear();
        speedFactor = speed;
        updateQuery();
        stopFetchingDataThread = false;
    }

    public void setStart(Date date) {
    	stopFetchingDataThread = true;

        // the internal fullyLock() method is called inside BlockingQueue's
        // clear() method code so no race condition here
        // buffer.clear();
        startDate = date;
        updateQuery();
        stopFetchingDataThread = false;
    }

    public void setEnd(Date date) {
    	stopFetchingDataThread = true;
        // buffer.clear();
        endDate = date;
        updateQuery();
        stopFetchingDataThread = false;
    }

    public void setQuery(String query) {
    	stopFetchingDataThread = true;
        // buffer.clear();
        this.query = query;
        updateQuery();
        stopFetchingDataThread = false;
    }

    /** Note: Mock the test of serializer */
    public void setSerializer(ISerializer s) {
        this.serializer = s;
    }

    /**
     * Get the data of the current parameters. Null value means two things:
     * - some error happens (server error, parsing of schema, timeout...),
     *  the call still moves on
     * - the current stream is all finished
{'fraction_non_alphanumeric': 0.08530305031031295, 'fraction_numerical': 0.004489634226858576, 'mean_word_length': 3.238388360380526, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1924141', 'n_tokens_mistral': 2075, 'n_tokens_neox': 2046, 'n_words': 585}