<commit_before>package eu.netide.core.caos;

import eu.netide.core.api.IBackendManager;
import eu.netide.core.api.IShimManager;
import eu.netide.core.api.IShimMessageListener;
import eu.netide.core.caos.composition.CompositionSpecification;
import eu.netide.core.caos.composition.CompositionSpecificationLoader;
import eu.netide.core.caos.composition.ExecutionFlowStatus;
import eu.netide.core.caos.execution.FlowExecutors;
import eu.netide.lib.netip.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.xml.bind.JAXBException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

/**
 * Implementation of ICompositionManager
 *
 * Created by timvi on 25.06.2015.
 */
public class CompositionManager implements ICompositionManager, IShimMessageListener {

    private static final Logger logger = LoggerFactory.getLogger(CompositionManager.class);
    private ExecutorService pool = Executors.newSingleThreadExecutor();

    // Settings
    private static String previousCompositionSpecificationXml = "";
    private String compositionSpecificationXml = "";
    private int maxModuleWaitSeconds = 60;

    // Fields
    private IShimManager shimManager;
    private IBackendManager backendManager;
    private CompositionSpecification compositionSpecification = new CompositionSpecification();
    private final Semaphore csLock = new Semaphore(1);
    private volatile boolean correctlyConfigured = false;
    private Future<?> reconfigurationFuture;


    /**
     * Called by blueprint on startup.
     */
    public void Start() {
        logger.debug("CompositionManager started.");
    }

    /**
     * Called by blueprint on shutdown.
     */
    public void Stop() {
        logger.debug("CompositionManager stopped.");
    }

    /**
     * Waits for required modules to connect to the backend
     *
     * @throws TimeoutException Occurs when after maxModuleWaitSeconds not all required modules are connected
     */
    private void ReconfigureAsync() throws TimeoutException {
        reconfigurationFuture = pool.submit(() -> {
            try {
                // wait for required modules to connect for at most maxModuleWaitSeconds
                long modulesUnconnected = compositionSpecification.getModules().size();
                logger.info("Waiting for required modules to connect, " + modulesUnconnected + " left...");
                int waitCount = 0;
                do {
                    modulesUnconnected = compositionSpecification.getModules().stream().filter(m -> !backendManager.getModules().anyMatch(mn -> mn.equals(m.getId()))).count();
                    if (modulesUnconnected > 0) {
                        try {
                            Thread.sleep(2000);
                            waitCount++;
                            if (waitCount % 10 == 0) {
                                logger.info((maxModuleWaitSeconds - waitCount) + " seconds remaining for " + modulesUnconnected + " module(s) to connect...");
                            }
                            if (waitCount >= maxModuleWaitSeconds) {
                                throw new TimeoutException("Required modules did not connect.");
                            }
                        } catch (InterruptedException e) {
                            logger.error("Interrupted while waiting for required modules to connect.", e);
                            return;
                        } catch (TimeoutException e) {
                            logger.error("Reconfiguration unsuccessful: timed out with " + modulesUnconnected + " unconnected modules.", e);
                            return;
                        }
                    }
                } while (modulesUnconnected > 0 && !Thread.interrupted());
                correctlyConfigured = true;
                logger.info("All required modules connected. Reconfiguration successful.");
            } catch (Exception ex) {
                logger.error("Error while reconfiguring.", ex);
                correctlyConfigured = false;
            }
        });
    }

    @Override
    public void OnShimMessage(Message message, String originId) {
        logger.info("CompositionManager received message from shim: " + message.toString());
        try {
            csLock.acquire(); // can only handle when not reconfiguring
            if (correctlyConfigured) {
                ExecutionFlowStatus status = new ExecutionFlowStatus(message);

                status = FlowExecutors.SEQUENTIAL.executeFlow(status, compositionSpecification.getComposition().stream(), shimManager, backendManager);

                // Send results
                for (Map.Entry<Long, List<Message>> entry : status.getResultMessages().entrySet()) {
                    entry.getValue().stream().forEach(shimManager::sendMessage);
                    logger.info("Sent " + entry.getValue().size() + " rules to switch " + entry.getKey());
                }

                logger.info("Flow execution finished.");
            } else {
                logger.error("Could not handle incoming message due to configuration error.", message);
            }
        } catch (Throwable e) {
            logger.error("An exception occurred while handling shim message.", e);
        } finally {
            csLock.release();
        }
    }

    /**
     * Sets the shim manager.
     *
     * @param manager the manager
     */
    public void setShimManager(IShimManager manager) {
        shimManager = manager;
        logger.info("ShimManager set.");
    }

    /**
     * Gets the shim manager.
     *
     * @return the shim manager
     */
    public IShimManager getShimManager() {
        return shimManager;
    }

    /**
     * Sets the backend manager.
     *
     * @param manager the manager
     */
    public void setBackendManager(IBackendManager manager) {
        backendManager = manager;
        logger.info("BackendManager set.");
    }

    /**
     * Gets the backend manager.
     *
     * @return the backend manager
     */
    public IBackendManager getBackendManager() {
        return backendManager;
    }

