<commit_before>package ru.dreamkas.utils.concurrent.collections;

import java.util.Comparator;
import java.util.List;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ConcurrentAutoFlushingBuffer<T> {
    private static final Logger log = LoggerFactory.getLogger(ConcurrentAutoFlushingBuffer.class);
    private final BucketExecutor<T> dataExecutor;
    private final ConcurrentSkipListSet<T> bucket;
    private final ScheduledExecutorService executor;
    private Integer packetSize = null;

    public ConcurrentAutoFlushingBuffer(int flushPeriod, Integer packetSize, Comparator<T> comparator, BucketExecutor<T> dataExecutor) {
        this.packetSize = packetSize;
        this.dataExecutor = dataExecutor;
        bucket = new ConcurrentSkipListSet<>(comparator != null ? comparator : (o1, o2) -> 0);
        executor = Executors.newSingleThreadScheduledExecutor();
        executor.scheduleWithFixedDelay(this::flush, flushPeriod, flushPeriod, TimeUnit.MILLISECONDS);
    }

    public void put(T item) {
        bucket.add(item);
    }

    public void putAll(List<T> items) {
        bucket.addAll(items);
    }

    public void immediatelyFlush() {
        Executors.newCachedThreadPool().execute(this::flush);
    }

    public void shutdown() {
        executor.shutdown();
    }

    private void flush() {
        List<T> data = bucket.stream().limit(packetSize == null ? bucket.size() : packetSize).collect(Collectors.toList());
        try {
            if (data.size() > 0) {
                dataExecutor.execute(data);
                bucket.removeAll(data);
            }
        } catch (ConcurrentAutoFlushingException e) {
            bucket.removeAll(e.getExpiredItems());
            log.error("DROP CASHED DATA FROM BUCKET", e);
        }
    }
}<commit_msg>START-305: Чек может отправиться в Кабинет дважды<commit_after>package ru.dreamkas.utils.concurrent.collections;

import java.util.Comparator;
import java.util.List;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ConcurrentAutoFlushingBuffer<T> {
    private static final Logger log = LoggerFactory.getLogger(ConcurrentAutoFlushingBuffer.class);
    private final BucketExecutor<T> dataExecutor;
    private final ConcurrentSkipListSet<T> bucket;
    private final ScheduledExecutorService executor;
    private Integer packetSize = null;

    public ConcurrentAutoFlushingBuffer(int flushPeriod, Integer packetSize, Comparator<T> comparator, BucketExecutor<T> dataExecutor) {
        this.packetSize = packetSize;
        this.dataExecutor = dataExecutor;
        bucket = new ConcurrentSkipListSet<>(comparator != null ? comparator : (o1, o2) -> 0);
        executor = Executors.newSingleThreadScheduledExecutor();
        executor.scheduleWithFixedDelay(this::flush, flushPeriod, flushPeriod, TimeUnit.MILLISECONDS);
    }

    public void put(T item) {
        bucket.add(item);
    }

    public void putAll(List<T> items) {
        bucket.addAll(items);
    }

    public void immediatelyFlush() {
        executor.execute(this::flush);
    }

    public void shutdown() {
        executor.shutdown();
    }

    private synchronized void flush() {
        List<T> data = bucket.stream().limit(packetSize == null ? bucket.size() : packetSize).collect(Collectors.toList());
        try {
            if (data.size() > 0) {
                dataExecutor.execute(data);
                bucket.removeAll(data);
            }
        } catch (ConcurrentAutoFlushingException e) {
            bucket.removeAll(e.getExpiredItems());
            log.error("DROP CASHED DATA FROM BUCKET", e);
        }
    }
}
{'fraction_non_alphanumeric': 0.1019155206286837, 'fraction_numerical': 0.0036836935166994107, 'mean_word_length': 3.379569892473118, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 23, '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': '988944', 'n_tokens_mistral': 1174, 'n_tokens_neox': 1152, 'n_words': 237}