<commit_before>/*
 * Copyright 2016 <EMAIL>
 *
 * 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.single;

import net.openhft.chronicle.bytes.Bytes;
import net.openhft.chronicle.bytes.BytesRingBufferStats;
import net.openhft.chronicle.bytes.BytesStore;
import net.openhft.chronicle.bytes.MappedBytes;
import net.openhft.chronicle.core.Jvm;
import net.openhft.chronicle.core.Maths;
import net.openhft.chronicle.core.OS;
import net.openhft.chronicle.core.io.IORuntimeException;
import net.openhft.chronicle.core.threads.EventLoop;
import net.openhft.chronicle.core.time.SystemTimeProvider;
import net.openhft.chronicle.core.time.TimeProvider;
import net.openhft.chronicle.core.util.ObjectUtils;
import net.openhft.chronicle.core.util.ThrowingBiFunction;
import net.openhft.chronicle.core.util.Updater;
import net.openhft.chronicle.queue.BufferMode;
import net.openhft.chronicle.queue.QueueOffsetSpec;
import net.openhft.chronicle.queue.RollCycle;
import net.openhft.chronicle.queue.RollCycles;
import net.openhft.chronicle.queue.impl.RollingChronicleQueue;
import net.openhft.chronicle.queue.impl.StoreFileListener;
import net.openhft.chronicle.queue.impl.TableStore;
import net.openhft.chronicle.queue.impl.WireStoreFactory;
import net.openhft.chronicle.queue.impl.table.ReadonlyTableStore;
import net.openhft.chronicle.queue.impl.table.SingleTableBuilder;
import net.openhft.chronicle.threads.EventGroup;
import net.openhft.chronicle.threads.TimeoutPauser;
import net.openhft.chronicle.threads.TimingPauser;
import net.openhft.chronicle.wire.*;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.crypto.spec.SecretKeySpec;
import java.io.File;
import java.lang.reflect.Constructor;
import java.nio.file.Path;
import java.time.LocalTime;
import java.time.ZoneId;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Supplier;

import static net.openhft.chronicle.core.pool.ClassAliasPool.CLASS_ALIASES;
import static net.openhft.chronicle.queue.ChronicleQueue.TEST_BLOCK_SIZE;
import static net.openhft.chronicle.queue.impl.single.SingleChronicleQueue.QUEUE_METADATA_FILE;
import static net.openhft.chronicle.wire.WireType.DEFAULT_ZERO_BINARY;
import static net.openhft.chronicle.wire.WireType.DELTA_BINARY;

public class SingleChronicleQueueBuilder implements Cloneable, Marshallable {
    public static final String DEFAULT_ROLL_CYCLE_PROPERTY = "net.openhft.queue.builder.defaultRollCycle";
    private static final Constructor ENTERPISE_QUEUE_CONSTRUCTOR;
    private static final String DEFAULT_EPOCH_PROPERTY = "net.openhft.queue.builder.defaultEpoch";
    private static final Logger LOGGER = LoggerFactory.getLogger(SingleChronicleQueueBuilder.class);

    static {
        CLASS_ALIASES.addAlias(WireType.class);
        CLASS_ALIASES.addAlias(SCQMeta.class, "SCQMeta");
        CLASS_ALIASES.addAlias(SCQRoll.class, "SCQSRoll");
        CLASS_ALIASES.addAlias(SCQIndexing.class, "SCQSIndexing");
        CLASS_ALIASES.addAlias(SingleChronicleQueueStore.class, "SCQStore");
        CLASS_ALIASES.addAlias(TimedStoreRecovery.class);

        {
            Constructor co;
            try {
                co = ((Class) Class.forName("software.chronicle.enterprise.queue.EnterpriseSingleChronicleQueue")).getDeclaredConstructors()[0];
                co.setAccessible(true);
            } catch (Exception e) {
                co = null;
            }
            ENTERPISE_QUEUE_CONSTRUCTOR = co;
        }

    }

    private WireType wireType = WireType.BINARY_LIGHT;
    private Long blockSize;
    private File path;
    private RollCycle rollCycle = loadDefaultRollCycle();
    private Long epoch; // default is 1970-01-01 00:00:00.000 UTC
    private Long bufferCapacity;
    private Integer indexSpacing;
    private Integer indexCount;

    public BufferMode writeBufferMode = BufferMode.None;
    public BufferMode readBufferMode = BufferMode.None;

    private Boolean enableRingBufferMonitoring;
    @Nullable
    private EventLoop eventLoop;
    private WireStoreFactory storeFactory = SingleChronicleQueueBuilder::createStore;
    /**
     * by default logs the performance stats of the ring buffer
     */
    @NotNull
    private Consumer<BytesRingBufferStats> onRingBufferStats = NoBytesRingBufferStats.NONE;
    private TimeProvider timeProvider = SystemTimeProvider.INSTANCE;
    private Supplier<TimingPauser> pauserSupplier = () -> new TimeoutPauser(500_000);
    private Long timeoutMS; // 10 seconds.
    private Integer sourceId;
    private StoreRecoveryFactory recoverySupplier = TimedStoreRecovery.FACTORY;
    private StoreFileListener storeFileListener;

    private Boolean readOnly;
    private Boolean strongAppenders;

    private TableStore<SCQMeta> metaStore;

    // enterprise stuff
    private int deltaCheckpointInterval = -1;
    private Supplier<BiConsumer<BytesStore, Bytes>> encodingSupplier;
    private Supplier<BiConsumer<BytesStore, Bytes>> decodingSupplier;
    private Updater<Bytes> messageInitializer;
    private Consumer<Bytes> messageHeaderReader;
    private SecretKeySpec key;

    private int maxTailers;
    private ThrowingBiFunction<Long, Integer, BytesStore, Exception> bufferBytesStoreCreator;
    private Long pretouchIntervalMillis;
    private LocalTime rollTime;
    private ZoneId rollTimeZone;
    private QueueOffsetSpec queueOffsetSpec;

    public static void addAliases() {
        // static initialiser.
    }
    /*
     * ========================
     * Builders
     * ========================
     */

    /**
     * @return an empty builder
     */
    public static SingleChronicleQueueBuilder builder() {
        return new SingleChronicleQueueBuilder();
    }

    @NotNull
    public static SingleChronicleQueueBuilder builder(@NotNull Path path, @NotNull WireType wireType) {
        return builder(path.toFile(), wireType);
    }

    @NotNull
    public static SingleChronicleQueueBuilder builder(@NotNull File file, @NotNull WireType wireType) {
        SingleChronicleQueueBuilder result = builder().wireType(wireType);
        if (file.isFile()) {
            if (!file.getName().endsWith(SingleChronicleQueue.SUFFIX)) {
                throw new IllegalArgumentException("Invalid file type: " + file.getName());
            }

            LOGGER.warn("Queues should be configured with the queue directory, not a specific filename. Actual file used: {}",
                    file.getParentFile());

            result.path(file.getParentFile());
        } else
            result.path(file);

        return result;
    }

    public static SingleChronicleQueueBuilder single() {
        SingleChronicleQueueBuilder builder = builder();
        builder.wireType(WireType.BINARY_LIGHT);
        return builder;
    }

    public static SingleChronicleQueueBuilder single(@NotNull String basePath) {
        return binary(basePath);
    }

    public static SingleChronicleQueueBuilder single(@NotNull File basePath) {
        return binary(basePath);
    }

    public static SingleChronicleQueueBuilder binary(@NotNull Path path) {
        return binary(path.toFile());
    }

    public static SingleChronicleQueueBuilder binary(@NotNull String basePath) {
        return binary(new File(basePath));
    }

    public static SingleChronicleQueueBuilder binary(@NotNull File basePathFile) {
        return builder(basePathFile, WireType.BINARY_LIGHT);
    }

    public static SingleChronicleQueueBuilder fieldlessBinary(@NotNull File name) {
        return builder(name, WireType.FIELDLESS_BINARY);
    }

    public static SingleChronicleQueueBuilder defaultZeroBinary(@NotNull File basePathFile) {
        return builder(basePathFile, DEFAULT_ZERO_BINARY);
    }

    public static SingleChronicleQueueBuilder deltaBinary(@NotNull File basePathFile) {
        return builder(basePathFile, DELTA_BINARY);
    }

    /**
     * @param name               the file name
     * @param deltaIntervalShift default value of 6, the shift for deltaInterval, the should be a
     *                           number between 0-63 ( inclusive ), default the delta messaging is
     *                           check pointed every 64 messages, so the default {@code
     *                           deltaIntervalShift  == 6}, as {@code 1 << 6 == 64 }
     * @return the EnterpriseChronicleQueueBuilder
     */
    public static SingleChronicleQueueBuilder deltaBinary(@NotNull File name, byte deltaIntervalShift) {
        @NotNull SingleChronicleQueueBuilder ret = deltaBinary(name);

        if (deltaIntervalShift < 0 || deltaIntervalShift > 63)
            throw new IllegalArgumentException("deltaIntervalShift=" + deltaIntervalShift + ", but " +
                    "should be a value between 0-63 inclusive");

        ret.deltaCheckpointInterval(1 << deltaIntervalShift);
        return ret;
    }

    protected SingleChronicleQueueBuilder() {
    }

    public WireStoreFactory storeFactory() {
        return storeFactory;
    }

    public boolean hasPretouchIntervalMillis() {
        return pretouchIntervalMillis != null;
    }

    @NotNull
    public SingleChronicleQueue build() {
        boolean needEnterprise = checkEnterpriseFeaturesRequested();
        preBuild();

        if (needEnterprise)
            return buildEnterprise();

        return new SingleChronicleQueue(this);
    }

    private boolean checkEnterpriseFeaturesRequested() {

        boolean result = false;
        if (readBufferMode != BufferMode.None)
            result = onlyAvailableInEnterprise("Buffering");
        if (writeBufferMode != BufferMode.None)
            result = onlyAvailableInEnterprise("Buffering");
        if (rollTimeZone != null && !rollTimeZone.getId().equals("UTC") && !rollTimeZone.getId().equals("Z"))
            result = onlyAvailableInEnterprise("Non-UTC roll time zone");
        if (wireType == WireType.DELTA_BINARY)
            result = onlyAvailableInEnterprise("Wire type " + wireType.name());
        if (encodingSupplier != null)
            result = onlyAvailableInEnterprise("Encoding");
        if (key != null)
            result = onlyAvailableInEnterprise("Encryption");
        if (hasPretouchIntervalMillis())
            result = onlyAvailableInEnterprise("Pretouching");

        return result;
    }

    private boolean onlyAvailableInEnterprise(final String feature) {
        if (ENTERPISE_QUEUE_CONSTRUCTOR == null)
            LOGGER.warn(feature + " is only supported in Chronicle Queue Enterprise. If you would like to use this feature, please contact <EMAIL> for more information.");
        return true;
    }

    @NotNull
    private SingleChronicleQueue buildEnterprise() {
        if (ENTERPISE_QUEUE_CONSTRUCTOR == null)
            throw new IllegalStateException("Enterprise features requested but Chronicle Queue Enterprise is not in the class path!");

        try {
            return (SingleChronicleQueue) ENTERPISE_QUEUE_CONSTRUCTOR.newInstance(this);
        } catch (Exception e) {
            throw new IllegalStateException("Couldn't create an instance of Enterprise queue", e);
        }

    }

    public SingleChronicleQueueBuilder aesEncryption(@Nullable byte[] keyBytes) {
        if (keyBytes == null) {
            codingSuppliers(null, null);
            return this;
        }
        key = new SecretKeySpec(keyBytes, "AES");
        return this;
    }

    public Updater<Bytes> messageInitializer() {
        return messageInitializer == null ? Bytes::clear : messageInitializer;
    }

    public Consumer<Bytes> messageHeaderReader() {
        return messageHeaderReader == null ? b -> {
        } : messageHeaderReader;
    }

    public SingleChronicleQueueBuilder messageHeader(Updater<Bytes> messageInitializer,
                                                     Consumer<Bytes> messageHeaderReader) {
        this.messageInitializer = messageInitializer;
        this.messageHeaderReader = messageHeaderReader;
        return this;
    }

    public SingleChronicleQueueBuilder rollTime(final LocalTime rollTime) {
        rollTime(rollTime, rollTimeZone);
        return this;
    }

    public ZoneId rollTimeZone() {
        return rollTimeZone;
    }

    public SingleChronicleQueueBuilder rollTimeZone(final ZoneId rollTimeZone) {
        rollTime(rollTime, rollTimeZone);
        return this;
    }

    public SingleChronicleQueueBuilder rollTime(@NotNull final LocalTime rollTime, final ZoneId zoneId) {
        this.rollTime = rollTime;
        this.rollTimeZone = zoneId;
        this.epoch = TimeUnit.SECONDS.toMillis(rollTime.toSecondOfDay());
        this.queueOffsetSpec = QueueOffsetSpec.ofRollTime(rollTime, zoneId);
        return this;
    }

    protected void initializeMetadata() {
        File metapath = metapath();
        validateRollCycle(metapath);
        SCQMeta metadata = new SCQMeta(new SCQRoll(rollCycle(), epoch()), deltaCheckpointInterval(),
                sourceId());
        try {

            boolean readOnly = readOnly();
            metaStore = SingleTableBuilder.binary(metapath, metadata).timeoutMS(timeoutMS())
                    .readOnly(readOnly).validateMetadata(!readOnly).build();
            // check if metadata was overridden
            SCQMeta newMeta = metaStore.metadata();
            if (sourceId() == 0)
                sourceId(newMeta.sourceId());

            if (readOnly && !newMeta.roll().format().equals(rollCycle().format())) {
                // roll cycle changed
                overrideRollCycleForFileNameLength(newMeta.roll().format().length());
            }
        } catch (IORuntimeException ex) {
            // readonly=true and file doesn't exist
            if (OS.isWindows())
                throw ex; // we cant have a read-only table store on windows so we have no option
            // but to throw the ex.
            metaStore = new ReadonlyTableStore<>(metadata);
        }
    }

    private void validateRollCycle(File metapath) {
        if (!metapath.exists()) {
            // no metadata, so we need to check if there're cq4 files and if so try to validate roll cycle
            // the code is slightly brutal and crude but should work for most cases. It will NOT work if files were created with
            // the following cycles: LARGE_HOURLY_SPARSE LARGE_HOURLY_XSPARSE LARGE_DAILY XLARGE_DAILY HUGE_DAILY HUGE_DAILY_XSPARSE
            // for such cases user MUST use correct roll cycle when creating the queue
            String[] list = path.list((d, name) -> name.endsWith(SingleChronicleQueue.SUFFIX));
            if (list != null && list.length > 0) {
                String filename = list[0];
                if (rollCycle().format().length() + 4 != filename.length()) {
                    // probably different roll cycle used
                    overrideRollCycleForFileNameLength(filename.length() - 4);
                }
            }
        }
    }

    private void overrideRollCycleForFileNameLength(int patternLength) {
        for (RollCycles cycle : RollCycles.values()) {
            if (cycle.format().length() == patternLength) {
                LOGGER.warn("Overriding roll cycle to " + cycle);
                rollCycle = cycle;
                break;
            }
        }
    }

    private File metapath() {
        final File storeFilePath;
        if ("".equals(path.getPath())) {
            storeFilePath = new File(QUEUE_METADATA_FILE);
        } else {
            storeFilePath = new File(path, QUEUE_METADATA_FILE);
            path.mkdirs();
        }
        return storeFilePath;
    }

    @NotNull
    QueueLock queueLock() {
        return isQueueReplicationAvailable() && !readOnly() ? new TSQueueLock(metaStore, pauserSupplier(), timeoutMS() * 3 / 2) : new NoopQueueLock();
    }

    @NotNull
    WriteLock writeLock() {
        return readOnly() ? new ReadOnlyWriteLock() : new TableStoreWriteLock(metaStore, pauserSupplier(), timeoutMS() * 3 / 2);
    }

    public SingleChronicleQueueBuilder enablePreloader(final long pretouchIntervalMillis) {
        this.pretouchIntervalMillis = pretouchIntervalMillis;
        return this;
    }

    public int deltaCheckpointInterval() {
        return deltaCheckpointInterval == -1 ? 64 : deltaCheckpointInterval;
    }

    public QueueOffsetSpec queueOffsetSpec() {
        return queueOffsetSpec == null ? QueueOffsetSpec.ofNone() : queueOffsetSpec;
    }

    TableStore<SCQMeta> metaStore() {
        return metaStore;
    }

    /**
     * RingBuffer tailers need to be preallocated. Only set this if using readBufferMode=Asynchronous.
     * By default 1 tailer will be created for the user.
     *
     * @param maxTailers number of tailers that will be required from this queue, not including the draining tailer
     * @return this
     */
    public SingleChronicleQueueBuilder maxTailers(int maxTailers) {
        this.maxTailers = maxTailers;
        return this;
    }

    /**
     * maxTailers
     *
     * @return number of tailers that will be required from this queue, not including the draining tailer
     */
    public int maxTailers() {
        return maxTailers;
    }

    public SingleChronicleQueueBuilder bufferBytesStoreCreator(ThrowingBiFunction<Long, Integer, BytesStore, Exception> bufferBytesStoreCreator) {
        this.bufferBytesStoreCreator = bufferBytesStoreCreator;
        return this;
    }

    /**
     * Creator for BytesStore for underlying ring buffer. Allows visibility of RB's data to be controlled.
     * See also RB_BYTES_STORE_CREATOR_NATIVE, RB_BYTES_STORE_CREATOR_MAPPED_FILE
     *
     * @return
     */
    @Nullable
    public ThrowingBiFunction<Long, Integer, BytesStore, Exception> bufferBytesStoreCreator() {
        return bufferBytesStoreCreator;
    }


    public long pretouchIntervalMillis() {
        return pretouchIntervalMillis == null ? 1_000L : pretouchIntervalMillis;
    }



    public SingleChronicleQueueBuilder path(String path) {
        return path(new File(path));
    }

    public SingleChronicleQueueBuilder path(final File path) {
        this.path = path;
        return this;
    }