    /**
     * Gets the current composition specification as XML string.
     *
     * @return composition specification xml
     */
    public String getCompositionSpecificationXml() {
        return compositionSpecificationXml;
    }

    /**
     * Sets the composition specification and triggers a reconfiguration if it was loaded successfully.
     *
     * @param compositionSpecificationXml The composition specification as XML string.
     */
    public void setCompositionSpecificationXml(String compositionSpecificationXml) {
        this.compositionSpecificationXml = compositionSpecificationXml;
        try {
            csLock.acquire();
            correctlyConfigured = false;
            this.compositionSpecification = CompositionSpecificationLoader.Load(this.compositionSpecificationXml);
            logger.info("Accepted new CompositionSpecification '" + this.compositionSpecificationXml + "'.");
            // Accepted specification can be used as fallback
            previousCompositionSpecificationXml = this.compositionSpecificationXml;
            if (reconfigurationFuture != null) {
                // abort running reconfiguration
                reconfigurationFuture.cancel(true);
            }
            ReconfigureAsync();
        } catch (InterruptedException | JAXBException ex) {
            if (compositionSpecificationXml.trim().isEmpty()) {
                // prevent stacktrace on empty specifications
                logger.error("Unable to read new CompositionSpecification. Provided XML was empty.");
            } else {
                logger.error("Unable to read new CompositionSpecification. Provided XML was '" + this.compositionSpecificationXml + "'.", ex);
            }
            logger.warn("Reusing previous specification '" + previousCompositionSpecificationXml + "'.");
            this.compositionSpecificationXml = previousCompositionSpecificationXml;
            if (reconfigurationFuture == null)
                correctlyConfigured = true;
        } catch (TimeoutException ex) {
            logger.error("TimeoutException occurred.", ex);
        } finally {
            csLock.release();
            logger.info("Set composition finished.");
        }
    }

    /**
     * Gets the maximum wait time for module connections in seconds.
     *
     * @return The maximum wait time in seconds.
     */
    public int getMaxModuleWaitSeconds() {
        return this.maxModuleWaitSeconds;
    }

    /**
     * Sets the maximum wait time for module connections in seconds.
     *
     * @param value The new maximum wait time in seconds.
     */
    public void setMaxModuleWaitSeconds(int value) {
        this.maxModuleWaitSeconds = value;
    }
}
<commit_msg>Added bypass for unsupported messages.<commit_after>package eu.netide.core.caos;

import eu.netide.core.api.IBackendManager;
import eu.netide.core.api.IShimManager;
import eu.netide.core.api.IShimMessageListener;
import eu.netide.core.caos.composition.CompositionSpecification;
import eu.netide.core.caos.composition.CompositionSpecificationLoader;
import eu.netide.core.caos.composition.ExecutionFlowStatus;
import eu.netide.core.caos.execution.FlowExecutors;
import eu.netide.lib.netip.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.xml.bind.JAXBException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

/**
 * Implementation of ICompositionManager
 *
 * Created by timvi on 25.06.2015.
 */
public class CompositionManager implements ICompositionManager, IShimMessageListener {

    private static final Logger logger = LoggerFactory.getLogger(CompositionManager.class);
    private ExecutorService pool = Executors.newSingleThreadExecutor();

    // Settings
    private static String previousCompositionSpecificationXml = "";
    private String compositionSpecificationXml = "";
    private int maxModuleWaitSeconds = 60;

    // Fields
    private IShimManager shimManager;
    private IBackendManager backendManager;
    private CompositionSpecification compositionSpecification = new CompositionSpecification();
    private final Semaphore csLock = new Semaphore(1);
    private volatile boolean correctlyConfigured = false;
    private Future<?> reconfigurationFuture;


    /**
     * Called by blueprint on startup.
     */
    public void Start() {
        logger.debug("CompositionManager started.");
    }

    /**
     * Called by blueprint on shutdown.
     */
    public void Stop() {
        logger.debug("CompositionManager stopped.");
    }

    /**
     * Waits for required modules to connect to the backend
     *
     * @throws TimeoutException Occurs when after maxModuleWaitSeconds not all required modules are connected
     */
    private void ReconfigureAsync() throws TimeoutException {
        reconfigurationFuture = pool.submit(() -> {
            try {
                // wait for required modules to connect for at most maxModuleWaitSeconds
                long modulesUnconnected = compositionSpecification.getModules().size();
                logger.info("Waiting for required modules to connect, " + modulesUnconnected + " left...");
                int waitCount = 0;
                do {
                    modulesUnconnected = compositionSpecification.getModules().stream().filter(m -> !backendManager.getModules().anyMatch(mn -> mn.equals(m.getId()))).count();
                    if (modulesUnconnected > 0) {
                        try {
                            Thread.sleep(2000);
                            waitCount++;
                            if (waitCount % 10 == 0) {
                                logger.info((maxModuleWaitSeconds - waitCount) + " seconds remaining for " + modulesUnconnected + " module(s) to connect...");
                            }
                            if (waitCount >= maxModuleWaitSeconds) {
                                throw new TimeoutException("Required modules did not connect.");
                            }
                        } catch (InterruptedException e) {
                            logger.error("Interrupted while waiting for required modules to connect.", e);
                            return;
                        } catch (TimeoutException e) {
                            logger.error("Reconfiguration unsuccessful: timed out with " + modulesUnconnected + " unconnected modules.", e);
                            return;
                        }
                    }
                } while (modulesUnconnected > 0 && !Thread.interrupted());
                correctlyConfigured = true;
                logger.info("All required modules connected. Reconfiguration successful.");
            } catch (Exception ex) {
                logger.error("Error while reconfiguring.", ex);
                correctlyConfigured = false;
            }
        });
    }

