<commit_before>package com.arpnetworking.tsdaggregator.publishing;

import com.arpnetworking.tsdaggregator.AggregatedData;
import com.arpnetworking.tsdaggregator.aggserver.AggregatorConnection;
import com.arpnetworking.tsdaggregator.aggserver.Messages;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.vertx.java.core.*;
import org.vertx.java.core.net.NetClient;
import org.vertx.java.core.net.NetSocket;
import org.vertx.java.platform.Verticle;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;

/**
 * Publisher to send data to an upstream aggregation server.
 *
 * @author barp
 */
public class AggServerPublisher extends Verticle implements AggregationPublisher {
    private static final Logger LOGGER = LoggerFactory.getLogger(AggServerPublisher.class);
    private final AtomicBoolean _connected = new AtomicBoolean(false);
    private final AtomicBoolean _connectionInProgress = new AtomicBoolean(false);
    private final String _aggServerHost;
    private final String _hostName;
    private final String _clusterName;
    private final int _aggServerPort;
    private final NetClient _client;
    private final ConcurrentLinkedQueue<AggregatedData[]> _pending = new ConcurrentLinkedQueue<AggregatedData[]>();
    @Nullable
    private NetSocket _socket = null;
    private final Vertx _vertx;

    public AggServerPublisher(@Nonnull String aggClusterLocation, String hostName, String clusterName) {
        _vertx = VertxFactory.newVertx();
        String[] split = aggClusterLocation.split(":");
        _aggServerHost = split[0];
        if (split.length > 1) {
            _aggServerPort = Integer.parseInt(split[1]);
        } else {
            _aggServerPort = 7065;
        }

        _hostName = hostName;
        _clusterName = clusterName;

        _client = _vertx.createNetClient();

        _client.setReconnectAttempts(1).setReconnectInterval(3000L).setConnectTimeout(5000).setTCPNoDelay(true)
                .setTCPKeepAlive(true);
        connectToAggServer();
    }

    private void connectToAggServer() {
        _connectionInProgress.set(true);
        LOGGER.info("attempting to connect to agg server at " + _aggServerHost + ":" + _aggServerPort);
        _client.connect(
                _aggServerPort, _aggServerHost, new AsyncResultHandler<NetSocket>() {
            @Override
            public void handle(@Nonnull final AsyncResult<NetSocket> event) {
                if (event.succeeded()) {
                    LOGGER.info("connected to aggregation server at " + _aggServerHost + ":" + _aggServerPort);
                    _socket = event.result();
                    _socket.exceptionHandler(createSocketExceptionHandler(_socket));
                    _socket.closeHandler(createSocketCloseHandler(_socket));
                    _connected.set(true);
                    _connectionInProgress.set(false);
                    @Nonnull Messages.HostIdentification hostIdent =
                            Messages.HostIdentification.newBuilder().setHostName(_hostName).setClusterName(_clusterName)
                                    .build();
                    @Nonnull AggregatorConnection.Message m = AggregatorConnection.Message.create(hostIdent);
                    _socket.write(m.getBuffer());
                    flushPending();
                } else if (event.failed()) {
                    LOGGER.warn(
                            "failed to connect to aggregation server at " + _aggServerHost + ":" + _aggServerPort);
                    _connectionInProgress.set(false);
                    _connected.set(false);
                    connectToAggServer();
                }
            }
        }
        );
    }

    @Nullable
    private Handler<Void> createSocketCloseHandler(final NetSocket socket) {
        return new Handler<Void>() {
            @Override
            public void handle(final Void event) {
                _socket = null;
                _connected.set(false);
                LOGGER.error("agg server publisher socket closed, attempting to reestablish connection", event);
                connectToAggServer();
            }
        };
    }

