<commit_before>/**
 * Copyright (C) 2010-14 diirt developers. See COPYRIGHT.TXT
 * All rights reserved. Use is subject to license terms. See LICENSE.TXT
 */
package org.csstudio.alarm.diirt.datasource;

import java.net.URI;
import java.net.URLDecoder;
import java.net.URLEncoder;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.csstudio.alarm.beast.client.AlarmTreeItem;
import org.csstudio.alarm.beast.client.AlarmTreePV;
import org.csstudio.alarm.beast.ui.clientmodel.AlarmClientModel;
import org.csstudio.alarm.beast.ui.clientmodel.AlarmClientModelConfigListener;
import org.csstudio.alarm.beast.ui.clientmodel.AlarmClientModelListener;
import org.diirt.datasource.ChannelHandler;
import org.diirt.datasource.DataSource;
import org.diirt.datasource.vtype.DataTypeSupport;
import org.eclipse.core.runtime.Platform;

/**
 * @author <NAME>
 *
 */
public class BeastDataSource extends DataSource implements AlarmClientModelConfigListener {
    private static final Logger log = Logger.getLogger(BeastDataSource.class.getName());

    private static final String PREF_QUALIFIER = "org.csstudio.alarm.diirt.datasource";
    private static final String PREF_KEY_COMPOSITE = "composite.model.name";
    private static final String PREF_DEFAULT_COMPOSITE_MODEL_NAME = "Composite";

    // The model, activeAlarms and acknowledgedAlarms is shared by the entire
    // datasource, the benefit of does this at the datasource level instead of
    // in each channel is that they need to be computed only once and only a single
    // copy needs to be maintained.
    private AlarmClientModel model;

    private Map<String, List<Consumer>> channelConsumers = Collections.synchronizedMap(new HashMap<String, List<Consumer>>());
    private Map<String, AlarmClientModel> models = Collections.synchronizedMap(new HashMap<String, AlarmClientModel>());
    private AtomicInteger loaded;
    private AtomicInteger toLoad;

    private Executor executor = Executors.newScheduledThreadPool(4);

    private BeastTypeSupport typeSupport;

    private BeastDataSourceConfiguration configuration;

    private BeastAlarmClientModelListener modelListener;

    private CompositeAlarmClientModel compositeModel;

    static {
        // Install type support for the types it generates.
        DataTypeSupport.install();
    }

    private class BeastAlarmClientModelListener implements AlarmClientModelListener{
        private BeastDataSource parent;

        BeastAlarmClientModelListener(BeastDataSource parent) {
            this.parent = parent;
        }

        @Override
        public void newAlarmConfiguration(AlarmClientModel model) {
            log.config("beast  datasource: new alarm configuration --- " + model);
            // new model loaded
            parent.loaded.incrementAndGet();
            if (!parent.compositeModel.equals(model)) parent.compositeModel.addAlarmClientModel(model);
            if (parent.loaded.get() == parent.toLoad.get()) {
                parent.compositeModel.setAllLoaded(true);
                log.log(Level.FINE, "All models loaded - notifying Composite model listeners.");
                parent.compositeModel.compositeModelsLoaded();
            }
            synchronized (parent.channelConsumers) {
                for (String channelName : parent.channelConsumers.keySet()) {
                    BeastChannelHandler channel = (BeastChannelHandler) getChannels()
                            .get(channelHandlerLookupName(channelName));
                    channel.reconnect();
                }
            }
            notifyCompositeBeastChannelListeners();
        }

        @Override
        public void serverTimeout(AlarmClientModel model) {
            log.warning("beast datasource: server timeout (server alive: " + model.isServerAlive() + ")");
            synchronized (parent.channelConsumers) {
                for (String channelName : parent.channelConsumers.keySet()) {
                    BeastChannelHandler channel = (BeastChannelHandler) getChannels()
                            .get(channelHandlerLookupName(channelName));
                    // notify the ChannelHandler that we lost connection
                    // (causes ConnectionChanged event + Listeners' PVReader.isConnected() will return the correct state)
                    channel.connectionStateChanged(false);
                }
            }
        }

        @Override
        public void serverModeUpdate(AlarmClientModel model, boolean maintenance_mode) {
            log.fine("beast  datasource: server mode update");
        }

