<commit_before>/*
 * Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy
 * of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed
 * under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
 * CONDITIONS OF ANY KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations under the License.
 */
package org.wso2.carbon.event.stream.core.internal;

import org.apache.axis2.engine.AxisConfiguration;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.wso2.carbon.base.MultitenantConstants;
import org.wso2.carbon.context.PrivilegedCarbonContext;
import org.wso2.carbon.core.multitenancy.utils.TenantAxisUtils;
import org.wso2.carbon.databridge.commons.Event;
import org.wso2.carbon.databridge.commons.StreamDefinition;
import org.wso2.carbon.databridge.commons.utils.DataBridgeCommonsUtils;
import org.wso2.carbon.event.stream.core.*;
import org.wso2.carbon.event.stream.core.exception.EventStreamConfigurationException;
import org.wso2.carbon.event.stream.core.exception.StreamDefinitionAlreadyDefinedException;
import org.wso2.carbon.event.stream.core.internal.ds.EventStreamServiceValueHolder;
import org.wso2.carbon.event.stream.core.internal.util.EventStreamConstants;
import org.wso2.carbon.event.stream.core.internal.util.SampleEventGenerator;
import org.wso2.carbon.event.stream.core.internal.util.helper.EventStreamConfigurationFileSystemInvoker;

import java.io.File;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

public class CarbonEventStreamService implements EventStreamService {

    private static final Log log = LogFactory.getLog(CarbonEventStreamService.class);
    private Map<Integer, Map<String, EventStreamConfiguration>> tenantSpecificEventStreamConfigs = new ConcurrentHashMap<Integer, Map<String, EventStreamConfiguration>>();

    public void removeEventStreamConfigurationFromMap(String fileName) {
        int tenantId = PrivilegedCarbonContext.getThreadLocalCarbonContext().getTenantId();
        Map<String, EventStreamConfiguration> eventStreamConfigs = tenantSpecificEventStreamConfigs.get(tenantId);
        String streamId = null;
        if (eventStreamConfigs != null) {
            for (EventStreamConfiguration eventStreamConfiguration : eventStreamConfigs.values()) {
                if (eventStreamConfiguration.getFileName().equals(fileName)) {
                    streamId = eventStreamConfiguration.getStreamDefinition().getStreamId();
                    break;
                }
            }
        }
        if (streamId != null) {
            eventStreamConfigs.remove(streamId);
            for (EventStreamListener eventStreamListener : EventStreamServiceValueHolder.getEventStreamListenerList()) {
                eventStreamListener.removedEventStream(tenantId,
                        DataBridgeCommonsUtils.getStreamNameFromStreamId(streamId),
                        DataBridgeCommonsUtils.getStreamVersionFromStreamId(streamId));
            }
        }
    }

    /**
     * @param name
     * @param version
     * @return
     */
    @Override
    public StreamDefinition getStreamDefinition(String name, String version) throws EventStreamConfigurationException {
        int tenantId = PrivilegedCarbonContext.getThreadLocalCarbonContext().getTenantId();
        Map<String, EventStreamConfiguration> eventStreamConfigs = tenantSpecificEventStreamConfigs.get(tenantId);
        if (eventStreamConfigs != null && eventStreamConfigs.containsKey(name + ":" + version)) {
            return eventStreamConfigs.get(name + ":" + version).getStreamDefinition();
        }
        return null;
    }

    /**
     * @param streamId
     * @return StreamDefinition and returns null if ont exist
     */
    @Override
    public StreamDefinition getStreamDefinition(String streamId) throws EventStreamConfigurationException {
        int tenantId = PrivilegedCarbonContext.getThreadLocalCarbonContext().getTenantId();
        Map<String, EventStreamConfiguration> eventStreamConfigs = tenantSpecificEventStreamConfigs.get(tenantId);
        if (eventStreamConfigs != null && eventStreamConfigs.containsKey(streamId)) {
            return eventStreamConfigs.get(streamId).getStreamDefinition();
        }
        return null;
    }

    public EventStreamConfiguration getEventStreamConfiguration(String streamId) {
        int tenantId = PrivilegedCarbonContext.getThreadLocalCarbonContext().getTenantId();
        Map<String, EventStreamConfiguration> eventStreamConfigs = tenantSpecificEventStreamConfigs.get(tenantId);
        if (eventStreamConfigs != null && eventStreamConfigs.containsKey(streamId)) {
            return eventStreamConfigs.get(streamId);
        }
        return null;
    }

