<commit_before>package eu.qualimaster.dataManagement;

import java.util.ArrayList;
import java.util.List;

import eu.qualimaster.dataManagement.common.AbstractDataManager;
import eu.qualimaster.dataManagement.events.ReferenceDataManagementEvent;
import eu.qualimaster.dataManagement.events.ShutdownEvent;
import eu.qualimaster.dataManagement.events.ReferenceDataManagementEvent.Command;
import eu.qualimaster.dataManagement.sinks.IDataSink;
import eu.qualimaster.dataManagement.sources.IDataSource;
import eu.qualimaster.dataManagement.storage.AbstractStorageManager;
import eu.qualimaster.dataManagement.storage.hbase.HBaseBatchStorageSupport;
import eu.qualimaster.dataManagement.storage.hbase.HBaseStorageTable;
import eu.qualimaster.dataManagement.strategies.IStorageStrategyDescriptor;
import eu.qualimaster.events.AbstractEvent;
import eu.qualimaster.events.EventHandler;
import eu.qualimaster.events.EventManager;
import eu.qualimaster.infrastructure.PipelineLifecycleEvent;

/**
 * Realizes the external interface of the storage manager.
 * 
 * @author <NAME>
 */
public class DataManager {

    /**
     * The data source data manager.
     */
    public static final DataSourceManager DATA_SOURCE_MANAGER = new DataSourceManager();

    /**
     * The data sink data manager.
     */
    public static final DataSinkManager DATA_SINK_MANAGER = new DataSinkManager();

    /**
     * The intermediary processing data storage manager.
     */
    public static final AbstractStorageManager<?> INTERMEDIARY_STORAGE_MANAGER = new HBaseStorageManager(
                    "intermediary.");

    /**
     * The intermediary processing data storage manager.
     */
    public static final AbstractStorageManager<?> ADAPTATION_STORAGE_MANAGER = new HBaseStorageManager("adaptation.");

    /**
     * The temporary data manager for replay mechanism - Default Redis-based
     * key-value store
     */
    public static final AbstractStorageManager<?> REPLAY_STORAGE_MANAGER = new ReplayStorageManager("replay.");

    /**
     * The data manager to store aggregated source volumes (for source volume
     * prediction).
     */
    public static final AbstractStorageManager<?> VOLUME_PREDICTION_STORAGE_MANAGER = new HBaseStorageManager(
                    "volumePrediction.");

    private static boolean started = false;
    private static boolean localMode = false;
    private static boolean isInitialized = false;

    /**
     * The handler for monitoring events.
     * 
     * @author <NAME>
     */
    private static class PipelineLifecycleEventEventHandler extends EventHandler<PipelineLifecycleEvent> {

        /**
         * Creates an adaptation event handler.
         */
        protected PipelineLifecycleEventEventHandler() {
            super(PipelineLifecycleEvent.class);
        }

        @Override
        protected void handle(PipelineLifecycleEvent event) {
            String pipelineName = event.getPipeline();
            switch (event.getStatus()) {
            case STOPPING:
                disconnectAll(pipelineName); // see D5.2 stopping a pipeline
                discardAll(pipelineName);
                break;
            case INITIALIZED:
                connectAll(pipelineName);
                // TODO this is just a workaround and shall be removed as soon
                // as possible
                long additionalDelay = DataManagementConfiguration.getPipelineStartNotificationDelay();
                if (additionalDelay > 0) {
                    try {
                        Thread.sleep(additionalDelay);
                    } catch (InterruptedException e) {
                    }
                }
                // now we are able to work -> switch to started
                EventManager.handle(new PipelineLifecycleEvent(event, PipelineLifecycleEvent.Status.STARTED));
                break;
            default:
                break;
            }
        }

    }

    /**
     * Handles a shutdown event.
     * 
     * @author <NAME>
     */
    private static class ShutdownEventHandler extends EventHandler<ShutdownEvent> {