    @Override
    public void OnShimMessage(Message message, String originId) {
        logger.info("CompositionManager received message from shim: " + message.toString());
        try {
            csLock.acquire(); // can only handle when not reconfiguring
            if (correctlyConfigured) {
                ExecutionFlowStatus status = new ExecutionFlowStatus(message);

                status = FlowExecutors.SEQUENTIAL.executeFlow(status, compositionSpecification.getComposition().stream(), shimManager, backendManager);

                // Send results
                for (Map.Entry<Long, List<Message>> entry : status.getResultMessages().entrySet()) {
                    entry.getValue().stream().forEach(shimManager::sendMessage);
                    logger.info("Sent " + entry.getValue().size() + " rules to switch " + entry.getKey());
                }

                logger.info("Flow execution finished.");
            } else {
                logger.error("Could not handle incoming message due to configuration error.", message);
            }
        } catch (UnsupportedOperationException e) {
            logger.warn("Received unsupported message for composition, attempting to relay instead.", e);
            try {
                backendManager.sendMessage(message);
            } catch (Throwable ex) {
                logger.error("Could not relay unsupported message.", ex);
            }
        } catch (Throwable e) {
            logger.error("An exception occurred while handling shim message.", e);
        } finally {
            csLock.release();
        }
    }

    /**
     * Sets the shim manager.
     *
     * @param manager the manager
     */
    public void setShimManager(IShimManager manager) {
        shimManager = manager;
        logger.info("ShimManager set.");
    }

    /**
     * Gets the shim manager.
     *
     * @return the shim manager
     */
    public IShimManager getShimManager() {
        return shimManager;
    }

    /**
     * Sets the backend manager.
     *
     * @param manager the manager
     */
    public void setBackendManager(IBackendManager manager) {
        backendManager = manager;
        logger.info("BackendManager set.");
    }

    /**
     * Gets the backend manager.
     *
     * @return the backend manager
     */
    public IBackendManager getBackendManager() {
        return backendManager;
    }

    /**
     * Gets the current composition specification as XML string.
     *
     * @return composition specification xml
     */
    public String getCompositionSpecificationXml() {
        return compositionSpecificationXml;
    }

    /**
     * Sets the composition specification and triggers a reconfiguration if it was loaded successfully.
     *
     * @param compositionSpecificationXml The composition specification as XML string.
     */
    public void setCompositionSpecificationXml(String compositionSpecificationXml) {
        this.compositionSpecificationXml = compositionSpecificationXml;
        try {
            csLock.acquire();
            correctlyConfigured = false;
            this.compositionSpecification = CompositionSpecificationLoader.Load(this.compositionSpecificationXml);
            logger.info("Accepted new CompositionSpecification '" + this.compositionSpecificationXml + "'.");
            // Accepted specification can be used as fallback
            previousCompositionSpecificationXml = this.compositionSpecificationXml;
            if (reconfigurationFuture != null) {
                // abort running reconfiguration
                reconfigurationFuture.cancel(true);
            }
            ReconfigureAsync();
        } catch (InterruptedException | JAXBException ex) {
            if (compositionSpecificationXml.trim().isEmpty()) {
                // prevent stacktrace on empty specifications
                logger.error("Unable to read new CompositionSpecification. Provided XML was empty.");
            } else {
                logger.error("Unable to read new CompositionSpecification. Provided XML was '" + this.compositionSpecificationXml + "'.", ex);
            }
            logger.warn("Reusing previous specification '" + previousCompositionSpecificationXml + "'.");
            this.compositionSpecificationXml = previousCompositionSpecificationXml;
            if (reconfigurationFuture == null)
                correctlyConfigured = true;
        } catch (TimeoutException ex) {
            logger.error("TimeoutException occurred.", ex);
        } finally {
            csLock.release();
            logger.info("Set composition finished.");
        }
    }

    /**
     * Gets the maximum wait time for module connections in seconds.
     *
     * @return The maximum wait time in seconds.
     */
    public int getMaxModuleWaitSeconds() {
        return this.maxModuleWaitSeconds;
    }

    /**
     * Sets the maximum wait time for module connections in seconds.
     *
     * @param value The new maximum wait time in seconds.
     */
    public void setMaxModuleWaitSeconds(int value) {
        this.maxModuleWaitSeconds = value;
    }
}

{'fraction_non_alphanumeric': 0.08794842122664903, 'fraction_numerical': 0.0025348542458808617, 'mean_word_length': 2.1534317984361424, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 4}, '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': '6616319', 'n_tokens_mistral': 4355, 'n_tokens_neox': 4189, 'n_words': 1124}