<commit_before>/*
 * Copyright (c) 2014, 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.andes.kernel.distruptor.inbound;

import com.lmax.disruptor.EventHandler;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.wso2.andes.amqp.AMQPUtils;
import org.wso2.andes.kernel.*;
import org.wso2.andes.server.ClusterResourceHolder;
import org.wso2.andes.server.store.MessageMetaDataType;
import org.wso2.andes.subscription.SubscriptionStore;

import java.util.HashSet;
import java.util.List;
import java.util.Set;

/**
 * This event processor goes through the ring buffer first and update AndesMessage data event objects.
 * NOTE: Only one instance of this processor should process events from the ring buffer
 */
public class MessagePreProcessor implements EventHandler<InboundEventContainer> {

    private static final Log log = LogFactory.getLog(MessagePreProcessor.class);
    private final SubscriptionStore subscriptionStore;
    private final MessageIDGenerator idGenerator;

    public MessagePreProcessor(SubscriptionStore subscriptionStore) {
        this.subscriptionStore = subscriptionStore;
        idGenerator = new MessageIDGenerator();
    }

    @Override
    public void onEvent(InboundEventContainer inboundEvent, long sequence, boolean endOfBatch ) throws Exception {
        switch (inboundEvent.getEventType()) {
            case MESSAGE_EVENT:
                updateRoutingInformation(inboundEvent, sequence);
                break;
            case SAFE_ZONE_DECLARE_EVENT:
                setSafeZoneLimit(inboundEvent, sequence);
                break;
        }
    }

    /**
     * Calculate the current safe zone for this node (using the last generated message ID)
     * @param event event
     * @param sequence position of the event at the event ring buffer
     */
    private void setSafeZoneLimit(InboundEventContainer event, long sequence) {
        long safeZoneLimit = idGenerator.getNextId();
        event.setSafeZoneLimit(safeZoneLimit);
        if(log.isDebugEnabled()){
            log.debug("[ Sequence " + sequence + " ] Pre processing message. Setting the Safe Zone " + safeZoneLimit);
        }
    }

    /**
     * Route the message to queue/queues of subscribers matching in AMQP way. Hierarchical topic message routing is
     * evaluated here. This will duplicate message for each "subscription destination (not message destination)" at
     * different nodes
     *
     * @param event InboundEventContainer containing the message list
     */
    private void updateRoutingInformation(InboundEventContainer event, long sequence) {

        // NOTE: This is the MESSAGE_EVENT and this is the first processing event for this message published to ring
        // Therefore there should be exactly one message in the list.
        // NO NEED TO CHECK FOR LIST SIZE
        AndesMessage message = event.messageList.get(0);

        AndesChannel andesChannel = event.getChannel();

        // Messages are processed in the order they arrive at ring buffer By this processor.
        // By setting message ID through message pre processor we assure, even in a multi publisher scenario, there is
        // no message id ordering issue at node level.
        setMessageID(message);

        if(log.isDebugEnabled()){
            log.debug("[ Sequence " + sequence + " ] Pre processing message. Message ID "
                    + message.getMetadata().getMessageID());
        }

        if (message.getMetadata().isTopic()) {
            handleTopicRoutine(event, message, andesChannel);
        } else {
            andesChannel.recordAdditionToBuffer(message.getContentChunkList().size());
            AndesMessageMetadata messageMetadata = message.getMetadata();
            messageMetadata.setStorageQueueName(messageMetadata.getDestination());
        }
    }

