<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.processor.common.util;

import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import org.apache.log4j.Logger;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
import org.wso2.carbon.event.processor.common.storm.manager.service.StormManagerService;
import org.wso2.carbon.event.processor.common.storm.manager.service.exception.EndpointNotFoundException;
import org.wso2.carbon.event.processor.common.storm.manager.service.exception.NotStormManagerException;
import org.wso2.carbon.event.processor.manager.commons.transport.client.TCPEventPublisher;
import org.wso2.carbon.event.processor.manager.commons.utils.HostAndPort;
import org.wso2.carbon.event.processor.manager.commons.utils.Utils;
import org.wso2.carbon.event.processor.manager.core.config.DistributedConfiguration;
import org.wso2.siddhi.query.api.definition.StreamDefinition;

import java.io.IOException;
import java.net.SocketException;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Executors;

/**
 * Sending events to a remote endpoint asynchronously.
 */
public class AsyncEventPublisher implements EventHandler<AsynchronousEventBuffer.DataHolder> {
    public enum DestinationType {STORM_RECEIVER, CEP_PUBLISHER}

    private transient Logger log = Logger.getLogger(AsyncEventPublisher.class);
    private String logPrefix;

    private String destinationTypeString;
    private String publisherTypeString;
    private DestinationType destinationType;

    private Set<StreamDefinition> streams;
    private String executionPlanName;
    private int tenantId;
    private String thisHostIp;
    private List<HostAndPort> managerServiceEndpoints;
    private DistributedConfiguration stormDeploymentConfig;

    private TCPEventPublisher tcpEventPublisher = null;
    private EndpointConnectionCreator endpointConnectionCreator;
    private AsynchronousEventBuffer eventSendBuffer = new AsynchronousEventBuffer<Object[]>(1024, this);

    private boolean shutdown = false;

    public AsyncEventPublisher(DestinationType destinationType, Set<StreamDefinition> streams,
                               List<HostAndPort> managerServiceEndpoints,
                               String executionPlanName, int tenantId, DistributedConfiguration stormDeploymentConfig) {
        this.destinationType = destinationType;
        this.streams = streams;
        this.executionPlanName = executionPlanName;
        this.tenantId = tenantId;
        this.managerServiceEndpoints = managerServiceEndpoints;
        this.stormDeploymentConfig = stormDeploymentConfig;
        this.endpointConnectionCreator = new EndpointConnectionCreator();
        this.destinationTypeString = (destinationType == DestinationType.STORM_RECEIVER) ? "Storm Receiver" : "CEP Publisher";
        this.publisherTypeString = (destinationType == DestinationType.STORM_RECEIVER) ? "CEP Receiver" : "Publisher Bolt";
        this.logPrefix = "[" + tenantId + ":" + executionPlanName + ":" + publisherTypeString + "] ";
    }

    /**
     * Initialize and try to make a connection with remote endpoint
     *
     * @param sync is this is set to true returns only after obtaining a connection to the remote endpoint. Otherwise initialization happens on a newly spawned
     *             thread and this method returns immediately.
     */
    public void initializeConnection(boolean sync) {
        try {
            this.thisHostIp = Utils.findAddress("localhost");

            if (sync) {
                endpointConnectionCreator.establishConnection();
            } else {
                Thread thread = new Thread(endpointConnectionCreator);
                thread.start();
            }
        } catch (SocketException e) {
            log.error(logPrefix + "Error while trying to obtain this host IP address", e);
        }
    }

    /**
     * Add event to the outbound event buffer and this call will return. Event will be sent asynchronously by disruptor consumer thread
     * via AsyncEventPublisher#onEvent.
     *
     * @param eventData
     * @param streamId
     */
    public void sendEvent(Object[] eventData, long timestamp, String streamId) {
        eventSendBuffer.addEvent(eventData, timestamp, streamId);
    }