        @SuppressWarnings({ "rawtypes", "unchecked" })
        @Override
        public void newAlarmState(AlarmClientModel alarmModel, AlarmTreePV pv, boolean parent_changed) {
            if (pv != null) {
                log.fine(pv.getPathName());
                synchronized (parent.channelConsumers) {
                    parent.channelConsumers.forEach((key, pathHandlers) -> {
                        if (BeastTypeSupport.getStrippedChannelName(key).equals(pv.getPathName().substring(1))
                                || BeastTypeSupport.getStrippedChannelName(key).equals(pv.getName())) {
                            if (pathHandlers != null) {
                                for (Consumer consumer : pathHandlers) {
                                    consumer.accept(pv);
                                }
                            }
                        }
                    });
                }
                // Notify parent nodes (regardless of parent_changed - because the parents' AlarmPVsCount)
                AlarmTreeItem parent = pv.getParent();
                while (parent != null) {
                    final String parentPath = parent.getPathName();
                    synchronized (this.parent.channelConsumers) {
                        this.parent.channelConsumers.forEach((key, pathHandlers) -> {
                            if (BeastTypeSupport.getStrippedChannelName(key).equals(parentPath.substring(1))) {
                                if (pathHandlers != null) {
                                    for (Consumer consumer : pathHandlers) {
                                        try {
                                            consumer.accept(getState(parentPath));
                                        } catch (Exception e) {
                                            e.printStackTrace();
                                        }
                                    }
                                }
                            }
                        });
                    }
                    parent = parent.getParent();
                }
            } else {
                // The AlarmClientModel has recovered from a disconnection or is notifying us that the first
                // messages have been received after initial connection.
                synchronized (parent.channelConsumers) {
                    for (String channelName : parent.channelConsumers.keySet()) {
                        BeastChannelHandler channel = (BeastChannelHandler) getChannels()
                                .get(channelHandlerLookupName(channelName));
                        if (channel!=null)
                            channel.reconnect(); // will send connection state + current AlarmTreeItem state
                    }
                }
            }
            notifyCompositeBeastChannelListeners();
        }

