<commit_before>package com.statful.sender;

import com.statful.client.StatfulMetricsOptions;
import com.statful.metric.DataPoint;
import io.vertx.core.*;
import io.vertx.core.datagram.DatagramSocket;
import io.vertx.core.datagram.DatagramSocketOptions;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;

import javax.annotation.Nonnull;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.Random;
import java.util.stream.Collectors;

/**
 * Responsible for the UDP transport
 */
public final class UDPSender extends MetricsHolder {

    /**
     * Logger for transport errors
     */
    private static final Logger LOGGER = LoggerFactory.getLogger(UDPSender.class);

    /**
     * Statful options to configure the sender
     */
    private final StatfulMetricsOptions options;

    /**
     * Holds the socket to avoid recreation
     */
    private DatagramSocket socket;

    /**
     * @param vertx   vertx instance to create the socket from
     * @param context of execution to run operations that need vertx initialized
     * @param options Statful options to configure host and port
     */
    public UDPSender(final Vertx vertx, final Context context, final StatfulMetricsOptions options) {
        super(options, new Sampler(options, new Random()));

        this.options = options;

        // the following code is being executed asynchronously on the same context, to make sure that vertx is properly initialized
        // so that we can open a socket and configure a interval
        context.runOnContext(aVoid -> {
            this.socket = vertx.createDatagramSocket(new DatagramSocketOptions());
            this.configureFlushInterval(vertx, this.options.getFlushInterval(), this.options.getFlushSize());
        });
    }

    @Override
    public void send(@Nonnull final List<DataPoint> metrics, @Nonnull final Handler<AsyncResult<Void>> sentHandler) {
        this.send(metrics, Optional.of(sentHandler));
    }

    @Override
    public void send(@Nonnull final List<DataPoint> metrics) {
        this.send(metrics, Optional.empty());
    }

    private void send(@Nonnull final List<DataPoint> metrics, final Optional<Handler<AsyncResult<Void>>> endHandler) {
        Objects.requireNonNull(metrics);

        if (metrics.isEmpty()) {
            // nothing to send
            return;
        }
        final String toSendMetrics = metrics.stream().map(DataPoint::toMetricLine).collect(Collectors.joining("\n"));

        socket.send(toSendMetrics, options.getPort(), options.getHost(), handler -> {
            if (handler.failed()) {
                LOGGER.warn("Failed to send metrics {}", toSendMetrics);
                endHandler.ifPresent(callerHandler -> callerHandler.handle(Future.failedFuture(handler.cause())));
            } else {
                endHandler.ifPresent(callerHandler -> callerHandler.handle(Future.succeededFuture()));
            }
        });
    }

    @Override
    public void close(final Handler<AsyncResult<Void>> handler) {
        this.socket.close(handler);
    }
}<commit_msg>fix(NA): Improve error log on failed metrics<commit_after>package com.statful.sender;

import com.statful.client.StatfulMetricsOptions;
import com.statful.metric.DataPoint;
import io.vertx.core.*;
import io.vertx.core.datagram.DatagramSocket;
import io.vertx.core.datagram.DatagramSocketOptions;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;

import javax.annotation.Nonnull;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.Random;
import java.util.stream.Collectors;

/**
 * Responsible for the UDP transport
 */
public final class UDPSender extends MetricsHolder {

    /**
     * Logger for transport errors
     */
    private static final Logger LOGGER = LoggerFactory.getLogger(UDPSender.class);

    /**
     * Statful options to configure the sender
     */
    private final StatfulMetricsOptions options;

    /**
     * Holds the socket to avoid recreation
     */
    private DatagramSocket socket;

    /**
     * @param vertx   vertx instance to create the socket from
     * @param context of execution to run operations that need vertx initialized
     * @param options Statful options to configure host and port
     */
    public UDPSender(final Vertx vertx, final Context context, final StatfulMetricsOptions options) {
        super(options, new Sampler(options, new Random()));

        this.options = options;

        // the following code is being executed asynchronously on the same context, to make sure that vertx is properly initialized
        // so that we can open a socket and configure a interval
        context.runOnContext(aVoid -> {
            this.socket = vertx.createDatagramSocket(new DatagramSocketOptions());
            this.configureFlushInterval(vertx, this.options.getFlushInterval(), this.options.getFlushSize());
        });
    }

    @Override
    public void send(@Nonnull final List<DataPoint> metrics, @Nonnull final Handler<AsyncResult<Void>> sentHandler) {
        this.send(metrics, Optional.of(sentHandler));
    }

    @Override
    public void send(@Nonnull final List<DataPoint> metrics) {
        this.send(metrics, Optional.empty());
    }

    private void send(@Nonnull final List<DataPoint> metrics, final Optional<Handler<AsyncResult<Void>>> endHandler) {
        Objects.requireNonNull(metrics);

        if (metrics.isEmpty()) {
            // nothing to send
            return;
        }
        final String toSendMetrics = metrics.stream().map(DataPoint::toMetricLine).collect(Collectors.joining("\n"));

        socket.send(toSendMetrics, options.getPort(), options.getHost(), handler -> {
            if (handler.failed()) {
                LOGGER.error("Failed to send metrics {}", handler.cause(),  toSendMetrics);
                endHandler.ifPresent(callerHandler -> callerHandler.handle(Future.failedFuture(handler.cause())));
            } else {
                endHandler.ifPresent(callerHandler -> callerHandler.handle(Future.succeededFuture()));
            }
        });
    }

    @Override
    public void close(final Handler<AsyncResult<Void>> handler) {
        this.socket.close(handler);
    }
}
{'fraction_non_alphanumeric': 0.10197263760738148, 'fraction_numerical': 0.0, 'mean_word_length': 3.0930989583333335, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '4035288', 'n_tokens_mistral': 1691, 'n_tokens_neox': 1653, 'n_words': 439}