    @Override
    public List<StreamDefinition> getAllStreamDefinitions() throws EventStreamConfigurationException {
        int tenantId = PrivilegedCarbonContext.getThreadLocalCarbonContext().getTenantId();
        Map<String, EventStreamConfiguration> eventStreamConfigs = tenantSpecificEventStreamConfigs.get(tenantId);
        List<StreamDefinition> list = new ArrayList<StreamDefinition>();
        if (eventStreamConfigs == null) {
            return list;
        }
        for (EventStreamConfiguration eventStreamConfiguration : eventStreamConfigs.values()) {
            list.add(eventStreamConfiguration.getStreamDefinition());
        }
        return list;
    }

    @Override
    public List<EventStreamConfiguration> getAllEventStreamConfigurations() throws EventStreamConfigurationException {
        int tenantId = PrivilegedCarbonContext.getThreadLocalCarbonContext().getTenantId();
        Map<String, EventStreamConfiguration> eventStreamConfigs = tenantSpecificEventStreamConfigs.get(tenantId);
        if (eventStreamConfigs == null) {
            return new ArrayList<EventStreamConfiguration>();
        }
        return new ArrayList<EventStreamConfiguration>(eventStreamConfigs.values());
    }

    public void addEventStreamConfig(EventStreamConfiguration eventStreamConfiguration)
            throws EventStreamConfigurationException {
        int tenantId = PrivilegedCarbonContext.getThreadLocalCarbonContext().getTenantId();
        Map<String, EventStreamConfiguration> eventStreamConfigs = tenantSpecificEventStreamConfigs.get(tenantId);
        if (eventStreamConfigs == null) {
            eventStreamConfigs = new HashMap<String, EventStreamConfiguration>();
            tenantSpecificEventStreamConfigs.put(tenantId, eventStreamConfigs);
        }
        eventStreamConfigs.put(eventStreamConfiguration.getStreamDefinition().getStreamId(), eventStreamConfiguration);
        for (EventStreamListener eventStreamListener : EventStreamServiceValueHolder.getEventStreamListenerList()) {
            eventStreamListener.addedEventStream(tenantId,
                    eventStreamConfiguration.getStreamDefinition().getName(),
                    eventStreamConfiguration.getStreamDefinition().getVersion());
        }<commit_msg>Committing the fixes for  https://wso2.org/jira/browse/CEP-1021<commit_after>/*
 * Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy
 * of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed
 * under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
 * CONDITIONS OF ANY KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations under the License.
 */
package org.wso2.carbon.event.stream.core.internal;

import org.apache.axis2.engine.AxisConfiguration;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.wso2.carbon.base.MultitenantConstants;
import org.wso2.carbon.context.PrivilegedCarbonContext;
import org.wso2.carbon.core.multitenancy.utils.TenantAxisUtils;
import org.wso2.carbon.databridge.commons.Event;
import org.wso2.carbon.databridge.commons.StreamDefinition;
import org.wso2.carbon.databridge.commons.utils.DataBridgeCommonsUtils;
import org.wso2.carbon.event.stream.core.*;
import org.wso2.carbon.event.stream.core.exception.EventStreamConfigurationException;
import org.wso2.carbon.event.stream.core.exception.StreamDefinitionAlreadyDefinedException;
import org.wso2.carbon.event.stream.core.internal.ds.EventStreamServiceValueHolder;
import org.wso2.carbon.event.stream.core.internal.util.EventStreamConstants;
import org.wso2.carbon.event.stream.core.internal.util.SampleEventGenerator;
import org.wso2.carbon.event.stream.core.internal.util.helper.EventStreamConfigurationFileSystemInvoker;

import java.io.File;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

public class CarbonEventStreamService implements EventStreamService {

    private static final Log log = LogFactory.getLog(CarbonEventStreamService.class);
    private Map<Integer, TreeMap<String, EventStreamConfiguration>> tenantSpecificEventStreamConfigs = new ConcurrentHashMap<Integer, TreeMap<String, EventStreamConfiguration>>();

    public void removeEventStreamConfigurationFromMap(String fileName) {
        int tenantId = PrivilegedCarbonContext.getThreadLocalCarbonContext().getTenantId();
        Map<String, EventStreamConfiguration> eventStreamConfigs = tenantSpecificEventStreamConfigs.get(tenantId);
        String streamId = null;
        if (eventStreamConfigs != null) {
            for (EventStreamConfiguration eventStreamConfiguration : eventStreamConfigs.values()) {
                if (eventStreamConfiguration.getFileName().equals(fileName)) {
                    streamId = eventStreamConfiguration.getStreamDefinition().getStreamId();
                    break;
                }
            }
        }
        if (streamId != null) {
            eventStreamConfigs.remove(streamId);
            for (EventStreamListener eventStreamListener : EventStreamServiceValueHolder.getEventStreamListenerList()) {
                eventStreamListener.removedEventStream(tenantId,
                        DataBridgeCommonsUtils.getStreamNameFromStreamId(streamId),
                        DataBridgeCommonsUtils.getStreamVersionFromStreamId(streamId));
            }
        }
    }

    /**
     * @param name
     * @param version
     * @return
     */
    @Override
    public StreamDefinition getStreamDefinition(String name, String version) throws EventStreamConfigurationException {
        int tenantId = PrivilegedCarbonContext.getThreadLocalCarbonContext().getTenantId();
        Map<String, EventStreamConfiguration> eventStreamConfigs = tenantSpecificEventStreamConfigs.get(tenantId);
        if (eventStreamConfigs != null && eventStreamConfigs.containsKey(name + ":" + version)) {
            return eventStreamConfigs.get(name + ":" + version).getStreamDefinition();
        }
        return null;
    }

    /**
     * @param streamId
     * @return StreamDefinition and returns null if ont exist
     */
    @Override
    public StreamDefinition getStreamDefinition(String streamId) throws EventStreamConfigurationException {
        int tenantId = PrivilegedCarbonContext.getThreadLocalCarbonContext().getTenantId();
        Map<String, EventStreamConfiguration> eventStreamConfigs = tenantSpecificEventStreamConfigs.get(tenantId);
        if (eventStreamConfigs != null && eventStreamConfigs.containsKey(streamId)) {
            return eventStreamConfigs.get(streamId).getStreamDefinition();
        }
        return null;
    }

    public EventStreamConfiguration getEventStreamConfiguration(String streamId) {
        int tenantId = PrivilegedCarbonContext.getThreadLocalCarbonContext().getTenantId();
        Map<String, EventStreamConfiguration> eventStreamConfigs = tenantSpecificEventStreamConfigs.get(tenantId);
        if (eventStreamConfigs != null && eventStreamConfigs.containsKey(streamId)) {
            return eventStreamConfigs.get(streamId);
        }
        return null;
    }

    @Override
    public List<StreamDefinition> getAllStreamDefinitions() throws EventStreamConfigurationException {
        int tenantId = PrivilegedCarbonContext.getThreadLocalCarbonContext().getTenantId();
        Map<String, EventStreamConfiguration> eventStreamConfigs = tenantSpecificEventStreamConfigs.get(tenantId);
        List<StreamDefinition> list = new ArrayList<StreamDefinition>();
        if (eventStreamConfigs == null) {
            return list;
        }
        for (EventStreamConfiguration eventStreamConfiguration : eventStreamConfigs.values()) {
            list.add(eventStreamConfiguration.getStreamDefinition());
        }
        return list;
    }

    @Override
    public List<EventStreamConfiguration> getAllEventStreamConfigurations() throws EventStreamConfigurationException {
        int tenantId = PrivilegedCarbonContext.getThreadLocalCarbonContext().getTenantId();
        Map<String, EventStreamConfiguration> eventStreamConfigs = tenantSpecificEventStreamConfigs.get(tenantId);
        if (eventStreamConfigs == null) {
            return new ArrayList<EventStreamConfiguration>();
        }
        return new ArrayList<EventStreamConfiguration>(eventStreamConfigs.values());
    }

    public void addEventStreamConfig(EventStreamConfiguration eventStreamConfiguration)
            throws EventStreamConfigurationException {
        int tenantId = PrivilegedCarbonContext.getThreadLocalCarbonContext().getTenantId();
        TreeMap<String, EventStreamConfiguration> eventStreamConfigs = tenantSpecificEventStreamConfigs.get(tenantId);
        if (eventStreamConfigs == null) {
            eventStreamConfigs = new TreeMap<String, EventStreamConfiguration>();
            tenantSpecificEventStreamConfigs.put(tenantId, eventStreamConfigs);
        }
        eventStreamConfigs.put(eventStreamConfiguration.getStreamDefinition().getStreamId(), eventStreamConfiguration);
        for (EventStreamListener eventStreamListener : EventStreamServiceValueHolder.getEventStreamListenerList()) {
            eventStreamListener.addedEventStream(tenantId,
                    eventStreamConfiguration.getStreamDefinition().getName(),
                    eventStreamConfiguration.getStreamDefinition().getVersion());
        }
{'fraction_non_alphanumeric': 0.08334464503868604, 'fraction_numerical': 0.003732862766390661, 'mean_word_length': 3.976359338061466, 'pattern_counts': {'":': 4, '<': 39, '<?xml version=': 0, '>': 39, 'https://': 1, 'lorem ipsum': 0, 'www.': 4, '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': '4830181', 'n_tokens_mistral': 3713, 'n_tokens_neox': 3630, 'n_words': 702}