<commit_before>package com.vlkan.log4j2.redis.appender;

import com.google.common.util.concurrent.RateLimiter;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

class RedisThrottler implements AutoCloseable {

    private final RedisThrottlerConfig config;

    private final RedisThrottlerReceiver receiver;

    private final boolean ignoreExceptions;

    private final BlockingQueue<byte[]> buffer;

    private final byte[][] batch;

    private final Thread flushTrigger;

    private final RateLimiter rateLimiter;

    private final DebugLogger logger;

    private final AtomicReference<Throwable> lastThrown = new AtomicReference<>(null);

    RedisThrottler(RedisThrottlerConfig config, RedisThrottlerReceiver receiver, boolean ignoreExceptions, boolean debugEnabled) {
        this.config = config;
        this.receiver = receiver;
        this.ignoreExceptions = ignoreExceptions;
        this.buffer = new ArrayBlockingQueue<>(config.getBufferSize());
        this.batch = new byte[config.getBatchSize()][];
        this.flushTrigger = createFlushTrigger();
        this.rateLimiter = config.getMaxEventCountPerSecond() > 0 ? RateLimiter.create(config.getMaxEventCountPerSecond()) : null;
        this.logger = new DebugLogger(RedisThrottler.class, debugEnabled);
    }

    private Thread createFlushTrigger() {
        return new Thread(new Runnable() {
            @Override
            @SuppressWarnings("InfiniteLoopStatement")
            public void run() {
                logger.debug("started");
                do {
                    logger.debug("flushing");
                    try {
                        flush();
                    } catch (InterruptedException ignored) {
                        logger.debug("interrupted");
                        Thread.currentThread().interrupt();
                        break;
                    }
                } while (true);
            }
        });
    }

    private void flush() throws InterruptedException {

        int batchIndex = 0;
        byte[] event;

        // Flush in batches.
        logger.debug("polling");
        while ((event = buffer.poll(config.getFlushPeriodMillis(), TimeUnit.MILLISECONDS)) != null) {
            if (logger.isEnabled()) {
                logger.debug("polled: %s", new String(event));
            }
            batch[batchIndex++] = event;
            if (batchIndex == batch.length) {
                safeConsumeEvents(batch);
                batchIndex = 0;
            }
        }

        // Flush individually.
        if (batchIndex > 0) {
            logger.debug("pushing %d individual events", batchIndex);
            int eventCount = batchIndex;
            batchIndex = 0;
            do {
                event = this.batch[batchIndex];
                safeConsumeEvent(event);
            } while (++batchIndex < eventCount);
        }

    }

    private void safeConsumeEvent(byte[] event) {
        try {
            if (logger.isEnabled()) {
                logger.debug("pushing single event: %s", new String(event));
            }
            receiver.consumeThrottledEvent(event);
        } catch (Throwable thrown) {
            if (logger.isEnabled()) {
                logger.debug("push failure: %s", thrown.getMessage());
                thrown.printStackTrace();
            }
            lastThrown.set(thrown);
        }
    }

    private void safeConsumeEvents(final byte[]... events) {
        try {
            logger.debug("pushing %d events", events.length);
            receiver.consumeThrottledEvents(events);
        } catch (Throwable thrown) {
            if (logger.isEnabled()) {<commit_msg>Make sure we follow flushPeriodMillis in throttler.<commit_after>package com.vlkan.log4j2.redis.appender;

import com.google.common.util.concurrent.RateLimiter;

import java.util.Arrays;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

class RedisThrottler implements AutoCloseable {

    private final RedisThrottlerConfig config;

    private final RedisThrottlerReceiver receiver;

    private final boolean ignoreExceptions;

    private final BlockingQueue<byte[]> buffer;

    private final byte[][] batch;

    private final Thread flushTrigger;

    private final RateLimiter rateLimiter;

    private final DebugLogger logger;

    private final AtomicReference<Throwable> lastThrown = new AtomicReference<>(null);

    RedisThrottler(RedisThrottlerConfig config, RedisThrottlerReceiver receiver, boolean ignoreExceptions, boolean debugEnabled) {
        this.config = config;
        this.receiver = receiver;
        this.ignoreExceptions = ignoreExceptions;
        this.buffer = new ArrayBlockingQueue<>(config.getBufferSize());
        this.batch = new byte[config.getBatchSize()][];
        this.flushTrigger = createFlushTrigger();
        this.rateLimiter = config.getMaxEventCountPerSecond() > 0 ? RateLimiter.create(config.getMaxEventCountPerSecond()) : null;
        this.logger = new DebugLogger(RedisThrottler.class, debugEnabled);
    }

    private Thread createFlushTrigger() {
        return new Thread(new Runnable() {
            @Override
            @SuppressWarnings("InfiniteLoopStatement")
            public void run() {
                logger.debug("started");
                do {
                    logger.debug("flushing");
                    try {
                        flush();
                    } catch (InterruptedException ignored) {
                        logger.debug("interrupted");
                        Thread.currentThread().interrupt();
                        break;
                    }
                } while (true);
            }
        });
    }

    private void flush() throws InterruptedException {

        int batchIndex = 0;
        byte[] event;

        // Flush in batches.
        logger.debug("polling");
        long waitPeriodMillis = config.getFlushPeriodMillis();
        while (waitPeriodMillis > 0) {
            long pollTimeMillis = System.currentTimeMillis();
            event = buffer.poll(waitPeriodMillis, TimeUnit.MILLISECONDS);
            if (event == null) {
                break;
            }
            if (logger.isEnabled()) {
                logger.debug("polled: %s", new String(event));
            }
            batch[batchIndex++] = event;
            if (batchIndex == batch.length) {
                safeConsumeEvents(batch);
                batchIndex = 0;
            }
            long pollPeriodMillis = System.currentTimeMillis() - pollTimeMillis;
            waitPeriodMillis -= pollPeriodMillis;
        }

        // Flush the last remaining.
        if (batchIndex > 0) {
            logger.debug("pushing remaining %d events", batchIndex);
            byte[][] subBatch = Arrays.copyOfRange(batch, 0, batchIndex);
            safeConsumeEvents(subBatch);
        }

    }

    private void safeConsumeEvents(final byte[]... events) {
        try {
            logger.debug("pushing %d events", events.length);
            receiver.consumeThrottledEvents(events);
        } catch (Throwable thrown) {
            if (logger.isEnabled()) {
{'fraction_non_alphanumeric': 0.09727679176263379, 'fraction_numerical': 0.0020322449532583662, 'mean_word_length': 2.091289782244556, 'pattern_counts': {'":': 0, '<': 12, '<?xml version=': 0, '>': 16, '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': '1118128', 'n_tokens_mistral': 1913, 'n_tokens_neox': 1867, 'n_words': 408}