QUEUE-24 cleanup implementation

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

@@ -0,0 +1,83 @@
+/*
+ *     Copyright (C) 2015  higherfrequencytrading.com
+ *
+ *     This program is free software: you can redistribute it and/or modify
+ *     it under the terms of the GNU Lesser General Public License as published by
+ *     the Free Software Foundation, either version 3 of the License.
+ *
+ *     This program is distributed in the hope that it will be useful,
+ *     but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *     GNU Lesser General Public License for more details.
+ *
+ *     You should have received a copy of the GNU Lesser General Public License
+ *     along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package net.openhft.chronicle.queue.impl;
+
+import net.openhft.chronicle.core.ReferenceCounted;
+import net.openhft.chronicle.wire.ReadMarshallable;
+import net.openhft.chronicle.wire.WriteMarshallable;
+import org.jetbrains.annotations.NotNull;
+
+import java.io.IOException;
+
+public interface WireStore extends ReferenceCounted  {
+
+    /**
+     *
+     * @return
+     */
+    int cycle();
+
+    /**
+     *
+     * @return
+     */
+    long dataPosition();
+
+    /**
+     *
+     * @return
+     */
+    long writePosition();
+
+    /**
+     *
+     * @return
+     */
+    long lastIndex();
+
+
+    /**
+     *
+     * @param cycle
+     * @return
+     * @throws IOException
+     */
+    boolean appendRollMeta(int cycle) throws IOException;
+
+    /**
+     *
+     * @param writer
+     * @return
+     * @throws IOException
+     */
+    long append(@NotNull WriteMarshallable writer) throws IOException;
+
+    /**
+     *
+     * @param position
+     * @param reader
+     * @return
+     * @throws IOException
+     */
+    long read(long position, @NotNull ReadMarshallable reader) throws IOException;
+
+    /**
+     *
+     * @param index
+     * @return
+     */
+    long positionForIndex(long index);
+}

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

@@ -20,6 +20,7 @@
 import net.openhft.chronicle.queue.ExcerptTailer;
 import net.openhft.chronicle.queue.RollDateCache;
 import net.openhft.chronicle.queue.impl.AbstractChronicleQueue;
+import net.openhft.chronicle.queue.impl.WireStore;
 import net.openhft.koloboke.collect.map.hash.HashIntObjMaps;
 
 import java.io.File;