        private void notifyCompositeBeastChannelListeners() {
            synchronized (parent.channelConsumers) {
                final BeastChannelHandler compositeChannel = (BeastChannelHandler) getChannels()
                        .get(channelHandlerLookupName(parent.compositeModel.getConfigurationName()));
                if (compositeChannel != null) {<commit_msg>Fix for deadlock after stopping and restarting IOCs.<commit_after>/**
 * Copyright (C) 2010-14 diirt developers. See COPYRIGHT.TXT
 * All rights reserved. Use is subject to license terms. See LICENSE.TXT
 */
package org.csstudio.alarm.diirt.datasource;

import java.net.URI;
import java.net.URLDecoder;
import java.net.URLEncoder;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.csstudio.alarm.beast.client.AlarmTreeItem;
import org.csstudio.alarm.beast.client.AlarmTreePV;
import org.csstudio.alarm.beast.ui.clientmodel.AlarmClientModel;
import org.csstudio.alarm.beast.ui.clientmodel.AlarmClientModelConfigListener;
import org.csstudio.alarm.beast.ui.clientmodel.AlarmClientModelListener;
import org.diirt.datasource.ChannelHandler;
import org.diirt.datasource.DataSource;
import org.diirt.datasource.vtype.DataTypeSupport;
import org.eclipse.core.runtime.Platform;

/**
 * @author <NAME>
 *
 */
public class BeastDataSource extends DataSource implements AlarmClientModelConfigListener {
    private static final Logger log = Logger.getLogger(BeastDataSource.class.getName());

    private static final String PREF_QUALIFIER = "org.csstudio.alarm.diirt.datasource";
    private static final String PREF_KEY_COMPOSITE = "composite.model.name";
    private static final String PREF_DEFAULT_COMPOSITE_MODEL_NAME = "Composite";

    // The model, activeAlarms and acknowledgedAlarms is shared by the entire
    // datasource, the benefit of does this at the datasource level instead of
    // in each channel is that they need to be computed only once and only a single
    // copy needs to be maintained.
    private AlarmClientModel model;

    private Map<String, List<Consumer>> channelConsumers = Collections.synchronizedMap(new HashMap<String, List<Consumer>>());
    private Map<String, AlarmClientModel> models = Collections.synchronizedMap(new HashMap<String, AlarmClientModel>());
    private AtomicInteger loaded;
    private AtomicInteger toLoad;

    private Executor executor = Executors.newScheduledThreadPool(4);

    private BeastTypeSupport typeSupport;

    private BeastDataSourceConfiguration configuration;

    private BeastAlarmClientModelListener modelListener;

    private CompositeAlarmClientModel compositeModel;

    static {
        // Install type support for the types it generates.
        DataTypeSupport.install();
    }

    private class BeastAlarmClientModelListener implements AlarmClientModelListener{
        private BeastDataSource parent;

        BeastAlarmClientModelListener(BeastDataSource parent) {
            this.parent = parent;
        }

        @Override
        public void newAlarmConfiguration(AlarmClientModel model) {
            log.config("beast  datasource: new alarm configuration --- " + model);
            // new model loaded
            parent.loaded.incrementAndGet();
            if (!parent.compositeModel.equals(model)) parent.compositeModel.addAlarmClientModel(model);
            if (parent.loaded.get() == parent.toLoad.get()) {
                parent.compositeModel.setAllLoaded(true);
                log.log(Level.FINE, "All models loaded - notifying Composite model listeners.");
                parent.compositeModel.compositeModelsLoaded();
            }
            synchronized (parent.channelConsumers) {
                for (String channelName : parent.channelConsumers.keySet()) {
                    BeastChannelHandler channel = (BeastChannelHandler) getChannels()
                            .get(channelHandlerLookupName(channelName));
                    channel.reconnect();
                }
            }
            notifyCompositeBeastChannelListeners();
        }

        @Override
        public void serverTimeout(AlarmClientModel model) {
            log.warning("beast datasource: server timeout (server alive: " + model.isServerAlive() + ")");
            synchronized (parent.channelConsumers) {
                for (String channelName : parent.channelConsumers.keySet()) {
                    BeastChannelHandler channel = (BeastChannelHandler) getChannels()
                            .get(channelHandlerLookupName(channelName));
                    // notify the ChannelHandler that we lost connection
                    // (causes ConnectionChanged event + Listeners' PVReader.isConnected() will return the correct state)
                    channel.connectionStateChanged(false);
                }
            }
        }

        @Override
        public void serverModeUpdate(AlarmClientModel model, boolean maintenance_mode) {
            log.fine("beast  datasource: server mode update");
        }

        @SuppressWarnings({ "rawtypes", "unchecked" })
        @Override
        public void newAlarmState(AlarmClientModel alarmModel, AlarmTreePV pv, boolean parent_changed) {
            if (pv != null) {
                log.fine(pv.getPathName());
                synchronized (parent.channelConsumers) {
                    parent.channelConsumers.forEach((key, pathHandlers) -> {
                        if (BeastTypeSupport.getStrippedChannelName(key).equals(pv.getPathName().substring(1))
                                || BeastTypeSupport.getStrippedChannelName(key).equals(pv.getName())) {
                            if (pathHandlers != null) {
                                for (Consumer consumer : pathHandlers) {
                                    consumer.accept(pv);
                                }
                            }
                        }
                    });
                }
                // Notify parent nodes (regardless of parent_changed - because the parents' AlarmPVsCount)
                AlarmTreeItem parent = pv.getParent();
                while (parent != null) {
                    final String parentPath = parent.getPathName();
                    synchronized (this.parent.channelConsumers) {
                        this.parent.channelConsumers.forEach((key, pathHandlers) -> {
                            if (BeastTypeSupport.getStrippedChannelName(key).equals(parentPath.substring(1))) {
                                if (pathHandlers != null) {
                                    for (Consumer consumer : pathHandlers) {
                                        try {
                                            consumer.accept(getState(parentPath));
                                        } catch (Exception e) {
                                            e.printStackTrace();
                                        }
                                    }
                                }
                            }
                        });
                    }
                    parent = parent.getParent();
                }
            } else {
                // The AlarmClientModel has recovered from a disconnection or is notifying us that the first
                // messages have been received after initial connection.
                    for (String channelName : parent.channelConsumers.keySet()) {
                        BeastChannelHandler channel = (BeastChannelHandler) getChannels()
                                .get(channelHandlerLookupName(channelName));
                        if (channel!=null) {
                            synchronized (channel) {
                                channel.reconnect(); // will send connection state + current AlarmTreeItem state
                            }
                        }
                    }
            }
            notifyCompositeBeastChannelListeners();
        }

        private void notifyCompositeBeastChannelListeners() {
            synchronized (parent.channelConsumers) {
                final BeastChannelHandler compositeChannel = (BeastChannelHandler) getChannels()
                        .get(channelHandlerLookupName(parent.compositeModel.getConfigurationName()));
                if (compositeChannel != null) {
{'fraction_non_alphanumeric': 0.08030128166190852, 'fraction_numerical': 0.0010933608698293143, 'mean_word_length': 1.988021778584392, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 28, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '989656', 'n_tokens_mistral': 3976, 'n_tokens_neox': 3817, 'n_words': 873}