    private void handleTopicRoutine(InboundEventContainer event, AndesMessage message, AndesChannel andesChannel) {
        String messageRoutingKey = message.getMetadata().getDestination();
        //get all topic subscriptions in the cluster matching to routing key
        //including hierarchical topic case
        Set<AndesSubscription> subscriptionList;
        try {

            // Get subscription list according to the message type
            AndesSubscription.SubscriptionType subscriptionType;

            if (MessageMetaDataType.META_DATA_MQTT == message.getMetadata().getMetaDataType()) {
                subscriptionType = AndesSubscription.SubscriptionType.MQTT;
            } else {
                subscriptionType = AndesSubscription.SubscriptionType.AMQP;
            }

            subscriptionList = subscriptionStore.getClusterSubscribersForDestination(messageRoutingKey, true, subscriptionType);

            // current message is removed from list and updated cloned messages added later
            event.messageList.clear();
            boolean isMessageRouted = false;
            Set<String> alreadyStoredQueueNames = new HashSet<String>();
            for (AndesSubscription subscription : subscriptionList) {
                if (!alreadyStoredQueueNames.contains(subscription.getStorageQueueName())) {
                    AndesMessage clonedMessage = cloneAndesMessageMetadataAndContent(message);<commit_msg>// Avoid adding QOS 0 MQTT messages to clean session = false subscribers if disconnected.<commit_after>/*
 * Copyright (c) 2014, 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.andes.kernel.distruptor.inbound;

import com.lmax.disruptor.EventHandler;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.wso2.andes.amqp.AMQPUtils;
import org.wso2.andes.kernel.*;
import org.wso2.andes.server.ClusterResourceHolder;
import org.wso2.andes.server.store.MessageMetaDataType;
import org.wso2.andes.subscription.SubscriptionStore;

import java.util.HashSet;
import java.util.List;
import java.util.Set;

/**
 * This event processor goes through the ring buffer first and update AndesMessage data event objects.
 * NOTE: Only one instance of this processor should process events from the ring buffer
 */
public class MessagePreProcessor implements EventHandler<InboundEventContainer> {

    private static final Log log = LogFactory.getLog(MessagePreProcessor.class);
    private final SubscriptionStore subscriptionStore;
    private final MessageIDGenerator idGenerator;

    public MessagePreProcessor(SubscriptionStore subscriptionStore) {
        this.subscriptionStore = subscriptionStore;
        idGenerator = new MessageIDGenerator();
    }

    @Override
    public void onEvent(InboundEventContainer inboundEvent, long sequence, boolean endOfBatch ) throws Exception {
        switch (inboundEvent.getEventType()) {
            case MESSAGE_EVENT:
                updateRoutingInformation(inboundEvent, sequence);
                break;
            case SAFE_ZONE_DECLARE_EVENT:
                setSafeZoneLimit(inboundEvent, sequence);
                break;
        }
    }

    /**
     * Calculate the current safe zone for this node (using the last generated message ID)
     * @param event event
     * @param sequence position of the event at the event ring buffer
     */
    private void setSafeZoneLimit(InboundEventContainer event, long sequence) {
        long safeZoneLimit = idGenerator.getNextId();
        event.setSafeZoneLimit(safeZoneLimit);
        if(log.isDebugEnabled()){
            log.debug("[ Sequence " + sequence + " ] Pre processing message. Setting the Safe Zone " + safeZoneLimit);
        }
    }

    /**
     * Route the message to queue/queues of subscribers matching in AMQP way. Hierarchical topic message routing is
     * evaluated here. This will duplicate message for each "subscription destination (not message destination)" at
     * different nodes
     *
     * @param event InboundEventContainer containing the message list
     */
    private void updateRoutingInformation(InboundEventContainer event, long sequence) {

        // NOTE: This is the MESSAGE_EVENT and this is the first processing event for this message published to ring
        // Therefore there should be exactly one message in the list.
        // NO NEED TO CHECK FOR LIST SIZE
        AndesMessage message = event.messageList.get(0);

        AndesChannel andesChannel = event.getChannel();

        // Messages are processed in the order they arrive at ring buffer By this processor.
        // By setting message ID through message pre processor we assure, even in a multi publisher scenario, there is
        // no message id ordering issue at node level.
        setMessageID(message);

        if(log.isDebugEnabled()){
            log.debug("[ Sequence " + sequence + " ] Pre processing message. Message ID "
                    + message.getMetadata().getMessageID());
        }

        if (message.getMetadata().isTopic()) {
            handleTopicRoutine(event, message, andesChannel);
        } else {
            andesChannel.recordAdditionToBuffer(message.getContentChunkList().size());
            AndesMessageMetadata messageMetadata = message.getMetadata();
            messageMetadata.setStorageQueueName(messageMetadata.getDestination());
        }
    }

    private void handleTopicRoutine(InboundEventContainer event, AndesMessage message, AndesChannel andesChannel) {
        String messageRoutingKey = message.getMetadata().getDestination();
        //get all topic subscriptions in the cluster matching to routing key
        //including hierarchical topic case
        Set<AndesSubscription> subscriptionList;
        try {

            // Get subscription list according to the message type
            AndesSubscription.SubscriptionType subscriptionType;

            if (MessageMetaDataType.META_DATA_MQTT == message.getMetadata().getMetaDataType()) {
                subscriptionType = AndesSubscription.SubscriptionType.MQTT;
            } else {
                subscriptionType = AndesSubscription.SubscriptionType.AMQP;
            }

            subscriptionList = subscriptionStore.getClusterSubscribersForDestination(messageRoutingKey, true, subscriptionType);

            // current message is removed from list and updated cloned messages added later
            event.messageList.clear();
            boolean isMessageRouted = false;
            Set<String> alreadyStoredQueueNames = new HashSet<String>();
            for (AndesSubscription subscription : subscriptionList) {
                if (!alreadyStoredQueueNames.contains(subscription.getStorageQueueName())) {

                    // Avoid adding QOS 0 MQTT messages to clean session = false subscribers if disconnected
                    if (AndesSubscription.SubscriptionType.MQTT == subscriptionType
                            && subscription.isDurable()
                            && !subscription.hasExternalSubscriptions()
                            && 0 == message.getMetadata().getQosLevel()) {
                        continue;
                    }

                    AndesMessage clonedMessage = cloneAndesMessageMetadataAndContent(message);
{'fraction_non_alphanumeric': 0.07259630948527031, 'fraction_numerical': 0.0031563612819682746, 'mean_word_length': 2.9403698979591835, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 11, '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': '7493375', 'n_tokens_mistral': 3002, 'n_tokens_neox': 2954, 'n_words': 983}