    /**
     * Callback from disruptor for the consumer to consume data. This is where events are actually dispatched to the remote end.
     * If an exception occurs when trying send data it will keep trying to send for ever until succeeds. Returns only
     * after sending the event.
     *
     * @param dataHolder
     * @param sequence
     * @param endOfBatch
     */
    @Override
    public void onEvent(AsynchronousEventBuffer.DataHolder dataHolder, long sequence, boolean endOfBatch) {
        while (tcpEventPublisher == null) {
            log.info(logPrefix + "Can't send event. TCP event publisher not initialized. Waiting " + stormDeploymentConfig.getTransportReconnectInterval() + "s");
            try {
                synchronized (this) {
                    if (shutdown) {
                        log.info(logPrefix + "Aborting retry to send events. AsyncEventPublisher has shutdown.");
                        return;
                    }
                }
                Thread.sleep(stormDeploymentConfig.getTransportReconnectInterval());
            } catch (InterruptedException e) {
                //ignore
            }
        }

        // TODO : comment on message lost of the last batch
        try {
            tcpEventPublisher.sendEvent(dataHolder.getStreamId(), dataHolder.getTimestamp(), (Object[]) dataHolder.getData(), endOfBatch);
        } catch (IOException e) {
            log.error(logPrefix + "Error while trying to send event to " + destinationTypeString + " at " +
                    tcpEventPublisher.getHostUrl(), e);
            reconnect();
            onEvent(dataHolder, sequence, endOfBatch);
        }
    }

    /**
     * First tires to reconnect to the already obtained end point. If failed re-initialize the connection.
     */
    private void reconnect() {
        String destinationHostPort = tcpEventPublisher.getHostUrl();
        resetTCPEventPublisher();

        // Retrying to connect to the existing endpoint.
        tcpEventPublisher = endpointConnectionCreator.connectToEndpoint(destinationHostPort, 3);
        if (tcpEventPublisher == null) {
            log.error(logPrefix + "Failed to connect to existing " + destinationTypeString + " at " + destinationHostPort + ". Reinitializing");
            initializeConnection(true);
        }
    }

    private void resetTCPEventPublisher() {
        tcpEventPublisher.terminate();
        tcpEventPublisher = null;
    }

