/*
 * Copyright (c) 2011-2013 Spotify AB
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package com.spotify.sparkey.extra;

import com.google.common.collect.Lists;
import com.spotify.sparkey.*;

import java.io.File;
import java.io.IOException;
import java.util.Collection;
import java.util.Iterator;

/**
 * A thread-safe Sparkey Reader.
 */
public class ThreadLocalSparkeyReader implements SparkeyReader {

  private final Collection<SparkeyReader> readers = Lists.newArrayList();
  private final ThreadLocal<SparkeyReader> threadLocalReader;

  public ThreadLocalSparkeyReader(File indexFile) throws IOException {
    this(Sparkey.open(indexFile));
  }

  public ThreadLocalSparkeyReader(final SparkeyReader reader) {
    this.readers.add(reader);
    this.threadLocalReader = new ThreadLocal<SparkeyReader>() {
      @Override
      protected SparkeyReader initialValue() {
        SparkeyReader r = reader.duplicate();
        readers.add(r);
        return r;
      }
    };
  }

  @Override
  public String getAsString(String key) throws IOException {
    return threadLocalReader.get().getAsString(key);
  }

  @Override
  public byte[] getAsByteArray(byte[] key) throws IOException {
    return threadLocalReader.get().getAsByteArray(key);
  }

  @Override
  public Entry getAsEntry(byte[] key) throws IOException {
    return threadLocalReader.get().getAsEntry(key);
  }

  @Override
  public void close() throws IOException {
    synchronized (readers) {
      for (SparkeyReader reader : readers) {
        reader.close();
      }
      readers.clear();
    }
    threadLocalReader.remove();
  }

  @Override
  public IndexHeader getIndexHeader() {
    return threadLocalReader.get().getIndexHeader();
  }

  @Override
  public LogHeader getLogHeader() {
    return threadLocalReader.get().getLogHeader();
  }

  @Override
  public SparkeyReader duplicate() {
    return this;
  }

  @Override
  public Iterator<Entry> iterator() {
    return threadLocalReader.get().iterator();
  }
}

@@ -1049,16 +1049,52 @@
 eader;%0A%0A
+  private volatile boolean closed;%0A%0A
   public
@@ -1152,24 +1152,24 @@
 Exception %7B%0A
-
     this(Spa
@@ -1256,24 +1256,124 @@
 r reader) %7B%0A
+    if (reader == null) %7B%0A      throw new IllegalArgumentException(%22reader may not be null%22);%0A    %7D%0A
     this.rea
@@ -1513,24 +1513,51 @@
 alValue() %7B%0A
+        assertNotClosed();%0A
         Spar
@@ -1586,24 +1586,59 @@
 uplicate();%0A
+        synchronized (readers) %7B%0A  
         read
@@ -1645,24 +1645,34 @@
 ers.add(r);%0A
+        %7D%0A
         retu
@@ -1762,32 +1762,55 @@
 s IOException %7B%0A
+    assertNotClosed();%0A
     return threa
@@ -1843,32 +1843,155 @@
 ring(key);%0A  %7D%0A%0A
+  private void assertNotClosed() %7B%0A    if (closed) %7B%0A      throw new IllegalStateException(%22Reader is closed%22);%0A    %7D%0A  %7D%0A%0A
   @Override%0A  pu
@@ -2042,32 +2042,55 @@
 s IOException %7B%0A
+    assertNotClosed();%0A
     return threa
@@ -2201,24 +2201,47 @@
 Exception %7B%0A
+    assertNotClosed();%0A
     return t
@@ -2336,24 +2336,43 @@
 Exception %7B%0A
+    closed = true;%0A
     synchron
@@ -2575,32 +2575,55 @@
 IndexHeader() %7B%0A
+    assertNotClosed();%0A
     return threa
@@ -2708,24 +2708,47 @@
 gHeader() %7B%0A
+    assertNotClosed();%0A
     return t
@@ -2836,24 +2836,47 @@
 plicate() %7B%0A
+    assertNotClosed();%0A
     return t
@@ -2889,32 +2889,32 @@
  %7D%0A%0A  @Override%0A
-
   public Iterato
@@ -2931,24 +2931,47 @@
 terator() %7B%0A
+    assertNotClosed();%0A
     return t

Fix race condition in ThreadLocalSparkeyReader