<commit_msg>added ->  public SingleChronicleQueueBuilder path(final Path path) {                  this.path = path.toFile();                  return this;              }<commit_after>/*
 * Copyright 2016 <EMAIL>
 *
 * 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.single;

import net.openhft.chronicle.bytes.Bytes;
import net.openhft.chronicle.bytes.BytesRingBufferStats;
import net.openhft.chronicle.bytes.BytesStore;
import net.openhft.chronicle.bytes.MappedBytes;
import net.openhft.chronicle.core.Jvm;
import net.openhft.chronicle.core.Maths;
import net.openhft.chronicle.core.OS;
import net.openhft.chronicle.core.io.IORuntimeException;
import net.openhft.chronicle.core.threads.EventLoop;
import net.openhft.chronicle.core.time.SystemTimeProvider;
import net.openhft.chronicle.core.time.TimeProvider;
import net.openhft.chronicle.core.util.ObjectUtils;
import net.openhft.chronicle.core.util.ThrowingBiFunction;
import net.openhft.chronicle.core.util.Updater;
import net.openhft.chronicle.queue.BufferMode;
import net.openhft.chronicle.queue.QueueOffsetSpec;
import net.openhft.chronicle.queue.RollCycle;
import net.openhft.chronicle.queue.RollCycles;
import net.openhft.chronicle.queue.impl.RollingChronicleQueue;
import net.openhft.chronicle.queue.impl.StoreFileListener;
import net.openhft.chronicle.queue.impl.TableStore;
import net.openhft.chronicle.queue.impl.WireStoreFactory;
import net.openhft.chronicle.queue.impl.table.ReadonlyTableStore;
import net.openhft.chronicle.queue.impl.table.SingleTableBuilder;
import net.openhft.chronicle.threads.EventGroup;
import net.openhft.chronicle.threads.TimeoutPauser;
import net.openhft.chronicle.threads.TimingPauser;
import net.openhft.chronicle.wire.*;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.crypto.spec.SecretKeySpec;
import java.io.File;
import java.lang.reflect.Constructor;
import java.nio.file.Path;
import java.time.LocalTime;
import java.time.ZoneId;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Supplier;

import static net.openhft.chronicle.core.pool.ClassAliasPool.CLASS_ALIASES;
import static net.openhft.chronicle.queue.ChronicleQueue.TEST_BLOCK_SIZE;
import static net.openhft.chronicle.queue.impl.single.SingleChronicleQueue.QUEUE_METADATA_FILE;
import static net.openhft.chronicle.wire.WireType.DEFAULT_ZERO_BINARY;
import static net.openhft.chronicle.wire.WireType.DELTA_BINARY;

public class SingleChronicleQueueBuilder implements Cloneable, Marshallable {
    public static final String DEFAULT_ROLL_CYCLE_PROPERTY = "net.openhft.queue.builder.defaultRollCycle";
    private static final Constructor ENTERPISE_QUEUE_CONSTRUCTOR;
    private static final String DEFAULT_EPOCH_PROPERTY = "net.openhft.queue.builder.defaultEpoch";
    private static final Logger LOGGER = LoggerFactory.getLogger(SingleChronicleQueueBuilder.class);

    static {
        CLASS_ALIASES.addAlias(WireType.class);
        CLASS_ALIASES.addAlias(SCQMeta.class, "SCQMeta");
        CLASS_ALIASES.addAlias(SCQRoll.class, "SCQSRoll");
        CLASS_ALIASES.addAlias(SCQIndexing.class, "SCQSIndexing");
        CLASS_ALIASES.addAlias(SingleChronicleQueueStore.class, "SCQStore");
        CLASS_ALIASES.addAlias(TimedStoreRecovery.class);

        {
            Constructor co;
            try {
                co = ((Class) Class.forName("software.chronicle.enterprise.queue.EnterpriseSingleChronicleQueue")).getDeclaredConstructors()[0];
                co.setAccessible(true);
            } catch (Exception e) {
                co = null;
            }
            ENTERPISE_QUEUE_CONSTRUCTOR = co;
        }

    }

    private WireType wireType = WireType.BINARY_LIGHT;
    private Long blockSize;
    private File path;
    private RollCycle rollCycle = loadDefaultRollCycle();
    private Long epoch; // default is 1970-01-01 00:00:00.000 UTC
    private Long bufferCapacity;
    private Integer indexSpacing;
    private Integer indexCount;

    public BufferMode writeBufferMode = BufferMode.None;
    public BufferMode readBufferMode = BufferMode.None;

    private Boolean enableRingBufferMonitoring;
    @Nullable
    private EventLoop eventLoop;
    private WireStoreFactory storeFactory = SingleChronicleQueueBuilder::createStore;
    /**
     * by default logs the performance stats of the ring buffer
     */
    @NotNull
    private Consumer<BytesRingBufferStats> onRingBufferStats = NoBytesRingBufferStats.NONE;
    private TimeProvider timeProvider = SystemTimeProvider.INSTANCE;
    private Supplier<TimingPauser> pauserSupplier = () -> new TimeoutPauser(500_000);
    private Long timeoutMS; // 10 seconds.
    private Integer sourceId;
    private StoreRecoveryFactory recoverySupplier = TimedStoreRecovery.FACTORY;
    private StoreFileListener storeFileListener;

    private Boolean readOnly;
    private Boolean strongAppenders;

    private TableStore<SCQMeta> metaStore;

    // enterprise stuff
    private int deltaCheckpointInterval = -1;
    private Supplier<BiConsumer<BytesStore, Bytes>> encodingSupplier;
    private Supplier<BiConsumer<BytesStore, Bytes>> decodingSupplier;
    private Updater<Bytes> messageInitializer;
    private Consumer<Bytes> messageHeaderReader;
    private SecretKeySpec key;

    private int maxTailers;
    private ThrowingBiFunction<Long, Integer, BytesStore, Exception> bufferBytesStoreCreator;
    private Long pretouchIntervalMillis;
    private LocalTime rollTime;
    private ZoneId rollTimeZone;
    private QueueOffsetSpec queueOffsetSpec;

    public static void addAliases() {
        // static initialiser.
    }
    /*
     * ========================
     * Builders
     * ========================
     */

    /**
     * @return an empty builder
     */
    public static SingleChronicleQueueBuilder builder() {
        return new SingleChronicleQueueBuilder();
    }

    @NotNull
    public static SingleChronicleQueueBuilder builder(@NotNull Path path, @NotNull WireType wireType) {
        return builder(path.toFile(), wireType);
    }

    @NotNull
    public static SingleChronicleQueueBuilder builder(@NotNull File file, @NotNull WireType wireType) {
        SingleChronicleQueueBuilder result = builder().wireType(wireType);
        if (file.isFile()) {
            if (!file.getName().endsWith(SingleChronicleQueue.SUFFIX)) {
                throw new IllegalArgumentException("Invalid file type: " + file.getName());
            }

            LOGGER.warn("Queues should be configured with the queue directory, not a specific filename. Actual file used: {}",
                    file.getParentFile());

            result.path(file.getParentFile());
        } else
            result.path(file);

        return result;
    }

    public static SingleChronicleQueueBuilder single() {
        SingleChronicleQueueBuilder builder = builder();
        builder.wireType(WireType.BINARY_LIGHT);
        return builder;
    }

    public static SingleChronicleQueueBuilder single(@NotNull String basePath) {
        return binary(basePath);
    }

    public static SingleChronicleQueueBuilder single(@NotNull File basePath) {
        return binary(basePath);
    }

    public static SingleChronicleQueueBuilder binary(@NotNull Path path) {
        return binary(path.toFile());
    }

    public static SingleChronicleQueueBuilder binary(@NotNull String basePath) {
        return binary(new File(basePath));
    }

    public static SingleChronicleQueueBuilder binary(@NotNull File basePathFile) {
        return builder(basePathFile, WireType.BINARY_LIGHT);
    }

    public static SingleChronicleQueueBuilder fieldlessBinary(@NotNull File name) {
        return builder(name, WireType.FIELDLESS_BINARY);
    }

    public static SingleChronicleQueueBuilder defaultZeroBinary(@NotNull File basePathFile) {
        return builder(basePathFile, DEFAULT_ZERO_BINARY);
    }

    public static SingleChronicleQueueBuilder deltaBinary(@NotNull File basePathFile) {
        return builder(basePathFile, DELTA_BINARY);
    }

    /**
     * @param name               the file name
     * @param deltaIntervalShift default value of 6, the shift for deltaInterval, the should be a
     *                           number between 0-63 ( inclusive ), default the delta messaging is
     *                           check pointed every 64 messages, so the default {@code
     *                           deltaIntervalShift  == 6}, as {@code 1 << 6 == 64 }
     * @return the EnterpriseChronicleQueueBuilder
     */
    public static SingleChronicleQueueBuilder deltaBinary(@NotNull File name, byte deltaIntervalShift) {
        @NotNull SingleChronicleQueueBuilder ret = deltaBinary(name);

        if (deltaIntervalShift < 0 || deltaIntervalShift > 63)
            throw new IllegalArgumentException("deltaIntervalShift=" + deltaIntervalShift + ", but " +
                    "should be a value between 0-63 inclusive");

        ret.deltaCheckpointInterval(1 << deltaIntervalShift);
        return ret;
    }

    protected SingleChronicleQueueBuilder() {
    }

    public WireStoreFactory storeFactory() {
        return storeFactory;
    }

    public boolean hasPretouchIntervalMillis() {
        return pretouchIntervalMillis != null;
    }

    @NotNull
    public SingleChronicleQueue build() {
        boolean needEnterprise = checkEnterpriseFeaturesRequested();
        preBuild();

        if (needEnterprise)
            return buildEnterprise();

        return new SingleChronicleQueue(this);
    }

    private boolean checkEnterpriseFeaturesRequested() {

        boolean result = false;
        if (readBufferMode != BufferMode.None)
            result = onlyAvailableInEnterprise("Buffering");
        if (writeBufferMode != BufferMode.None)
            result = onlyAvailableInEnterprise("Buffering");
        if (rollTimeZone != null && !rollTimeZone.getId().equals("UTC") && !rollTimeZone.getId().equals("Z"))
            result = onlyAvailableInEnterprise("Non-UTC roll time zone");
        if (wireType == WireType.DELTA_BINARY)
            result = onlyAvailableInEnterprise("Wire type " + wireType.name());
        if (encodingSupplier != null)
            result = onlyAvailableInEnterprise("Encoding");
        if (key != null)
            result = onlyAvailableInEnterprise("Encryption");
        if (hasPretouchIntervalMillis())
            result = onlyAvailableInEnterprise("Pretouching");

        return result;
    }

    private boolean onlyAvailableInEnterprise(final String feature) {
        if (ENTERPISE_QUEUE_CONSTRUCTOR == null)
            LOGGER.warn(feature + " is only supported in Chronicle Queue Enterprise. If you would like to use this feature, please contact <EMAIL> for more information.");
        return true;
    }

    @NotNull
    private SingleChronicleQueue buildEnterprise() {
        if (ENTERPISE_QUEUE_CONSTRUCTOR == null)
            throw new IllegalStateException("Enterprise features requested but Chronicle Queue Enterprise is not in the class path!");

        try {
            return (SingleChronicleQueue) ENTERPISE_QUEUE_CONSTRUCTOR.newInstance(this);
        } catch (Exception e) {
            throw new IllegalStateException("Couldn't create an instance of Enterprise queue", e);
        }

    }

    public SingleChronicleQueueBuilder aesEncryption(@Nullable byte[] keyBytes) {
        if (keyBytes == null) {
            codingSuppliers(null, null);
            return this;
        }
        key = new SecretKeySpec(keyBytes, "AES");
        return this;
    }

    public Updater<Bytes> messageInitializer() {
        return messageInitializer == null ? Bytes::clear : messageInitializer;
    }

    public Consumer<Bytes> messageHeaderReader() {
        return messageHeaderReader == null ? b -> {
        } : messageHeaderReader;
    }

    public SingleChronicleQueueBuilder messageHeader(Updater<Bytes> messageInitializer,
                                                     Consumer<Bytes> messageHeaderReader) {
        this.messageInitializer = messageInitializer;
        this.messageHeaderReader = messageHeaderReader;
        return this;
    }

    public SingleChronicleQueueBuilder rollTime(final LocalTime rollTime) {
        rollTime(rollTime, rollTimeZone);
        return this;
    }

    public ZoneId rollTimeZone() {
        return rollTimeZone;
    }

    public SingleChronicleQueueBuilder rollTimeZone(final ZoneId rollTimeZone) {
        rollTime(rollTime, rollTimeZone);
        return this;
    }

    public SingleChronicleQueueBuilder rollTime(@NotNull final LocalTime rollTime, final ZoneId zoneId) {
        this.rollTime = rollTime;
        this.rollTimeZone = zoneId;
        this.epoch = TimeUnit.SECONDS.toMillis(rollTime.toSecondOfDay());
        this.queueOffsetSpec = QueueOffsetSpec.ofRollTime(rollTime, zoneId);
        return this;
    }

    protected void initializeMetadata() {
        File metapath = metapath();
        validateRollCycle(metapath);
        SCQMeta metadata = new SCQMeta(new SCQRoll(rollCycle(), epoch()), deltaCheckpointInterval(),
                sourceId());
        try {

            boolean readOnly = readOnly();
            metaStore = SingleTableBuilder.binary(metapath, metadata).timeoutMS(timeoutMS())
                    .readOnly(readOnly).validateMetadata(!readOnly).build();
            // check if metadata was overridden
            SCQMeta newMeta = metaStore.metadata();
            if (sourceId() == 0)
                sourceId(newMeta.sourceId());

            if (readOnly && !newMeta.roll().format().equals(rollCycle().format())) {
                // roll cycle changed
                overrideRollCycleForFileNameLength(newMeta.roll().format().length());
            }
        } catch (IORuntimeException ex) {
            // readonly=true and file doesn't exist
            if (OS.isWindows())
                throw ex; // we cant have a read-only table store on windows so we have no option
            // but to throw the ex.
            metaStore = new ReadonlyTableStore<>(metadata);
        }
    }

    private void validateRollCycle(File metapath) {
        if (!metapath.exists()) {
            // no metadata, so we need to check if there're cq4 files and if so try to validate roll cycle
            // the code is slightly brutal and crude but should work for most cases. It will NOT work if files were created with
            // the following cycles: LARGE_HOURLY_SPARSE LARGE_HOURLY_XSPARSE LARGE_DAILY XLARGE_DAILY HUGE_DAILY HUGE_DAILY_XSPARSE
            // for such cases user MUST use correct roll cycle when creating the queue
            String[] list = path.list((d, name) -> name.endsWith(SingleChronicleQueue.SUFFIX));
            if (list != null && list.length > 0) {
                String filename = list[0];
                if (rollCycle().format().length() + 4 != filename.length()) {
                    // probably different roll cycle used
                    overrideRollCycleForFileNameLength(filename.length() - 4);
                }
            }
        }
    }

    private void overrideRollCycleForFileNameLength(int patternLength) {
        for (RollCycles cycle : RollCycles.values()) {
            if (cycle.format().length() == patternLength) {
                LOGGER.warn("Overriding roll cycle to " + cycle);
                rollCycle = cycle;
                break;
            }
        }
    }

    private File metapath() {
        final File storeFilePath;
        if ("".equals(path.getPath())) {
            storeFilePath = new File(QUEUE_METADATA_FILE);
        } else {
            storeFilePath = new File(path, QUEUE_METADATA_FILE);
            path.mkdirs();
        }
        return storeFilePath;
    }

    @NotNull
    QueueLock queueLock() {
        return isQueueReplicationAvailable() && !readOnly() ? new TSQueueLock(metaStore, pauserSupplier(), timeoutMS() * 3 / 2) : new NoopQueueLock();
    }

    @NotNull
    WriteLock writeLock() {
        return readOnly() ? new ReadOnlyWriteLock() : new TableStoreWriteLock(metaStore, pauserSupplier(), timeoutMS() * 3 / 2);
    }

    public SingleChronicleQueueBuilder enablePreloader(final long pretouchIntervalMillis) {
        this.pretouchIntervalMillis = pretouchIntervalMillis;
        return this;
    }

    public int deltaCheckpointInterval() {
        return deltaCheckpointInterval == -1 ? 64 : deltaCheckpointInterval;
    }

    public QueueOffsetSpec queueOffsetSpec() {
        return queueOffsetSpec == null ? QueueOffsetSpec.ofNone() : queueOffsetSpec;
    }

    TableStore<SCQMeta> metaStore() {
        return metaStore;
    }

    /**
     * RingBuffer tailers need to be preallocated. Only set this if using readBufferMode=Asynchronous.
     * By default 1 tailer will be created for the user.
     *
     * @param maxTailers number of tailers that will be required from this queue, not including the draining tailer
     * @return this
     */
    public SingleChronicleQueueBuilder maxTailers(int maxTailers) {
        this.maxTailers = maxTailers;
        return this;
    }

    /**
     * maxTailers
     *
     * @return number of tailers that will be required from this queue, not including the draining tailer
     */
    public int maxTailers() {
        return maxTailers;
    }

    public SingleChronicleQueueBuilder bufferBytesStoreCreator(ThrowingBiFunction<Long, Integer, BytesStore, Exception> bufferBytesStoreCreator) {
        this.bufferBytesStoreCreator = bufferBytesStoreCreator;
        return this;
    }

    /**
     * Creator for BytesStore for underlying ring buffer. Allows visibility of RB's data to be controlled.
     * See also RB_BYTES_STORE_CREATOR_NATIVE, RB_BYTES_STORE_CREATOR_MAPPED_FILE
     *
     * @return
     */
    @Nullable
    public ThrowingBiFunction<Long, Integer, BytesStore, Exception> bufferBytesStoreCreator() {
        return bufferBytesStoreCreator;
    }


    public long pretouchIntervalMillis() {
        return pretouchIntervalMillis == null ? 1_000L : pretouchIntervalMillis;
    }



    public SingleChronicleQueueBuilder path(String path) {
        return path(new File(path));
    }

    public SingleChronicleQueueBuilder path(final File path) {
        this.path = path;
        return this;
    }

    public SingleChronicleQueueBuilder path(final Path path) {
        this.path = path.toFile();
        return this;
    }

{'fraction_non_alphanumeric': 0.086410635155096, 'fraction_numerical': 0.003851023422663009, 'mean_word_length': 2.9933642300400254, 'pattern_counts': {'":': 0, '<': 53, '<?xml version=': 0, '>': 54, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1895172', 'n_tokens_mistral': 10553, 'n_tokens_neox': 10090, 'n_words': 2519}