package be.bagofwords.db.bloomfilter;

import be.bagofwords.db.DataInterface;
import be.bagofwords.db.LayeredDataInterface;
import be.bagofwords.iterator.CloseableIterator;

import java.util.concurrent.locks.ReentrantLock;

public class BloomFilterDataInterface<T extends Object> extends LayeredDataInterface<T> {

    private static final double INITIAL_FPP = 0.001;
    private final static double MAX_FPP = INITIAL_FPP * 20;
    private final DataInterface<LongBloomFilterWithCheckSum> bloomFilterDataInterface;
    private final ReentrantLock modifyBloomFilterLock;
    private LongBloomFilterWithCheckSum bloomFilter;
    private boolean bloomFilterWasWrittenToDisk;
    private long currentKeyForNewBloomFilterCreation = Long.MAX_VALUE;

    private long timeOfLastRead;

    public BloomFilterDataInterface(DataInterface<T> baseInterface, DataInterface<LongBloomFilterWithCheckSum> bloomFilterDataInterface) {
        super(baseInterface);
        this.bloomFilterDataInterface = bloomFilterDataInterface;
        this.modifyBloomFilterLock = new ReentrantLock();
        this.bloomFilter = bloomFilterDataInterface.read(getName());
    }

    @Override
    public void optimizeForReading() {
        baseInterface.optimizeForReading();
        if (bloomFilter == null) {
            createNewBloomFilter();
        }
    }

    @Override
    public T readInt(long key) {
        timeOfLastRead = System.currentTimeMillis();
        LongBloomFilterWithCheckSum currentBloomFilter = bloomFilter;
        if (currentBloomFilter == null && modifyBloomFilterLock.tryLock()) {
            createNewBloomFilter();
            currentBloomFilter = bloomFilter;
            modifyBloomFilterLock.unlock();
        }
        if (currentBloomFilter == null || currentKeyForNewBloomFilterCreation < key) {
            //we are still creating the bloom filter
            return baseInterface.read(key);
        } else {
            if (currentBloomFilter.mightContain(key)) {
                return baseInterface.read(key);
            } else {
                return null;
            }
        }
    }


    @Override
    public void dropAllData() {
        modifyBloomFilterLock.lock();
        try {
            baseInterface.dropAllData();
            createNewBloomFilterNonSynchronized();
        } finally {
            modifyBloomFilterLock.unlock();
        }
    }

    @Override
    public boolean mightContain(long key) {
        LongBloomFilterWithCheckSum currentBloomFilter = bloomFilter;
        if (currentBloomFilter == null && modifyBloomFilterLock.tryLock()) {
            createNewBloomFilter();
            currentBloomFilter = bloomFilter;
            modifyBloomFilterLock.unlock();
        }
        if (currentBloomFilter == null || currentKeyForNewBloomFilterCreation < key) {
            //we are still creating the bloom filter
            return baseInterface.mightContain(key);
        } else {
            return currentBloomFilter.mightContain(key);
        }
    }

    @Override
    public void valuesChanged(long[] keys) {
        super.valuesChanged(keys);
        modifyBloomFilterLock.lock();
        if (bloomFilter != null) {
            if (keys.length > 0) {
                bloomFilterWasWrittenToDisk = false;
            }
            for (Long key : keys) {
                bloomFilter.put(key);
            }
            if (bloomFilter.expectedFpp() > MAX_FPP) {
                if (System.currentTimeMillis() - timeOfLastRead < 1000) {
                    //read in last second, create a new bloom filter
                    createNewBloomFilterNonSynchronized();
                } else {
                    //no reads in last second, let's drop the bloom filter for now
                    bloomFilter = null;
                }
            }
        }
        modifyBloomFilterLock.unlock();
    }

    private void createNewBloomFilterNonSynchronized() {
        currentKeyForNewBloomFilterCreation = Long.MIN_VALUE;
        long numOfValuesForBloomFilter = Math.max(1000000, baseInterface.apprSize());
        bloomFilter = new LongBloomFilterWithCheckSum(numOfValuesForBloomFilter, INITIAL_FPP);
        bloomFilter.setDataCheckSum(baseInterface.dataCheckSum());
        long start = System.currentTimeMillis();
        int numOfKeys = 0;
        CloseableIterator<Long> it = baseInterface.keyIterator();
        while (it.hasNext()) {
            long key = it.next();
            bloomFilter.put(key);
            numOfKeys++;
            currentKeyForNewBloomFilterCreation = key;
        }
        it.close();
        currentKeyForNewBloomFilterCreation = Long.MAX_VALUE;
        long taken = (System.currentTimeMillis() - start);
//        UI.write("Created bloomfilter " + getName() + " in " + taken + " ms for " + numOfKeys + " keys, size is " + bloomFilter.getBits().size() / (8 * 1024) + " kbytes.");
    }

    private void createNewBloomFilter() {
        modifyBloomFilterLock.lock();
        createNewBloomFilterNonSynchronized();
        modifyBloomFilterLock.unlock();
    }

    @Override
    public void flush() {
        baseInterface.flush();
        if (!bloomFilterWasWrittenToDisk) {
            bloomFilterDataInterface.write(getName(), bloomFilter);
            bloomFilterWasWrittenToDisk = true;
        }
    }

    @Override
    protected void doClose() {
        bloomFilter = null;
        baseInterface.close();
    }
}

@@ -3114,32 +3114,50 @@
 terLock.lock();%0A
+        try %7B%0A    
         if (bloo
@@ -3167,32 +3167,36 @@
 lter != null) %7B%0A
+    
             if (
@@ -3222,32 +3222,36 @@
                 
+    
 bloomFilterWasWr
@@ -3275,34 +3275,42 @@
 se;%0A            
+   
+ 
 %7D%0A
+    
             for 
@@ -3337,32 +3337,36 @@
                 
+   
+ 
 bloomFilter.put(
@@ -3375,34 +3375,40 @@
 y);%0A            
-%7D%0A
+    %7D%0A  
             if (
@@ -3395,32 +3395,34 @@
 %7D%0A              
+  
 if (bloomFilter.
@@ -3444,24 +3444,28 @@
  MAX_FPP) %7B%0A
+    
             
@@ -3546,16 +3546,20 @@
         
+ 
+   
 //read i
@@ -3591,32 +3591,36 @@
 ew bloom filter%0A
+    
                 
@@ -3670,32 +3670,36 @@
                 
+    
 %7D else %7B%0A       
@@ -3683,32 +3683,36 @@
        %7D else %7B%0A
+    
                 
@@ -3790,32 +3790,36 @@
                 
+    
 bloomFilter = nu
@@ -3814,32 +3814,54 @@
 mFilter = null;%0A
+                    %7D%0A
                 
@@ -3877,33 +3877,47 @@
      %7D%0A        %7D
-%0A
+ finally %7B%0A    
         modifyBl
@@ -3932,32 +3932,42 @@
 rLock.unlock();%0A
+        %7D%0A
     %7D%0A%0A    priva
@@ -5477,39 +5477,8 @@
 ll;%0A
-        baseInterface.close();%0A
     

Make sure we also release write lock on error