@@ -31,7 +32,7 @@ class SingleChronicleQueue extends AbstractChronicleQueue {
 
     private final SingleChronicleQueueBuilder builder;
     private final RollDateCache dateCache;
-    private final Map<Integer, SingleChronicleQueueStore> stores;
+    private final Map<Integer, SingleChronicleQueueWireStore> stores;
     private int firstCycle;
 
     protected SingleChronicleQueue(final SingleChronicleQueueBuilder builder) throws IOException {
@@ -59,12 +60,12 @@ SingleChronicleQueueBuilder builder() {
         return this.builder;
     }
 
-    synchronized SingleChronicleQueueStore storeForCycle(int cycle) throws IOException {
-        SingleChronicleQueueStore format = stores.get(cycle);
+    synchronized WireStore storeForCycle(int cycle) throws IOException {
+        SingleChronicleQueueWireStore format = stores.get(cycle);
         if(null == format) {
             stores.put(
                 cycle,
-                format = new SingleChronicleQueueStore(
+                format = new SingleChronicleQueueWireStore(
                     builder,
                     cycle,
                     this.dateCache.formatFor(cycle)).buildHeader()
@@ -76,7 +77,7 @@ synchronized SingleChronicleQueueStore storeForCycle(int cycle) throws IOExcepti
         return format;
     }
 
-    synchronized void release(SingleChronicleQueueStore store) {
+    synchronized void release(WireStore store) {
         store.release();
         if(store.refCount() <= 0) {
             stores.remove(store.cycle());

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

@@ -19,6 +19,7 @@
 import net.openhft.chronicle.queue.ChronicleQueue;
 import net.openhft.chronicle.queue.ExcerptAppender;
 import net.openhft.chronicle.queue.ExcerptTailer;
+import net.openhft.chronicle.queue.impl.WireStore;
 import net.openhft.chronicle.wire.ReadMarshallable;
 import net.openhft.chronicle.wire.WireUtil;
 import net.openhft.chronicle.wire.WriteMarshallable;
@@ -35,7 +36,7 @@ static class Appender implements ExcerptAppender {
 
         private int cycle;
         private long index;
-        private SingleChronicleQueueStore store;
+        private WireStore store;
 
         Appender(SingleChronicleQueue queue) throws IOException {
             this.queue = queue;
@@ -84,7 +85,7 @@ static class Tailer implements ExcerptTailer {
 
         private int cycle;
         private long position;
-        private SingleChronicleQueueStore store;
+        private WireStore store;
 
         Tailer(SingleChronicleQueue queue) throws IOException {
             this.queue = queue;

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

@@ -16,6 +16,7 @@
 package net.openhft.chronicle.queue.impl.single;
 
 import net.openhft.chronicle.core.OS;
+import net.openhft.chronicle.core.values.IntValue;
 import net.openhft.chronicle.core.values.LongValue;
 import net.openhft.chronicle.wire.*;
 import org.jetbrains.annotations.NotNull;
@@ -28,7 +29,7 @@
 class SingleChronicleQueueHeader implements Marshallable {
 
     private enum Fields implements WireKey {
-        type,
+        type, version,
         uuid, created, user, host,
         indexCount, indexSpacing,
         writePosition, dataPosition, index2Index, lastIndex,
@@ -39,12 +40,14 @@ private enum RollFields implements WireKey {
         cycle, length, format, timeZone, nextCycle, nextCycleMetaPosition
     }
 
-    public static final String QUEUE_TYPE = "SCV4";
+    public static final String QUEUE_TYPE = "SingleChronicleQueue";
+    public static final String QUEUE_VERSION = "4.0";
     public static final String CLASS_ALIAS = "Header";
     public static final long PADDED_SIZE = 512;
 
     // fields which can be serialized/deserialized in the normal way.
     private String type;
+    private String version;
     private UUID uuid;
     private ZonedDateTime created;
     private String user;
@@ -62,6 +65,7 @@ private enum RollFields implements WireKey {
 
     SingleChronicleQueueHeader(SingleChronicleQueueBuilder builder) {
         this.type = QUEUE_TYPE;
+        this.version = QUEUE_VERSION;
         this.uuid = UUID.randomUUID();
         this.created = ZonedDateTime.now();
         this.user = System.getProperty("user.name");
@@ -98,9 +102,10 @@ int indexCount() {
     @Override
     public void writeMarshallable(@NotNull WireOut out) {
         out.write(Fields.type).text(type)
+            .write(Fields.version).text(version)
             .write(Fields.uuid).uuid(uuid)
-            .write(Fields.writePosition).int64forBinding(WireUtil.SPB_HEADER_BYTE_SIZE)
-            .write(Fields.dataPosition).int64forBinding(WireUtil.SPB_HEADER_BYTE_SIZE)
+            .write(Fields.writePosition).int64forBinding(WireUtil.HEADER_OFFSET)
+            .write(Fields.dataPosition).int64forBinding(WireUtil.HEADER_OFFSET)
             .write(Fields.created).zonedDateTime(created)
             .write(Fields.user).text(user)
             .write(Fields.host).text(host)
@@ -109,13 +114,12 @@ public void writeMarshallable(@NotNull WireOut out) {
             .write(Fields.index2Index).int64forBinding(0L)
             .write(Fields.lastIndex).int64forBinding(-1L)
             .write(Fields.roll).marshallable(roll);
-
-        //out.addPadding((int) (PADDED_SIZE - out.bytes().writePosition()));
     }
 
     @Override
     public void readMarshallable(@NotNull WireIn in) {
         in.read(Fields.type).text(this, (o, i) -> o.type = i)
+            .read(Fields.version).text(this, (o, i) -> o.version = i)
             .read(Fields.uuid).uuid(this, (o, i) -> o.uuid = i)
             .read(Fields.writePosition).int64(this.writePosition, this, (o, i) -> o.writePosition = i)
             .read(Fields.dataPosition).int64(this.dataPosition, this, (o, i) -> o.dataPosition = i)
@@ -178,7 +182,7 @@ public int getNextRollCycle() {
     }
 
     public boolean casNextRollCycle(int rollCycle) {
-        return this.roll.nextCycle.compareAndSwapValue(-1L, rollCycle);
+        return this.roll.nextCycle.compareAndSwapValue(-1, rollCycle);
     }
 
     private class Roll implements Marshallable {
@@ -187,9 +191,8 @@ private class Roll implements Marshallable {
         private String format;
         private ZoneId zoneId;
 
-        //TODO: it appears there is a problem with IntValue (LongValue as workaround)
-        private LongValue cycle;
-        private LongValue nextCycle;
+        private IntValue cycle;
+        private IntValue nextCycle;
 
         // LongValue is right here
         private LongValue nextCycleMetaPosition;
@@ -206,21 +209,21 @@ private class Roll implements Marshallable {
 
         @Override
         public void writeMarshallable(@NotNull WireOut out) {
-            out.write(RollFields.cycle).int64forBinding(-1)
+            out.write(RollFields.cycle).int32forBinding(-1)
                 .write(RollFields.length).int32(length)
                 .write(RollFields.format).text(format)
-                    .write(RollFields.timeZone).text(zoneId.getId())
-                .write(RollFields.nextCycle).int64forBinding(-1)
+                .write(RollFields.timeZone).text(zoneId.getId())
+                .write(RollFields.nextCycle).int32forBinding(-1)
                 .write(RollFields.nextCycleMetaPosition).int64forBinding(-1);
         }
 
         @Override
         public void readMarshallable(@NotNull WireIn in) {
-            in.read(RollFields.cycle).int64(this.cycle, this, (o, i) -> o.cycle = i)
+            in.read(RollFields.cycle).int32(this.cycle, this, (o, i) -> o.cycle = i)
                 .read(RollFields.length).int32(this, (o, i) -> o.length = i)
                 .read(RollFields.format).text(this, (o, i) -> o.format = i)
-                    .read(RollFields.timeZone).text(this, (o, i) -> o.zoneId = ZoneId.of(i))
-                .read(RollFields.nextCycle).int64(this.nextCycle, this, (o, i) -> o.nextCycle = i)
+                .read(RollFields.timeZone).text(this, (o, i) -> o.zoneId = ZoneId.of(i))
+                .read(RollFields.nextCycle).int32(this.nextCycle, this, (o, i) -> o.nextCycle = i)
                 .read(RollFields.nextCycleMetaPosition).int64(this.nextCycleMetaPosition, this, (o, i) -> o.nextCycleMetaPosition = i);
         }
     }