        protected ShutdownEventHandler() {
            super(ShutdownEvent.class);
        }
<commit_msg>extended cleanup at pipeline startup <commit_after>package eu.qualimaster.dataManagement;

import java.util.ArrayList;
import java.util.List;

import eu.qualimaster.dataManagement.common.AbstractDataManager;
import eu.qualimaster.dataManagement.events.ReferenceDataManagementEvent;
import eu.qualimaster.dataManagement.events.ShutdownEvent;
import eu.qualimaster.dataManagement.events.ReferenceDataManagementEvent.Command;
import eu.qualimaster.dataManagement.sinks.IDataSink;
import eu.qualimaster.dataManagement.sources.IDataSource;
import eu.qualimaster.dataManagement.storage.AbstractStorageManager;
import eu.qualimaster.dataManagement.storage.hbase.HBaseBatchStorageSupport;
import eu.qualimaster.dataManagement.storage.hbase.HBaseStorageTable;
import eu.qualimaster.dataManagement.strategies.IStorageStrategyDescriptor;
import eu.qualimaster.events.AbstractEvent;
import eu.qualimaster.events.EventHandler;
import eu.qualimaster.events.EventManager;
import eu.qualimaster.infrastructure.PipelineLifecycleEvent;

/**
 * Realizes the external interface of the storage manager.
 * 
 * @author <NAME>
 */
public class DataManager {

    /**
     * The data source data manager.
     */
    public static final DataSourceManager DATA_SOURCE_MANAGER = new DataSourceManager();

    /**
     * The data sink data manager.
     */
    public static final DataSinkManager DATA_SINK_MANAGER = new DataSinkManager();

    /**
     * The intermediary processing data storage manager.
     */
    public static final AbstractStorageManager<?> INTERMEDIARY_STORAGE_MANAGER = new HBaseStorageManager(
                    "intermediary.");

    /**
     * The intermediary processing data storage manager.
     */
    public static final AbstractStorageManager<?> ADAPTATION_STORAGE_MANAGER = new HBaseStorageManager("adaptation.");

    /**
     * The temporary data manager for replay mechanism - Default Redis-based
     * key-value store
     */
    public static final AbstractStorageManager<?> REPLAY_STORAGE_MANAGER = new ReplayStorageManager("replay.");

    /**
     * The data manager to store aggregated source volumes (for source volume
     * prediction).
     */
    public static final AbstractStorageManager<?> VOLUME_PREDICTION_STORAGE_MANAGER = new HBaseStorageManager(
                    "volumePrediction.");

    private static boolean started = false;
    private static boolean localMode = false;
    private static boolean isInitialized = false;

    /**
     * The handler for monitoring events.
     * 
     * @author <NAME>
     */
    private static class PipelineLifecycleEventEventHandler extends EventHandler<PipelineLifecycleEvent> {

        /**
         * Creates an adaptation event handler.
         */
        protected PipelineLifecycleEventEventHandler() {
            super(PipelineLifecycleEvent.class);
        }

        @Override
        protected void handle(PipelineLifecycleEvent event) {
            String pipelineName = event.getPipeline();
            switch (event.getStatus()) {
            case STARTING:
                // to be on the safe side if a previous execution left something over
                discardAll(pipelineName);
                break;
            case STOPPING:
                disconnectAll(pipelineName); // see D5.2 stopping a pipeline
                discardAll(pipelineName);
                break;
            case INITIALIZED:
                connectAll(pipelineName);
                // waiting is a workaround
                long additionalDelay = DataManagementConfiguration.getPipelineStartNotificationDelay();
                if (additionalDelay > 0) {
                    try {
                        Thread.sleep(additionalDelay);
                    } catch (InterruptedException e) {
                    }
                }
                // now we are able to work -> switch to started
                EventManager.handle(new PipelineLifecycleEvent(event, PipelineLifecycleEvent.Status.STARTED));
                break;
            default:
                break;
            }
        }

    }

    /**
     * Handles a shutdown event.
     * 
     * @author <NAME>
     */
    private static class ShutdownEventHandler extends EventHandler<ShutdownEvent> {

        protected ShutdownEventHandler() {
            super(ShutdownEvent.class);
        }

{'fraction_non_alphanumeric': 0.0759508484493856, 'fraction_numerical': 0.0007021650087770626, 'mean_word_length': 2.586235837180025, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 25, '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': '536873', 'n_tokens_mistral': 2155, 'n_tokens_neox': 2095, 'n_words': 512}