<commit_before>/*
 * Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * WSO2 Inc. licenses this file to you 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.receiver.core.internal.type.wso2event;

import org.wso2.carbon.databridge.commons.Attribute;
import org.wso2.carbon.databridge.commons.Event;
import org.wso2.carbon.databridge.commons.StreamDefinition;
import org.wso2.carbon.event.receiver.core.InputMapper;
import org.wso2.carbon.event.receiver.core.config.EventReceiverConfiguration;
import org.wso2.carbon.event.receiver.core.config.EventReceiverConstants;
import org.wso2.carbon.event.receiver.core.config.InputMapping;
import org.wso2.carbon.event.receiver.core.config.InputMappingAttribute;
import org.wso2.carbon.event.receiver.core.config.mapping.WSO2EventInputMapping;
import org.wso2.carbon.event.receiver.core.exception.EventReceiverConfigurationException;
import org.wso2.carbon.event.receiver.core.exception.EventReceiverProcessingException;
import org.wso2.carbon.event.receiver.core.exception.EventReceiverStreamValidationException;
import org.wso2.carbon.event.receiver.core.internal.ds.EventReceiverServiceValueHolder;
import org.wso2.carbon.event.receiver.core.internal.util.EventReceiverUtil;
import org.wso2.carbon.event.receiver.core.internal.util.helper.EventReceiverConfigurationHelper;
import org.wso2.carbon.event.stream.core.EventStreamService;
import org.wso2.carbon.event.stream.core.exception.EventStreamConfigurationException;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;

public class WSO2EventInputMapper implements InputMapper {
    private EventReceiverConfiguration eventReceiverConfiguration = null;
    private StreamDefinition exportedStreamDefinition = null;
    private StreamDefinition importedStreamDefinition = null;
    private Map<InputDataType, int[]> inputDataTypeSpecificPositionMap = null;
    private boolean arbitraryMapsEnabled = false;
    private int inputStreamSize = 0;
    private int outMetaSize = 0, outCorrelationSize = 0, outPayloadSize = 0;

    public WSO2EventInputMapper(EventReceiverConfiguration eventReceiverConfiguration,
                                StreamDefinition exportedStreamDefinition)
            throws EventReceiverConfigurationException {
        //TODO It is better if logic to check from stream definition store can be moved outside of input mapper.
        EventStreamService eventStreamService = EventReceiverServiceValueHolder.getEventStreamService();

        this.eventReceiverConfiguration = eventReceiverConfiguration;
        WSO2EventInputMapping wso2EventInputMapping = (WSO2EventInputMapping) this.eventReceiverConfiguration.getInputMapping();

        String fromStreamName = wso2EventInputMapping.getFromEventName();
        String fromStreamVersion = wso2EventInputMapping.getFromEventVersion();
        this.arbitraryMapsEnabled = wso2EventInputMapping.isArbitraryMapsEnabled();

        if (fromStreamName == null || fromStreamVersion == null || (fromStreamName.isEmpty()) || (fromStreamVersion.isEmpty())) {
            importedStreamDefinition = exportedStreamDefinition;
        } else {
            try {
                this.importedStreamDefinition = eventStreamService.getStreamDefinition(fromStreamName, fromStreamVersion);
            } catch (EventStreamConfigurationException e) {
                throw new EventReceiverStreamValidationException("Error while retrieving stream definition : " + e.getMessage(), fromStreamName + ":" + fromStreamVersion);
            }
        }
        this.exportedStreamDefinition = exportedStreamDefinition;
        boolean candidateForArbitraryMaps = isCandidateForArbitraryAttributes(eventReceiverConfiguration.getInputMapping(), importedStreamDefinition);
        if (!candidateForArbitraryMaps) {
            validateInputStreamAttributes();
        }

        if (importedStreamDefinition != null && eventReceiverConfiguration.getInputMapping().isCustomMappingEnabled() && !candidateForArbitraryMaps) {
            this.inputDataTypeSpecificPositionMap = new HashMap<InputDataType, int[]>();
            Map<Integer, Integer> payloadDataMap = new TreeMap<Integer, Integer>();
            Map<Integer, Integer> metaDataMap = new TreeMap<Integer, Integer>();
            Map<Integer, Integer> correlationDataMap = new TreeMap<Integer, Integer>();

            List<Attribute> allAttributes = new ArrayList<Attribute>();
            if (importedStreamDefinition.getMetaData() != null && !importedStreamDefinition.getMetaData().isEmpty()) {
                allAttributes.addAll(importedStreamDefinition.getMetaData());
            }
            if (importedStreamDefinition.getCorrelationData() != null && !importedStreamDefinition.getCorrelationData().isEmpty()) {
                allAttributes.addAll(importedStreamDefinition.getCorrelationData());
            }
            if (importedStreamDefinition.getPayloadData() != null && !importedStreamDefinition.getPayloadData().isEmpty()) {
                allAttributes.addAll(importedStreamDefinition.getPayloadData());
            }<commit_msg>Fixing NPE when wso2event custom mapping import stream is not available<commit_after>/*
 * Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * WSO2 Inc. licenses this file to you 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.receiver.core.internal.type.wso2event;

import org.wso2.carbon.databridge.commons.Attribute;
import org.wso2.carbon.databridge.commons.Event;
import org.wso2.carbon.databridge.commons.StreamDefinition;
import org.wso2.carbon.event.receiver.core.InputMapper;
import org.wso2.carbon.event.receiver.core.config.EventReceiverConfiguration;
import org.wso2.carbon.event.receiver.core.config.EventReceiverConstants;
import org.wso2.carbon.event.receiver.core.config.InputMapping;
import org.wso2.carbon.event.receiver.core.config.InputMappingAttribute;
import org.wso2.carbon.event.receiver.core.config.mapping.WSO2EventInputMapping;
import org.wso2.carbon.event.receiver.core.exception.EventReceiverConfigurationException;
import org.wso2.carbon.event.receiver.core.exception.EventReceiverProcessingException;
import org.wso2.carbon.event.receiver.core.exception.EventReceiverStreamValidationException;
import org.wso2.carbon.event.receiver.core.internal.ds.EventReceiverServiceValueHolder;
import org.wso2.carbon.event.receiver.core.internal.util.EventReceiverUtil;
import org.wso2.carbon.event.receiver.core.internal.util.helper.EventReceiverConfigurationHelper;
import org.wso2.carbon.event.stream.core.EventStreamService;
import org.wso2.carbon.event.stream.core.exception.EventStreamConfigurationException;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;

public class WSO2EventInputMapper implements InputMapper {
    private EventReceiverConfiguration eventReceiverConfiguration = null;
    private StreamDefinition exportedStreamDefinition = null;
    private StreamDefinition importedStreamDefinition = null;
    private Map<InputDataType, int[]> inputDataTypeSpecificPositionMap = null;
    private boolean arbitraryMapsEnabled = false;
    private int inputStreamSize = 0;
    private int outMetaSize = 0, outCorrelationSize = 0, outPayloadSize = 0;

    public WSO2EventInputMapper(EventReceiverConfiguration eventReceiverConfiguration,
                                StreamDefinition exportedStreamDefinition)
            throws EventReceiverConfigurationException {
        //TODO It is better if logic to check from stream definition store can be moved outside of input mapper.
        EventStreamService eventStreamService = EventReceiverServiceValueHolder.getEventStreamService();

        this.eventReceiverConfiguration = eventReceiverConfiguration;
        WSO2EventInputMapping wso2EventInputMapping = (WSO2EventInputMapping) this.eventReceiverConfiguration.getInputMapping();

        String fromStreamName = wso2EventInputMapping.getFromEventName();
        String fromStreamVersion = wso2EventInputMapping.getFromEventVersion();
        this.arbitraryMapsEnabled = wso2EventInputMapping.isArbitraryMapsEnabled();

        if (fromStreamName == null || fromStreamVersion == null || (fromStreamName.isEmpty()) || (fromStreamVersion.isEmpty())) {
            importedStreamDefinition = exportedStreamDefinition;
        } else {
            try {
                this.importedStreamDefinition = eventStreamService.getStreamDefinition(fromStreamName, fromStreamVersion);
            } catch (EventStreamConfigurationException e) {
                throw new EventReceiverStreamValidationException("Error while retrieving stream definition : " + e.getMessage(), fromStreamName + ":" + fromStreamVersion);
            }
        }
        this.exportedStreamDefinition = exportedStreamDefinition;
        if (importedStreamDefinition == null) {
            // Import stream for custom mapping is not available
            String fromStream = fromStreamName + EventReceiverConstants.STREAM_NAME_VER_DELIMITER + fromStreamVersion;
            throw new EventReceiverStreamValidationException("Imported stream " + fromStream + " is not available", fromStream);
        }
        boolean candidateForArbitraryMaps = isCandidateForArbitraryAttributes(eventReceiverConfiguration.getInputMapping(), importedStreamDefinition);
        if (!candidateForArbitraryMaps) {
            validateInputStreamAttributes();
        }

        if (importedStreamDefinition != null && eventReceiverConfiguration.getInputMapping().isCustomMappingEnabled() && !candidateForArbitraryMaps) {
            this.inputDataTypeSpecificPositionMap = new HashMap<InputDataType, int[]>();
            Map<Integer, Integer> payloadDataMap = new TreeMap<Integer, Integer>();
            Map<Integer, Integer> metaDataMap = new TreeMap<Integer, Integer>();
            Map<Integer, Integer> correlationDataMap = new TreeMap<Integer, Integer>();

            List<Attribute> allAttributes = new ArrayList<Attribute>();
            if (importedStreamDefinition.getMetaData() != null && !importedStreamDefinition.getMetaData().isEmpty()) {
                allAttributes.addAll(importedStreamDefinition.getMetaData());
            }
            if (importedStreamDefinition.getCorrelationData() != null && !importedStreamDefinition.getCorrelationData().isEmpty()) {
                allAttributes.addAll(importedStreamDefinition.getCorrelationData());
            }
            if (importedStreamDefinition.getPayloadData() != null && !importedStreamDefinition.getPayloadData().isEmpty()) {
                allAttributes.addAll(importedStreamDefinition.getPayloadData());
            }
{'fraction_non_alphanumeric': 0.08551129155036792, 'fraction_numerical': 0.007358538442019792, 'mean_word_length': 4.269162210338681, 'pattern_counts': {'":': 2, '<': 23, '<?xml version=': 0, '>': 23, 'https://': 0, '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': '1572602', 'n_tokens_mistral': 3016, 'n_tokens_neox': 2934, 'n_words': 641}