    @Nullable
    private Handler<Throwable> createSocketExceptionHandler(final NetSocket socket) {
        return new Handler<Throwable>() {
            @Override
            public void handle(final Throwable event) {
                try {
                    if (socket != null) {
                        socket.close();
                    }
                } catch (Exception ignored) { }
                _socket = null;
                _connected.set(false);
                LOGGER.error("error on agg server publisher socket, attempting to reestablish connection", event);
                connectToAggServer();<commit_msg>second attempt at fixing connection bug<commit_after>package com.arpnetworking.tsdaggregator.publishing;

import com.arpnetworking.tsdaggregator.AggregatedData;
import com.arpnetworking.tsdaggregator.aggserver.AggregatorConnection;
import com.arpnetworking.tsdaggregator.aggserver.Messages;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.vertx.java.core.*;
import org.vertx.java.core.net.NetClient;
import org.vertx.java.core.net.NetSocket;
import org.vertx.java.platform.Verticle;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;

/**
 * Publisher to send data to an upstream aggregation server.
 *
 * @author barp
 */
public class AggServerPublisher extends Verticle implements AggregationPublisher {
    private static final Logger LOGGER = LoggerFactory.getLogger(AggServerPublisher.class);
    private final AtomicBoolean _connected = new AtomicBoolean(false);
    private final AtomicBoolean _connectionInProgress = new AtomicBoolean(false);
    private final String _aggServerHost;
    private final String _hostName;
    private final String _clusterName;
    private final int _aggServerPort;
    private final NetClient _client;
    private final ConcurrentLinkedQueue<AggregatedData[]> _pending = new ConcurrentLinkedQueue<AggregatedData[]>();
    @Nullable
    private NetSocket _socket = null;
    private final Vertx _vertx;

    public AggServerPublisher(@Nonnull String aggClusterLocation, String hostName, String clusterName) {
        _vertx = VertxFactory.newVertx();
        String[] split = aggClusterLocation.split(":");
        _aggServerHost = split[0];
        if (split.length > 1) {
            _aggServerPort = Integer.parseInt(split[1]);
        } else {
            _aggServerPort = 7065;
        }

        _hostName = hostName;
        _clusterName = clusterName;

        _client = _vertx.createNetClient();

        _client.setReconnectAttempts(1).setReconnectInterval(3000L).setConnectTimeout(5000).setTCPNoDelay(true)
                .setTCPKeepAlive(true);
        connectToAggServer();
    }

    private void connectToAggServer() {
        if (_connectionInProgress.get() || _connected.get()) {
            LOGGER.warn("agg server publisher trying to connect when another connection is active or attempt in progress");
            return;
        }
        _connectionInProgress.set(true);
        LOGGER.info("attempting to connect to agg server at " + _aggServerHost + ":" + _aggServerPort);
        _client.connect(
                _aggServerPort, _aggServerHost, new AsyncResultHandler<NetSocket>() {
            @Override
            public void handle(@Nonnull final AsyncResult<NetSocket> event) {
                if (event.succeeded()) {
                    LOGGER.info("connected to aggregation server at " + _aggServerHost + ":" + _aggServerPort);
                    _socket = event.result();
                    _socket.exceptionHandler(createSocketExceptionHandler(_socket));
                    _socket.closeHandler(createSocketCloseHandler(_socket));
                    _connected.set(true);
                    _connectionInProgress.set(false);
                    @Nonnull Messages.HostIdentification hostIdent =
                            Messages.HostIdentification.newBuilder().setHostName(_hostName).setClusterName(_clusterName)
                                    .build();
                    @Nonnull AggregatorConnection.Message m = AggregatorConnection.Message.create(hostIdent);
                    _socket.write(m.getBuffer());
                    flushPending();
                } else if (event.failed()) {
                    LOGGER.warn(
                            "failed to connect to aggregation server at " + _aggServerHost + ":" + _aggServerPort);
                    _connectionInProgress.set(false);
                    _connected.set(false);
                    connectToAggServer();
                }
            }
        }
        );
    }

    @Nullable
    private Handler<Void> createSocketCloseHandler(final NetSocket socket) {
        return new Handler<Void>() {
            @Override
            public void handle(final Void event) {
                try {
                    if (socket != null) {
                        socket.close();
                    }
                } catch (Exception ignored) { }
                _socket = null;
                _connected.set(false);
                LOGGER.error("agg server publisher socket closed, attempting to reestablish connection", event);
                connectToAggServer();
            }
        };
    }

    @Nullable
    private Handler<Throwable> createSocketExceptionHandler(final NetSocket socket) {
        return new Handler<Throwable>() {
            @Override
            public void handle(final Throwable event) {
                try {
                    if (socket != null) {
                        socket.close();
                    }
                } catch (Exception ignored) { }
                _socket = null;
                _connected.set(false);
                LOGGER.error("error on agg server publisher socket, attempting to reestablish connection", event);
                connectToAggServer();
{'fraction_non_alphanumeric': 0.08472916054461749, 'fraction_numerical': 0.003526300323244196, 'mean_word_length': 2.2588573252473667, 'pattern_counts': {'":': 8, '<': 19, '<?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': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1119565', 'n_tokens_mistral': 2648, 'n_tokens_neox': 2481, 'n_words': 561}