package com.thefactory.datastore;

import java.util.Comparator;
import java.util.TreeMap;
import java.util.Iterator;
import java.util.Map;
import java.util.NoSuchElementException;

public class MemoryTablet {    
    private final TreeMap<Slice, Slice> backing;
    private long size = 0;

    public static Slice tombstone = new Slice(new byte[] {(byte)0x74, (byte)0x6f, (byte)0x6d, (byte)0x62});

    public MemoryTablet() {
        backing = new TreeMap<Slice, Slice>(
            new Comparator<Slice>() {
                public int compare(Slice x, Slice y) {
                    return Slice.compare(x, y);
                }
            }
        );        
    }

    public void set(Slice key, Slice value) {
        synchronized(backing) {
            backing.put(key, value);
            size += key.getLength() + value.getLength();
        }
    }

    public void delete(Slice key) {
        set(key, tombstone);
    }

    public void close() {
        backing.clear();
    }

    public Iterator<KV> find() {
        return find(null);
    }

    public Iterator<KV> find(final Slice term) {
        if (backing.size() == 0) {
            return empty();
        }

        synchronized(backing) {
            return new Iterator<KV>() {
                Iterator<Map.Entry<Slice, Slice>> itemIterator;
                {
                    if (term == null){
                        itemIterator = backing.entrySet().iterator();
                    } else {
                        itemIterator = backing.tailMap(term, true).entrySet().iterator();
                    }

                }

                public boolean hasNext() {
                    return itemIterator.hasNext();
                }

                public KV next() {
                    Map.Entry<Slice, Slice> item = itemIterator.next();
                    KV ret = new KV();
                    if(item.getValue() == tombstone){
                        ret.tombstone(item.getKey());
                    } else {
                        ret.reset(item.getKey(), item.getValue());
                    }
                    return ret;
                }

                public void remove() {
                    throw new UnsupportedOperationException();
                }
            };
        }
    }

    public void apply(Batch batch) {
        if (batch.isEmpty()) {
            return;
        }

        Iterator<KV> kvs = batch.pairs();
        while(kvs.hasNext()) {
            KV kv = kvs.next();
            if(kv.isDeleted()) {
                delete(kv.getKey().detach());
            } else {
                set(kv.getKey().detach(), kv.getValue().detach());
            }
        }
    }

    public long size() {
        return size;
    }

    private Iterator<KV> empty() {
        return new Iterator<KV>() {
            public boolean hasNext() {
                return false;
            }
            
            public KV next() {
                throw new NoSuchElementException();
            }
                
            public void remove() {
                throw new UnsupportedOperationException();
            }
        };
    }
}

@@ -74,20 +74,45 @@
 va.util.
-Tree
+concurrent.ConcurrentSkipList
 Map;%0Aimp
@@ -249,20 +249,34 @@
 e final 
-Tree
+ConcurrentSkipList
 Map%3CSlic
@@ -485,12 +485,26 @@
 new 
-Tree
+ConcurrentSkipList
 Map%3C
@@ -775,68 +775,61 @@
     
-synchronized(backing) %7B%0A            backing.put(key, value);
+backing.put(key, value);%0A        synchronized(this) %7B
 %0A   
@@ -1222,45 +1222,8 @@
   %7D%0A
-%0A        synchronized(backing) %7B%0A    
     
@@ -1262,28 +1262,24 @@
             
-    
 Iterator%3CMap
@@ -1330,16 +1330,10 @@
     
-    %7B%0A  
+%7B%0A
     
@@ -1340,26 +1340,24 @@
             
-  
 if (term == 
@@ -1363,20 +1363,16 @@
  null)%7B%0A
-    
         
@@ -1437,36 +1437,32 @@
                 
-   
- 
 %7D else %7B%0A       
@@ -1446,36 +1446,32 @@
        %7D else %7B%0A
-    
                 
@@ -1552,35 +1552,27 @@
             
-    %7D%0A%0A    
+%7D%0A%0A
             
@@ -1566,36 +1566,32 @@
 %0A            %7D%0A%0A
-    
             publ
@@ -1625,28 +1625,24 @@
             
- 
-   
 return itemI
@@ -1668,34 +1668,27 @@
             
-    %7D%0A%0A   
+%7D%0A%0A
             
@@ -1679,33 +1679,32 @@
  %7D%0A%0A            
- 
 public KV next()
@@ -1698,36 +1698,32 @@
 lic KV next() %7B%0A
-    
                 
@@ -1766,36 +1766,32 @@
 terator.next();%0A
-    
                 
@@ -1825,20 +1825,16 @@
         
-    
 if(item.
@@ -1875,28 +1875,24 @@
             
-    
 ret.tombston
@@ -1917,36 +1917,32 @@
                 
- 
-   
 %7D else %7B%0A       
@@ -1926,36 +1926,32 @@
        %7D else %7B%0A
-    
                 
@@ -2009,34 +2009,26 @@
             
-  
-  %7D%0A    
+%7D%0A
             
@@ -2051,35 +2051,27 @@
             
-   
- 
 %7D%0A%0A
-    
             
@@ -2085,36 +2085,32 @@
 void remove() %7B%0A
-    
                 
@@ -2156,36 +2156,32 @@
 ();%0A            
-    
 %7D%0A            %7D;
@@ -2178,24 +2178,10 @@
     
- 
-   %7D;%0A        
 %7D
+;
 %0A   

Use ConcurrentSkipList for MemoryTablet backing