CHRON-130 Create and Epoc, so something other than jan 1970 can be used

chronicle-queue/pom.xml

@@ -49,7 +49,7 @@
             <dependency>
                 <groupId>net.openhft</groupId>
                 <artifactId>chronicle-bom</artifactId>
-                <version>1.10.41</version>
+                <version>1.10.43-SNAPSHOT</version>
                 <type>pom</type>
                 <scope>import</scope>
             </dependency>

chronicle-queue/src/main/java/net/openhft/chronicle/queue/impl/AbstractChronicleQueue.java

@@ -80,10 +80,12 @@ public WireType wireType() {
 
     /**
      * @param cycle
+     * @param epoc   an epoc offset as the number of number of milliseconds since January
+     *                 1, 1970,  00:00:00 GMT
      * @return
      * @throws IOException
      */
-    protected abstract WireStore storeForCycle(long cycle) throws IOException;
+    protected abstract WireStore storeForCycle(long cycle, final long epoc) throws IOException;
 
     /**
      * @param store

chronicle-queue/src/main/java/net/openhft/chronicle/queue/impl/Excerpts.java

@@ -154,6 +154,7 @@ public long cycle() {
     public static class StoreAppender extends DefaultAppender<AbstractChronicleQueue> {
 
         private final MappedBytes writeContext;
+        private long epoc;
         private long cycle;
         private long index = -1;
         private WireStore store;
@@ -170,7 +171,7 @@ public StoreAppender(@NotNull AbstractChronicleQueue queue) throws IOException {
             if (this.cycle <= 0)
                 throw new IllegalArgumentException();
 
-            this.store = queue.storeForCycle(this.cycle);
+            this.store = queue.storeForCycle(this.cycle, this.epoc);
             this.index = this.store.lastIndex();
 
             final MappedFile mappedFile = store.mappedFile();
@@ -230,7 +231,7 @@ private WireStore store() throws IOException {
                 }
 
                 this.cycle = nextCycle;
-                this.store = queue.storeForCycle(this.cycle);
+                this.store = queue.storeForCycle(this.cycle, epoc);
                 //this.store.acquireBytesAtWritePositionForWrite(this.writeContext.bytes);
             }
 
@@ -253,6 +254,7 @@ public static class StoreTailer implements ExcerptTailer {
         private MappedBytes readContext;
 
         private long cycle;
+        private long epoc;
         private long index;
         private WireStore store;
 
@@ -411,7 +413,7 @@ private StoreTailer cycle(long cycle) throws IOException {
                 }
                 this.cycle = cycle;
                 this.index = -1;
-                this.store = this.queue.storeForCycle(this.cycle);
+                this.store = this.queue.storeForCycle(this.cycle, this.epoc);
                 final MappedFile mappedFile = store.mappedFile();
                 this.readContext = new MappedBytes(mappedFile);
 

chronicle-queue/src/main/java/net/openhft/chronicle/queue/impl/WireStore.java

@@ -36,6 +36,12 @@ public interface WireStore extends ReferenceCounted, Marshallable {
      */
     long cycle();
 
+    /**
+     * @return an epoc offset as the number of number of milliseconds since January 1, 1970,
+     * 00:00:00 GMT
+     */
+    long epoc();
+
     /**
      * @return the first readable position
      */

chronicle-queue/src/main/java/net/openhft/chronicle/queue/impl/WireStorePool.java

@@ -15,25 +15,57 @@
  */
 package net.openhft.chronicle.queue.impl;
 
-import net.openhft.koloboke.collect.map.hash.HashLongObjMaps;
 import org.jetbrains.annotations.NotNull;
 
 import java.io.IOException;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 public class WireStorePool {
     private final WireStoreSupplier supplier;
-    private final Map<Long, WireStore> stores;
+
+    private class RollDetails {
+
+        private long cycle;
+        private long epoc;
+
+        public RollDetails(long cycle, long epoc) {
+            this.cycle = cycle;
+            this.epoc = epoc;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (!(o instanceof RollDetails)) return false;
+
+            RollDetails rollDetails = (RollDetails) o;
+
+            if (cycle != rollDetails.cycle) return false;
+            return epoc == rollDetails.epoc;
+
+        }
+
+        @Override
+        public int hashCode() {
+            int result = (int) (cycle ^ (cycle >>> 32));
+            result = 31 * result + (int) (epoc ^ (epoc >>> 32));
+            return result;
+        }
+    }
+
+    private final Map<RollDetails, WireStore> stores;
 
     public WireStorePool(@NotNull WireStoreSupplier supplier) {
         this.supplier = supplier;
-        this.stores = HashLongObjMaps.newMutableMap();
+        this.stores = new ConcurrentHashMap<>();
     }
 
-    public synchronized WireStore acquire(long cycle) throws IOException {
-        WireStore store = stores.get(cycle);
-        if(store == null) {
-            stores.put(cycle, store = this.supplier.apply(cycle));
+    public synchronized WireStore acquire(long cycle, final long epoc) throws IOException {
+        final RollDetails rollDetails = new RollDetails(cycle, epoc);
+        WireStore store = stores.get(rollDetails);
+        if (store == null) {
+            stores.put(rollDetails, store = this.supplier.apply(cycle, epoc));
         } else {
             store.reserve();
         }
@@ -43,8 +75,8 @@ public synchronized WireStore acquire(long cycle) throws IOException {
 
     public synchronized void release(WireStore store) {
         store.release();
-        if(store.refCount() <= 0) {
-            stores.remove(store.cycle());
+        if (store.refCount() <= 0) {
+            stores.remove(new RollDetails(store.cycle(), store.epoc()));
         }
     }
 

chronicle-queue/src/main/java/net/openhft/chronicle/queue/impl/WireStoreSupplier.java

@@ -15,8 +15,8 @@
  */
 package net.openhft.chronicle.queue.impl;
 
-import java.util.function.LongFunction;
+import java.util.function.BiFunction;
 
 @FunctionalInterface
-public interface WireStoreSupplier extends LongFunction<WireStore>{
+public interface WireStoreSupplier extends BiFunction<Long, Long, WireStore> {
 }

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

@@ -54,9 +54,10 @@ protected SingleChronicleQueue(final SingleChronicleQueueBuilder builder) throws
         this.cycle = builder.rollCycle();
         this.dateCache = new RollDateCache(this.cycle);
         this.builder = builder;
+
         this.pool = WireStorePool.withSupplier(this::newStore);
         this.firstCycle = -1;
-        this.pool.acquire(cycle());
+        storeForCycle(cycle(), builder.epoc());
     }
 
     @NotNull
@@ -72,8 +73,8 @@ public ExcerptTailer createTailer() throws IOException {
     }
 
     @Override
-    protected WireStore storeForCycle(long cycle) throws IOException {
-        return this.pool.acquire(cycle);
+    protected WireStore storeForCycle(long cycle, final long epoc) throws IOException {
+        return this.pool.acquire(cycle, epoc);
     }
 
     @Override
@@ -168,13 +169,12 @@ public WireType wireType() {
     //
     // *************************************************************************
 
-    protected WireStore newStore(final long cycle) {
+    protected WireStore newStore(final long cycle, final long epoc) {
 
         final String cycleFormat = this.dateCache.formatFor(cycle);
         final File cycleFile = new File(this.builder.path(), cycleFormat + ".chronicle");
 
-
-        File parentFile = cycleFile.getParentFile();
+        final File parentFile = cycleFile.getParentFile();
         if (parentFile != null & !parentFile.exists()) {
             parentFile.mkdirs();
         }
@@ -191,7 +191,7 @@ protected WireStore newStore(final long cycle) {
 
         Function<MappedFile, WireStore> supplyStore = mappedFile -> new SingleChronicleQueueStore
                 (SingleChronicleQueue.this.builder.rollCycle(), SingleChronicleQueue.this
-                        .builder.wireType(), mappedFile);
+                        .builder.wireType(), mappedFile, epoc);
 
 
         if (cycleFile.exists()) {

chronicle-queue/src/main/java/net/openhft/chronicle/queue/impl/single/SingleChronicleQueueBuilder.java

@@ -33,6 +33,7 @@ public class SingleChronicleQueueBuilder implements ChronicleQueueBuilder {
     private RollCycle rollCycle;
 
     private int appendTimeout;
+    private long epoc;
 
     public SingleChronicleQueueBuilder(String path) {
         this(new File(path));
@@ -44,6 +45,7 @@ public SingleChronicleQueueBuilder(File path) {
         this.wireType = WireType.BINARY;
         this.appendTimeout = 10_000; // 10 seconds;
         this.rollCycle = RollCycles.DAYS;
+        this.epoc = 0;
     }
 
     public File path() {
@@ -82,6 +84,23 @@ public SingleChronicleQueueBuilder rollCycle(RollCycle rollCycle) {
         return this;
     }
 
+
+    /**
+     * sets epoc offset in milliseconds
+     *
+     * @param epoc sets an epoc offset as the number of number of milliseconds since January 1,
+     *             1970,  00:00:00 GMT
+     * @return {@code this}
+     */
+    public SingleChronicleQueueBuilder epoc(long epoc) {
+        this.epoc = epoc;
+        return this;
+    }
+
+    public long epoc() {
+        return epoc;
+    }
+
     public RollCycle rollCycle() {
         return this.rollCycle;
     }
@@ -130,6 +149,6 @@ public static SingleChronicleQueueBuilder raw(File name) {
 
     public static SingleChronicleQueueBuilder raw(String name) {
         return new SingleChronicleQueueBuilder(name)
-            .wireType(WireType.RAW);
+                .wireType(WireType.RAW);
     }
 }