<commit_before>package org.voovan.tools.collection;

import org.rocksdb.RocksIterator;
import org.voovan.tools.TByte;
import org.voovan.tools.TDateTime;
import org.voovan.tools.TEnv;
import org.voovan.tools.TString;

import java.io.Closeable;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * RocksDB 的 DelayQueue 封装 <br>
 * 最小延时单位: 秒
 * @author: helyho
 * Voovan Framework.
 * WebSite: https://github.com/helyho/Voovan
 * Licence: Apache v2 License
 */
public class RocksDelayQueue<E extends Delayed> implements Queue<E> {

    private RocksMap root;
    private RocksMap<String, E> container;
    private ConcurrentLinkedQueue<String> queueCache = new ConcurrentLinkedQueue<>();
    private volatile String lastKey;

    public RocksDelayQueue(RocksMap rocksMap, String name) {
        this.root = rocksMap;
        this.container = rocksMap.duplicate(name);
        this.container.getReadOptions().setTotalOrderSeek(true);

        Timer timer = new Timer();
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                RocksMap.RocksMapIterator iterator0 = null;
                synchronized (container) {
                    String fromKey = lastKey == null ? null : container.isKeyExists(lastKey) ? lastKey : null;
                    String toKey = lastKey == null ? TString.radixConvert(System.currentTimeMillis(), 62) : null;
                    iterator0 = container.iterator(fromKey, toKey, fromKey == null ? 0 : 1, 0);
                }

                try (RocksMap.RocksMapIterator iterator = iterator0) {
                    RocksMap.RocksMapEntry rocksMapEntry = null;
                    while ((rocksMapEntry = iterator.next()) != null) {
                        String key = rocksMapEntry.getKey().toString();
                        lastKey = key;
                        queueCache.add(key);
                    }
                }

            }
        }, 1000, 1000);
    }

    public RocksMap getRoot() {
        return root;
    }

    public RocksMap<String, E> getContainer() {
        return container;
    }

    private synchronized String offerSeq(long delayMilliSecond){
        String ret = TString.radixConvert((System.currentTimeMillis() + delayMilliSecond)/1000, 62) + TString.radixConvert(TDateTime.currentTimeNanos(), 62);
        return ret;
    }

    private synchronized String currentMilliTime(){
        String ret =  TString.radixConvert((System.currentTimeMillis())/1000, 62);
        return ret;
    }


    @Override
    public boolean add(E e) {
        container.put(offerSeq(e.getDelay(TimeUnit.MILLISECONDS)), e);
        return true;
    }

    @Override
    public boolean addAll(Collection<? extends E> c) {
        c.stream().forEach(e->add(e));
        return true;
    }



    @Override
    public int size() {
        return container.size();
    }

    @Override
    public boolean isEmpty() {
        return container.isEmpty();
    }

    @Override
    public boolean offer(E e) {
        return add(e);
    }

    @Override
    public E remove() {
        return poll();
    }

    public E take(long timeout, TimeUnit unit) throws TimeoutException {
        E e = poll();

        long time = TimeUnit.SECONDS.convert(timeout, unit);
        while(time>0) {
            if (e == null) {
                e = poll();
            }

            if(e!=null) {
                break;
            }

            time--;
            synchronized (this) {
                try {
                    this.wait(1000);
                } catch (InterruptedException ex) {
                    ex.printStackTrace();
                }
            }
        }

        if (e == null) {
            throw new TimeoutException();
        }

        return e;

    }

    ArrayList mmm = new ArrayList();

    @Override
    public E poll() {
        //RocksDB 迭代器多线程的并发问题, 有可能已经在其他线程 poll 过这个 key, 所以这里使用循环获取
        while(true) {
            String seq = queueCache.poll();
            if (seq != null) {
                E e = container.remove(seq);
                if(e==null) {
                    continue;
                }
                return e;
            } else {
                return null;
            }
        }
    }

    @Override
    public E element() {
        E e = poll();<commit_msg>imp: RocksDelayQueue 的 bug 修复<commit_after>package org.voovan.tools.collection;

import org.rocksdb.RocksIterator;
import org.voovan.tools.TByte;
import org.voovan.tools.TDateTime;
import org.voovan.tools.TEnv;
import org.voovan.tools.TString;

import java.io.Closeable;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * RocksDB 的 DelayQueue 封装 <br>
 * 最小延时单位: 秒
 * @author: helyho
 * Voovan Framework.
 * WebSite: https://github.com/helyho/Voovan
 * Licence: Apache v2 License
 */
public class RocksDelayQueue<E extends Delayed> implements Queue<E> {

    private RocksMap root;
    private RocksMap<String, E> container;
    private LinkedBlockingQueue<String> queueCache = new LinkedBlockingQueue<>();
    private volatile String lastKey;

    public RocksDelayQueue(RocksMap rocksMap, String name) {
        this.root = rocksMap;
        this.container = rocksMap.duplicate(name);
        this.container.getReadOptions().setTotalOrderSeek(true);

        Timer timer = new Timer();
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                RocksMap.RocksMapIterator iterator0 = null;
                synchronized (container) {
                    String fromKey = lastKey == null ? null : container.get(lastKey)!=null ? lastKey : null;
                    String toKey = lastKey == null ? TString.radixConvert(System.currentTimeMillis(), 62) : null;
                    iterator0 = container.iterator(fromKey, toKey, fromKey == null ? 0 : 1, 0);
                }

                try (RocksMap.RocksMapIterator iterator = iterator0) {
                    RocksMap.RocksMapEntry rocksMapEntry = null;
                    while ((rocksMapEntry = iterator.next()) != null) {
                        String key = rocksMapEntry.getKey().toString();
                        lastKey = key;
                        queueCache.add(key);
                    }
                }
            }
        }, 1000, 1000);
    }

    public RocksMap getRoot() {
        return root;
    }

    public RocksMap<String, E> getContainer() {
        return container;
    }

    private synchronized String offerSeq(long delayMilliSecond){
        String ret = TString.radixConvert((System.currentTimeMillis() + delayMilliSecond)/1000, 62) + TString.radixConvert(TDateTime.currentTimeNanos(), 62);
        return ret;
    }

    private synchronized String currentMilliTime(){
        String ret =  TString.radixConvert((System.currentTimeMillis())/1000, 62);
        return ret;
    }


    @Override
    public boolean add(E e) {
        container.put(offerSeq(e.getDelay(TimeUnit.MILLISECONDS)), e);
        return true;
    }

    @Override
    public boolean addAll(Collection<? extends E> c) {
        c.stream().forEach(e->add(e));
        return true;
    }



    @Override
    public int size() {
        return container.size();
    }

    @Override
    public boolean isEmpty() {
        return container.isEmpty();
    }

    @Override
    public boolean offer(E e) {
        return add(e);
    }

    @Override
    public E remove() {
        return poll();
    }

    public E take(long timeout, TimeUnit unit) throws TimeoutException {
        E e = poll();

        long time = TimeUnit.SECONDS.convert(timeout, unit);
        while(time>0) {
            if (e == null) {
                e = poll();
            }

            if(e!=null) {
                break;
            }

            time--;
            synchronized (this) {
                try {
                    this.wait(1000);
                } catch (InterruptedException ex) {
                    ex.printStackTrace();
                }
            }
        }

        if (e == null) {
            throw new TimeoutException();
        }

        return e;

    }

    ArrayList mmm = new ArrayList();

    @Override
    public E poll() {
        //RocksDB 迭代器多线程的并发问题, 有可能已经在其他线程 poll 过这个 key, 所以这里使用循环获取
        while(true) {
            String seq = queueCache.poll();
            if (seq != null) {
                E e = container.remove(seq);
                if(e==null) {
                    continue;
                }

                mmm.add(seq);
                return e;
            } else {
                return null;
            }
        }
    }

    @Override
    public E element() {
        E e = poll();
{'fraction_non_alphanumeric': 0.1043507817811013, 'fraction_numerical': 0.008157715839564922, 'mean_word_length': 1.752416588712192, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 23, 'https://': 2, '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': '2893427', 'n_tokens_mistral': 2624, 'n_tokens_neox': 2486, 'n_words': 578}