#315 Use a weak collection to hold resources which need closing to av…

…oid a memory leak.

src/main/java/net/openhft/chronicle/queue/impl/single/SingleChronicleQueue.java

@@ -43,7 +43,6 @@
 import java.nio.file.Path;
 import java.text.ParseException;
 import java.util.*;
-import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -88,7 +87,7 @@ public class SingleChronicleQueue implements RollingChronicleQueue {
     @NotNull
     private final BiFunction<RollingChronicleQueue, Wire, WireStore> storeFactory;
     private final StoreRecoveryFactory recoverySupplier;
-    private final Set<Runnable> closers = new CopyOnWriteArraySet<>();
+    private final Map<Object, Consumer> closers = new WeakHashMap<>();
     private final boolean readOnly;
     long firstAndLastCycleTime = 0;
     int firstCycle = Integer.MAX_VALUE, lastCycle = Integer.MIN_VALUE;
@@ -401,8 +400,10 @@ public long countExcerpts(long lowerIndex, long upperIndex) throws IllegalStateE
         return result;
     }
 
-    public void addCloseListener(Runnable closer) {
-        closers.add(closer);
+    public <T> void addCloseListener(T key, Consumer<T> closer) {
+        synchronized (closers) {
+            closers.put(key, closer);
+        }
     }
 
     @Override
@@ -414,7 +415,10 @@ public boolean isClosed() {
     public void close() {
         if (isClosed.getAndSet(true))
             return;
-        closers.forEach(Runnable::run);
+        synchronized (closers) {
+            closers.forEach((k, v) -> v.accept(k));
+            closers.clear();
+        }
         this.pool.close();
     }
 

src/main/java/net/openhft/chronicle/queue/impl/single/SingleChronicleQueueExcerpts.java

@@ -71,7 +71,6 @@ public interface InternalAppender {
      * // StoreAppender
      */
     static class StoreAppender implements ExcerptAppender, ExcerptContext, InternalAppender {
-        static final int HEAD_ROOM = 1 << 20;
         @NotNull
         private final SingleChronicleQueue queue;
         private final StoreAppenderContext context;
@@ -91,7 +90,7 @@ static class StoreAppender implements ExcerptAppender, ExcerptContext, InternalA
 
         StoreAppender(@NotNull SingleChronicleQueue queue) {
             this.queue = queue;
-            queue.addCloseListener(StoreAppender.this::close);
+            queue.addCloseListener(this, StoreAppender::close);
             context = new StoreAppenderContext();
         }
 
@@ -132,7 +131,7 @@ public void writeText(CharSequence text) throws UnrecoverableTimeoutException {
             }
         }
 
-        private void close() {
+        void close() {
             Wire w0 = wireForIndex;
             wireForIndex = null;
             if (w0 != null)
@@ -364,6 +363,7 @@ public DocumentContext writingDocument(long index) {
             assert checkAppendingThread();
             context.wire = acquireBufferWire();
             context.wire.headerNumber(index);
+            context.isClosed = false;
             return context;
         }
 
@@ -789,7 +789,7 @@ public StoreTailer(@NotNull final SingleChronicleQueue queue) {
             this.queue = queue;
             this.setCycle(Integer.MIN_VALUE);
             this.index = 0;
-            queue.addCloseListener(StoreTailer.this::close);
+            queue.addCloseListener(this, StoreTailer::close);
             indexSpacingMask = queue.rollCycle().defaultIndexSpacing() - 1;
             toStart();
         }