QUEUE-18 - Add functionality to chronicle queue 4 to support text wire

chronicle-queue/src/main/java/net/openhft/chronicle/queue/ChronicleQueueBuilder.java

@@ -15,42 +15,88 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package net.openhft.chronicle.queue;
 
 import net.openhft.chronicle.queue.impl.SingleChronicleQueue;
 import net.openhft.chronicle.wire.BinaryWire;
+import net.openhft.chronicle.wire.TextWire;
 import net.openhft.chronicle.wire.Wire;
 import org.jetbrains.annotations.NotNull;
 
+import java.io.File;
 import java.io.IOException;
 
 /**
  * Created by peter.lawrey on 30/01/15.
  */
-public class ChronicleQueueBuilder {
+public class ChronicleQueueBuilder implements Cloneable {
     private String name;
-    private long blockSize = 64 << 20;
-    private Class<? extends Wire> wireType = BinaryWire.class;
+    private long blockSize;
+    private Class<? extends Wire> wireType;
+
+    public ChronicleQueueBuilder(File name) {
+        this(name.getAbsolutePath());
+    }
 
     public ChronicleQueueBuilder(String name) {
         this.name = name;
+        this.blockSize = 64L << 20;
+        this.wireType = BinaryWire.class;
+    }
+
+    public String name() {
+        return this.name;
     }
 
-    @NotNull
     public ChronicleQueueBuilder blockSize(int blockSize) {
         this.blockSize = blockSize;
         return this;
     }
 
+    public long blockSize() {
+        return this.blockSize;
+    }
+
     public ChronicleQueueBuilder wireType(Class<? extends Wire> wireType) {
         this.wireType = wireType;
         return this;
     }
 
+    public Class<? extends Wire> wireType() {
+        return this.wireType;
+    }
 
     @NotNull
     public SingleChronicleQueue build() throws IOException {
-        return new SingleChronicleQueue(name, blockSize, wireType);
+        return new SingleChronicleQueue(this);
+    }
+
+    @NotNull
+    @SuppressWarnings("CloneDoesntDeclareCloneNotSupportedException")
+    @Override
+    public ChronicleQueueBuilder clone() {
+        try {
+            return (ChronicleQueueBuilder) super.clone();
+        } catch (CloneNotSupportedException e) {
+            throw new AssertionError(e);
+        }
+    }
+
+    public static ChronicleQueueBuilder binary(File name) {
+        return binary(name.getAbsolutePath());
+    }
+
+    public static ChronicleQueueBuilder binary(String name) {
+        return new ChronicleQueueBuilder(name)
+                .wireType(BinaryWire.class);
+    }
+
+    public static ChronicleQueueBuilder text(File name) {
+        return text(name.getAbsolutePath());
+    }
+
+    public static ChronicleQueueBuilder text(String name) {
+        return new ChronicleQueueBuilder(name)
+                .wireType(TextWire.class);
     }
 }

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

@@ -1,3 +1,20 @@
+/*
+ * Copyright 2015 Higher Frequency Trading
+ *
+ * http://www.higherfrequencytrading.com
+ *
+ * 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 net.openhft.chronicle.queue.impl;
 
 import net.openhft.chronicle.queue.ChronicleQueue;
@@ -7,12 +24,8 @@
  * Created by Rob Austin
  */
 abstract class AbstractChronicle implements ChronicleQueue, DirectChronicleQueue {
-
     abstract Wire wire();
-
     abstract Class<? extends Wire> wireType();
-
     abstract long indexToIndex();
-
     abstract long newIndex();
 }

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

@@ -24,13 +24,12 @@
 import net.openhft.chronicle.wire.*;
 import org.jetbrains.annotations.NotNull;
 
-import java.io.IOException;
 import java.time.ZonedDateTime;
 import java.util.UUID;
 
 /**
- * Data structure to bind to an off heap representation.  This is required to support persistence
- * and sharing of this data structure.
+ * Data structure to bind to an off heap representation. This is required to
+ * support persistence and sharing of this data structure.
  */
 class Header implements Marshallable {
     public static final long PADDED_SIZE = 512;
@@ -85,7 +84,7 @@ enum Field implements WireKey {
     @Override
     public void writeMarshallable(@NotNull WireOut out) {
         out.write(Field.uuid).uuid(uuid)
-                .write(Field.writeByte).int64forBinding(PADDED_SIZE)
+                .write(Field.writeByte).int64forBinding(PADDED_SIZE + 2)
                 .write(Field.created).zonedDateTime(created)
                 .write(Field.user).text(user)
                 .write(Field.host).text(host)
@@ -125,6 +124,5 @@ public static void main(String... args) {
         TextWire tw = new TextWire(NativeBytes.nativeBytes());
         tw.writeDocument(true, w -> w.write(() -> "header").marshallable(h));
         System.out.println(tw.bytes().flip().toString());
-
     }
 }

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

@@ -19,10 +19,10 @@
 package net.openhft.chronicle.queue.impl;
 
 import net.openhft.chronicle.queue.ExcerptTailer;
-import net.openhft.chronicle.wire.*;
+import net.openhft.chronicle.wire.ByteableLongArrayValues;
+import net.openhft.chronicle.wire.util.WireUtil;
 import org.jetbrains.annotations.NotNull;
 
-import static java.lang.ThreadLocal.withInitial;
 import static net.openhft.chronicle.queue.impl.Indexer.IndexOffset.toAddress0;
 import static net.openhft.chronicle.queue.impl.Indexer.IndexOffset.toAddress1;
 import static net.openhft.chronicle.queue.impl.SingleChronicleQueue.UNINITIALISED;
@@ -41,22 +41,10 @@ public class Indexer {
     private ThreadLocal<ByteableLongArrayValues> array;
 
     public Indexer(@NotNull final AbstractChronicle chronicle) {
-        this.array = newLongArrayValuesPool(chronicle.wireType());
+        this.array = WireUtil.newLongArrayValuesPool(chronicle.wireType());
         this.chronicle = chronicle;
     }
 
-    public static ThreadLocal<ByteableLongArrayValues>
-    newLongArrayValuesPool(Class<? extends Wire> wireType) {
-
-        if (TextWire.class.isAssignableFrom(wireType))
-            return withInitial(TextLongArrayReference::new);
-        if (BinaryWire.class.isAssignableFrom(wireType))
-            return withInitial(BinaryLongArrayReference::new);
-        else
-            throw new IllegalStateException("todo, unsupported type=" + wireType);
-
-    }
-
     /**
      * sans through every excerpts and records every 64th address in the index2indexs'
      *

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

@@ -15,7 +15,6 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package net.openhft.chronicle.queue.impl;
 
 import net.openhft.chronicle.bytes.Bytes;
@@ -28,23 +27,23 @@
 import org.jetbrains.annotations.Nullable;
 
 import java.util.function.Consumer;
-import java.util.function.Function;
 
 /**
  * Created by peter.lawrey on 30/01/15.
  */
 public class SingleAppender implements ExcerptAppender {
 
     @NotNull
-    private final DirectChronicleQueue chronicle;
-    private final Bytes buffer = NativeBytes.nativeBytes();
+    private final SingleChronicleQueue chronicle;
+    private final Bytes buffer;
     private final Wire wire;
 
     private long lastWrittenIndex = -1;
 
-    public SingleAppender(ChronicleQueue chronicle, Function<Bytes, Wire> bytesToWire) {
-        this.chronicle = (DirectChronicleQueue) chronicle;
-        wire = bytesToWire.apply(buffer);
+    public SingleAppender(@NotNull final SingleChronicleQueue chronicle) {
+        this.buffer = NativeBytes.nativeBytes();
+        this.chronicle = chronicle;
+        this.wire = chronicle.createWire(buffer);
     }
 
     @Nullable
@@ -68,10 +67,11 @@ public void writeDocument(@NotNull Consumer<WireOut> writer) {
     @Override
     public long lastWrittenIndex() {
         if (lastWrittenIndex == -1) {
-            String message = "No document has been written using this appender, so the " +
-                    "lastWrittenIndex() is not available.";
-            throw new IllegalStateException(message);
+            throw new IllegalStateException(
+                    "No document has been written using this appender, so the "
+                    + "lastWrittenIndex() is not available.");
         }
+
         return lastWrittenIndex;
     }