    @Override
    protected void finalize() {
        if (tcpEventPublisher != null) {
            tcpEventPublisher.terminate();<commit_msg>fixing build<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.processor.common.util;

import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import org.apache.log4j.Logger;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
import org.wso2.carbon.event.processor.common.storm.manager.service.StormManagerService;
import org.wso2.carbon.event.processor.common.storm.manager.service.exception.EndpointNotFoundException;
import org.wso2.carbon.event.processor.common.storm.manager.service.exception.NotStormManagerException;
import org.wso2.carbon.event.processor.manager.commons.transport.client.TCPEventPublisher;
import org.wso2.carbon.event.processor.manager.commons.utils.HostAndPort;
import org.wso2.carbon.event.processor.manager.commons.utils.Utils;
import org.wso2.carbon.event.processor.manager.core.config.DistributedConfiguration;
import org.wso2.siddhi.query.api.definition.StreamDefinition;

import java.io.IOException;
import java.net.SocketException;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Executors;

/**
 * Sending events to a remote endpoint asynchronously.
 */
public class AsyncEventPublisher implements EventHandler<AsynchronousEventBuffer.DataHolder> {
    public enum DestinationType {STORM_RECEIVER, CEP_PUBLISHER}

    private transient Logger log = Logger.getLogger(AsyncEventPublisher.class);
    private String logPrefix;

    private String destinationTypeString;
    private String publisherTypeString;
    private DestinationType destinationType;

    private Set<StreamDefinition> streams;
    private String executionPlanName;
    private int tenantId;
    private String thisHostIp;
    private List<HostAndPort> managerServiceEndpoints;
    private DistributedConfiguration stormDeploymentConfig;

    private TCPEventPublisher tcpEventPublisher = null;
    private EndpointConnectionCreator endpointConnectionCreator;
    private AsynchronousEventBuffer eventSendBuffer = new AsynchronousEventBuffer<Object[]>(1024, this);

    private boolean shutdown = false;

    public AsyncEventPublisher(DestinationType destinationType, Set<StreamDefinition> streams,
                               List<HostAndPort> managerServiceEndpoints,
                               String executionPlanName, int tenantId, DistributedConfiguration stormDeploymentConfig) {
        this.destinationType = destinationType;
        this.streams = streams;
        this.executionPlanName = executionPlanName;
        this.tenantId = tenantId;
        this.managerServiceEndpoints = managerServiceEndpoints;
        this.stormDeploymentConfig = stormDeploymentConfig;
        this.endpointConnectionCreator = new EndpointConnectionCreator();
        this.destinationTypeString = (destinationType == DestinationType.STORM_RECEIVER) ? "Storm Receiver" : "CEP Publisher";
        this.publisherTypeString = (destinationType == DestinationType.STORM_RECEIVER) ? "CEP Receiver" : "Publisher Bolt";
        this.logPrefix = "[" + tenantId + ":" + executionPlanName + ":" + publisherTypeString + "] ";
    }

    /**
     * Initialize and try to make a connection with remote endpoint
     *
     * @param sync is this is set to true returns only after obtaining a connection to the remote endpoint. Otherwise initialization happens on a newly spawned
     *             thread and this method returns immediately.
     */
    public void initializeConnection(boolean sync) {
        try {
            this.thisHostIp = Utils.findAddress("localhost");

            if (sync) {
                endpointConnectionCreator.establishConnection();
            } else {
                Thread thread = new Thread(endpointConnectionCreator);
                thread.start();
            }
        } catch (SocketException e) {
            log.error(logPrefix + "Error while trying to obtain this host IP address", e);
        }
    }

    /**
     * Add event to the outbound event buffer and this call will return. Event will be sent asynchronously by disruptor consumer thread
     * via AsyncEventPublisher#onEvent.
     *
     * @param eventData
     * @param streamId
     */
    public void sendEvent(Object[] eventData, long timestamp, String streamId) {
        eventSendBuffer.addEvent(eventData, timestamp, streamId);
    }

    /**
     * Callback from disruptor for the consumer to consume data. This is where events are actually dispatched to the remote end.
     * If an exception occurs when trying send data it will keep trying to send for ever until succeeds. Returns only
     * after sending the event.
     *
     * @param dataHolder
     * @param sequence
     * @param endOfBatch
     */
    @Override
    public void onEvent(AsynchronousEventBuffer.DataHolder dataHolder, long sequence, boolean endOfBatch) {
        while (tcpEventPublisher == null) {
            log.info(logPrefix + "Can't send event. TCP event publisher not initialized. Waiting " + stormDeploymentConfig.getTransportReconnectInterval() + "s");
            try {
                synchronized (this) {
                    if (shutdown) {
                        log.info(logPrefix + "Aborting retry to send events. AsyncEventPublisher has shutdown.");
                        return;
                    }
                }
                Thread.sleep(stormDeploymentConfig.getTransportReconnectInterval());
            } catch (InterruptedException e) {
                //ignore
            }
        }

        // TODO : comment on message lost of the last batch
        try {
            tcpEventPublisher.sendEvent(dataHolder.getStreamId(), dataHolder.getTimestamp(), (Object[]) dataHolder.getData(), endOfBatch);
        } catch (IOException e) {
            log.error(logPrefix + "Error while trying to send event to " + destinationTypeString + " at " +
                    tcpEventPublisher.getHostUrl(), e);
            reconnect();
            onEvent(dataHolder, sequence, endOfBatch);
        }
    }

    /**
     * First tires to reconnect to the already obtained end point. If failed re-initialize the connection.
     */
    private void reconnect() {
        String destinationHostPort = tcpEventPublisher.getHostUrl();
        resetTCPEventPublisher();

        // Retrying to connect to the existing endpoint.
        tcpEventPublisher = endpointConnectionCreator.connectToEndpoint(destinationHostPort, 3);
        if (tcpEventPublisher == null) {
            log.error(logPrefix + "Failed to connect to existing " + destinationTypeString + " at " + destinationHostPort + ". Reinitializing");
            initializeConnection(true);
        }
    }

    private void resetTCPEventPublisher() {
        tcpEventPublisher.shutdown();
        tcpEventPublisher = null;
    }

    @Override
    protected void finalize() {
        if (tcpEventPublisher != null) {
            tcpEventPublisher.shutdown();
{'fraction_non_alphanumeric': 0.0815568785609116, 'fraction_numerical': 0.003200819409768901, 'mean_word_length': 3.083115525352849, 'pattern_counts': {'":': 4, '<': 15, '<?xml version=': 0, '>': 15, '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': '6750092', 'n_tokens_mistral': 4083, 'n_tokens_neox': 3902, 'n_words': 1100}