@Test
public void test() {
    File dir = getTmpDir();
    try (final ChronicleQueue queue = SingleChronicleQueueBuilder.builder(dir, wireType)
            .testBlockSize()
            .build()) {

        final ExcerptTailer tailer = queue.createTailer();
        assertFalse(tailer.readDocument(r -> r.read(TestKey.test).int32()));

        final ExcerptAppender appender = queue.acquireAppender();
        appender.writeDocument(w -> w.write(TestKey.test).int32(1));
        Jvm.pause(100);
        assertTrue(tailer.readDocument(r -> r.read(TestKey.test).int32()));
    }
}
 

@Override
public void refresh() {
    throwExceptionIfClosed();

    if (readOnly) {
        return;
    }
    while (true) {
        long currentMax = maxCycleValue.getVolatileValue();
        final File[] queueFiles = queuePath.toFile().
                listFiles((d, f) -> f.endsWith(SingleChronicleQueue.SUFFIX));
        int min = UNSET_MIN_CYCLE;
        int max = UNSET_MAX_CYCLE;
        if (queueFiles != null) {
            for (File queueFile : queueFiles) {
                min = Math.min(fileToCycleFunction.applyAsInt(queueFile), min);
                max = Math.max(fileToCycleFunction.applyAsInt(queueFile), max);
            }
        }
        minCycleValue.setOrderedValue(min);
        if (maxCycleValue.compareAndSwapValue(currentMax, max))
            break;
        Jvm.nanoPause();
    }
}
 

StoreAppender(@NotNull final SingleChronicleQueue queue,
              @NotNull final WireStorePool storePool,
              final boolean checkInterrupts) {
    this.queue = queue;
    this.storePool = storePool;
    this.checkInterrupts = checkInterrupts;
    this.writeLock = queue.writeLock();
    this.context = new StoreAppenderContext();

    // always put references to "this" last.
    queue.addCloseListener(this);

    queue.cleanupStoreFilesWithNoData();
    int cycle = queue.cycle();
    int lastCycle = queue.lastCycle();
    if (lastCycle != cycle && lastCycle >= 0)
        // ensure that the EOF is written on the last cycle
        setCycle2(lastCycle, false);
    finalizer = Jvm.isResourceTracing() ? new Finalizer() : null;
}
 

private static void logToStandardOutMessageReceivedInERROR(@NotNull final Wire wire) {
    @NotNull final Bytes<?> bytes = wire.bytes();

    final long position = bytes.writePosition();
    final long limit = bytes.writeLimit();
    try {
        try {

            LOG.info("\nreceives IN ERROR:\n" +
                    "```yaml\n" +
                    Wires.fromSizePrefixedBlobs(wire) +
                    "```\n");
            YamlLogging.title = "";
            YamlLogging.writeMessage("");

        } catch (Exception e) {
            String x = Bytes.toString(bytes);
            Jvm.warn().on(TcpChannelHub.class, x, e);
        }
    } finally {
        bytes.writeLimit(limit);
        bytes.writePosition(position);
    }
}
 

boolean checkWritePositionHeaderNumber() {
    if (wire == null || wire.headerNumber() == Long.MIN_VALUE) return true;
    try {
        long pos = positionOfHeader;

        long seq1 = queue.rollCycle().toSequenceNumber(wire.headerNumber() + 1) - 1;
        long seq2 = store.sequenceForPosition(this, pos, true);

        if (seq1 != seq2) {
            String message = "~~~~~~~~~~~~~~ " +
                    "thread: " + Thread.currentThread().getName() +
                    " pos: " + pos +
                    " header: " + wire.headerNumber() +
                    " seq1: " + seq1 +
                    " seq2: " + seq2;
            AssertionError ae = new AssertionError(message);
            ae.printStackTrace();
            throw ae;
        }
    } catch (Exception e) {
        // TODO FIX
        Jvm.warn().on(getClass(), e);
        throw Jvm.rethrow(e);
    }
    return true;
}
 

public StartAndMonitor(ChronicleQueue queue, String name, int writePauseMs, int sleepBetweenMillis) {
            final SlowToSerialiseAndDeserialise object = new SlowToSerialiseAndDeserialise(writePauseMs);
            Thread thread = new Thread(() -> {
                try {
                    while (running.get()) {
                        long loopStart = System.nanoTime();
                        final ExcerptAppender appender = queue.acquireAppender();
//                        System.out.println("about to open");
                        try (final DocumentContext ctx = appender.writingDocument()) {
//                            System.out.println("about to write");
                            ctx.wire().getValueOut().marshallable(object);
//                            System.out.println("about to close");
                        }
//                        System.out.println("closed");
                        long timeTaken = System.nanoTime() - loopStart;
                        histo.sampleNanos(timeTaken);
                        Jvm.pause(sleepBetweenMillis);
                    }
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }, name);
            thread.start();
        }
 

private void logToStandardOutMessageSent(@NotNull final WireOut wire, @NotNull final ByteBuffer outBuffer) {
    if (!YamlLogging.showClientWrites())
        return;

    @NotNull Bytes<?> bytes = wire.bytes();

    try {

        if (bytes.readRemaining() > 0)
            LOG.info(((!YamlLogging.title.isEmpty()) ? "### " + YamlLogging
                    .title + "\n" : "") + "" +
                    YamlLogging.writeMessage() + (YamlLogging.writeMessage().isEmpty() ?
                    "" : "\n\n") +
                    "sends:\n\n" +
                    "```yaml\n" +
                    Wires.fromSizePrefixedBlobs(bytes) +
                    "```");
        YamlLogging.title = "";
        YamlLogging.writeMessage("");

    } catch (Exception e) {
        Jvm.warn().on(TcpChannelHub.class, Bytes.toString(bytes), e);
    }
}
 

@SuppressWarnings("resource")
@Override
public Throwable call() {
    ChronicleQueue queue0 = null;
    try (ChronicleQueue queue = queueBuilder(path).build()) {
        queue0 = queue;
        ExcerptAppender appender = queue.acquireAppender();
        System.out.println("Starting pretoucher");
        while (!Thread.currentThread().isInterrupted() && !queue.isClosed()) {
            Jvm.pause(50);
            appender.pretouch();
        }
    } catch (Throwable e) {
        if (queue0 != null && queue0.isClosed())
            return null;
        exception = e;
        return e;
    }
    return null;
}
 

protected void writeBytesInternal(final long index, @NotNull final BytesStore bytes, boolean metadata) {
    final int cycle = queue.rollCycle().toCycle(index);

    if (wire == null)
        setCycle2(cycle, true);
    else if (this.cycle < cycle)
        rollCycleTo(cycle);

    boolean rollbackDontClose = index != wire.headerNumber() + 1;
    if (rollbackDontClose) {
        if (index > wire.headerNumber() + 1)
            throw new IllegalStateException("Unable to move to index " + Long.toHexString(index) + " beyond the end of the queue, current: " + Long.toHexString(wire.headerNumber()));
        Jvm.warn().on(getClass(), "Trying to overwrite index " + Long.toHexString(index) + " which is before the end of the queue", new StackTrace());
        return;
    }
    writeBytesInternal(bytes, metadata);
}
 

private void log0(@NotNull ByteBuffer bytes, int start, int end) {
    @NotNull final StringBuilder sb = new StringBuilder(desc);
    sb.append(" len: ").append(end - start).append(" - ");
    if (end - start > 128) {
        for (int i = start; i < start + 64; i++)
            appendByte(bytes, sb, i);
        sb.append(" ... ");
        for (int i = end - 64; i < end; i++)
            appendByte(bytes, sb, i);
    } else {
        for (int i = start; i < end; i++)
            appendByte(bytes, sb, i);
    }

    Jvm.debug().on(getClass(), sb.toString());
}
 

/**
 * Extra checks that get done for all test cases.
 * <p>
 * <p>Triggers test case Assert.failure if any thread assertions have Assert.failed,
 * by rethrowing, in the test harness thread, any exception recorded
 * earlier by threadRecordFailure.
 * <p>
 * <p>Triggers test case Assert.failure if interrupt status is set in the main thread.
 */
@After
public void tearDown() throws InterruptedException {
    Throwable t = threadFailure.getAndSet(null);
    if (t != null) {
        if (t instanceof Error)
            throw (Error) t;
        else if (t instanceof RuntimeException)
            throw (RuntimeException) t;
        else if (t instanceof Exception)
            throw Jvm.rethrow(t);
        else {
            AssertionFailedError afe =
                    new AssertionFailedError(t.toString());
            afe.initCause(t);
            throw afe;
        }
    }

    if (Thread.interrupted())
        throw new AssertionFailedError("interrupt status set in main thread");

    checkForkJoinPoolThreadLeaks();
    System.gc();
}
 

public <R> R check(Call instance) {
    R r1 = null;
    R r2 = null;
    for (int i = 0; i < 50; i++) {
        r1 = (R) instance.method(map1);
        r2 = (R) instance.method(map2);

        if (r1 != null && r1.equals(r2))
            return r1;

        if (i > 30) {
            Jvm.pause(i);
        } else {
            Thread.yield();
        }
    }

    Assert.assertEquals(map1, map2);
    System.out.print(map1);
    System.out.print(map2);

    if (r1 != null)
        Assert.assertEquals(r1.toString(), r2.toString());

    return (R) r1;
}
 

public static void waitTillEqual(Map map1, Map map2, int timeOutMs) {
    int numberOfTimesTheSame = 0;
    long startTime = System.currentTimeMillis();
    for (int t = 0; t < timeOutMs + 100; t++) {
        // not map1.equals(map2), the reason is described above
        if (map1.equals(map2)) {
            numberOfTimesTheSame++;
            Jvm.pause(1);
            if (numberOfTimesTheSame == 10) {
                System.out.println("same");
                break;
            }
        }
        Jvm.pause(1);
        if (System.currentTimeMillis() - startTime > timeOutMs)
            break;
    }
}
 

private void waitForTheHeaderToBeBuilt(@NotNull Bytes bytes) throws IOException {
    for (int i = 0; i < 1000; i++) {
        long magic = bytes.readVolatileLong(MAGIC_OFFSET);
        if (magic == BUILDING) {
            try {
                Jvm.pause(10);
            } catch (InterruptedException e) {
                throw new IOException("Interrupted waiting for the header to be built");
            }
        } else if (magic == QUEUE_CREATED) {
            return;

        } else {
            throw new AssertionError("Invalid magic number " + Long.toHexString(magic) + " in file " + name());
        }
    }
    throw new AssertionError("Timeout waiting to build the file " + name());
}
 

/**
 * This method does not update the index, as indexes are not used for meta data
 *
 * @param buffer
 * @return the addressForRead of the appended data
 */
private long appendMetaDataReturnAddress(@NotNull Bytes buffer) {
    long length = checkRemainingForAppend(buffer);

    LongValue writeByte = header.writeByte();
    long lastByte = writeByte.getVolatileValue();

    for (; ; ) {
        if (bytes.compareAndSwapInt(lastByte, 0, NOT_COMPLETE | (int) length)) {
            long lastByte2 = lastByte + 4 + buffer.remaining();
            bytes.write(lastByte + 4, buffer);
            writeByte.setOrderedValue(lastByte2);
            bytes.writeOrderedInt(lastByte, (int) (META_DATA | length));
            return lastByte;
        }
        int length2 = length30(bytes.readVolatileInt());
        bytes.skip(length2);
        try {
            Jvm.checkInterrupted();
        } catch (InterruptedException e) {
            throw new InterruptedRuntimeException(e);
        }
    }
}
 

@Override
public boolean writeEOF(@NotNull Wire wire, long timeoutMS) {
    throwExceptionIfClosed();

    String fileName = mappedFile.file().getAbsolutePath();

    // just in case we are about to release this
    if (wire.bytes().tryReserve(this)) {
        try {
            return writeEOFAndShrink(wire, timeoutMS);

        } finally {
            wire.bytes().release(this);
        }
    }

    try (MappedBytes bytes = MappedBytes.mappedBytes(mappedFile.file(), mappedFile.chunkSize())) {
        Wire wire0 = WireType.valueOf(wire).apply(bytes);
        return writeEOFAndShrink(wire0, timeoutMS);

    } catch (Exception e) {
        Jvm.warn().on(getClass(), "unable to write the EOF file=" + fileName, e);
        return false;
    }
}
 

@Override
public long lastIndexAppended() {
    throwExceptionIfClosed();

    if (lastIndex != Long.MIN_VALUE)
        return lastIndex;

    if (lastPosition == Long.MIN_VALUE || wire == null) {
        throw new IllegalStateException("nothing has been appended, so there is no last index");
    }

    try {
        long sequenceNumber = store.sequenceForPosition(this, lastPosition, true);
        long index = queue.rollCycle().toIndex(lastCycle, sequenceNumber);
        lastIndex(index);
        return index;
    } catch (Exception e) {
        throw Jvm.rethrow(e);
    }
}
 

private boolean headerNumberCheck(@NotNull final AbstractWire wire) {

        wire.headNumberCheck((actual, position) -> {
            try {
                final long expecting = store.sequenceForPosition(this, position, false);
                if (actual == expecting)
                    return true;
                Jvm.warn().on(getClass(), new AssertionError("header number check failed " +
                        "expecting=" + expecting +
                        "  !=  actual=" + actual));

                return false;
            } catch (Exception e) {
                Jvm.warn().on(getClass(), "", e);
                return false;
            }
        });

        return true;
    }
 

public static void main(String[] args) {
    System.out.println("\nWaiting for messages");
    try (ChronicleQueue q = SingleChronicleQueueBuilder.binary("in").rollCycle(RollCycles.TEST8_DAILY).build()) {
        OMSIn logging = Mocker.logging(OMSIn.class, "read - ", System.out);
        MethodReader reader = q.createTailer().methodReader(logging);
        while (true) {
            if (!reader.readOne())
                Jvm.pause(50);
        }
    }
}
 

public static void runTest(String path) {
    String input = "src/test/resources/" + path + "/in.yaml";
    String output = "src/test/resources/" + path + "/out.yaml";
    TextMethodTester tester = new TextMethodTester<>(input, OMSImpl::new, OMSOut.class, output);
    try {
        tester.run();
        assertEquals(tester.expected(), tester.actual());
    } catch (IOException e) {
        Jvm.rethrow(e);
    }
}
 

AvroHelper() {
    try {
        schema = new Schema.Parser().parse(getClass().getClassLoader().getResourceAsStream("user.avsc"));
        datumWriter = new GenericDatumWriter<>(schema);
        datumReader = new GenericDatumReader<>(schema);
    } catch (IOException ex) {
        Jvm.rethrow(ex);
    }
}
 

public town.lost.processor.events.EventWithHistory history(net.openhft.chronicle.wire.VanillaMessageHistory arg0) {
  Method _method_ = this.methods[0];
  Object[] _a_ = this.argsTL.get()[1];
  _a_[0] = arg0;
  try {
    return (town.lost.processor.events.EventWithHistory) handler.get().invoke(this, _method_, _a_);
  } catch (Throwable throwable) {
    throw Jvm.rethrow(throwable);
  }
}
 

public void eventOne(town.lost.processor.events.EventOne arg0) {
  Method _method_ = this.methods[1];
  Object[] _a_ = this.argsTL.get()[1];
  _a_[0] = arg0;
  try {
    handler.get().invoke(this, _method_, _a_);
  } catch (Throwable throwable) {
    throw Jvm.rethrow(throwable);
  }
}
 

@Override
public Throwable call() {
    ChronicleQueue queue = writerQueue(path);
    try (final ExcerptAppender appender = queue.acquireAppender()) {
        Jvm.pause(random.nextInt(DELAY_WRITER_RANDOM_MS));
        final long startTime = System.nanoTime();
        int loopIteration = 0;
        while (true) {
            final int value = write(appender);

            while (System.nanoTime() < (startTime + (loopIteration * SLEEP_PER_WRITE_NANOS))) {
                // spin
            }
            loopIteration++;

            if (value >= expectedNumberOfMessages) {
                return null;
            }
        }
    } catch (Throwable e) {
        exception = e;
        return e;
    } finally {
        if (queue != sharedWriterQueue)
            queue.close();
    }
}
 

public void close() {
  Method _method_ = this.methods[3];
  Object[] _a_ = this.argsTL.get()[0];
  try {
    handler.get().invoke(this, _method_, _a_);
  } catch (Throwable throwable) {
    throw Jvm.rethrow(throwable);
  }
}
 

@After
public void after() {
    threadDump.assertNoNewThreads();

    if (Jvm.hasException(exceptionKeyIntegerMap)) {
        Jvm.dumpException(exceptionKeyIntegerMap);
        fail();
    }
    Jvm.resetExceptionHandlers();
}
 

public InMemoryLongColumn(TimeSeries timeSeries, String name, BytesLongLookup lookup, long capacity) {
    super(timeSeries, name);
    this.lookup = lookup;
    long value = lookup.sizeFor(capacity);
    this.bytes = Jvm.isDebug()
            ? Bytes.wrapForRead(ByteBuffer.allocateDirect(Math.toIntExact(value)))
            : NativeBytesStore.lazyNativeBytesStoreWithFixedCapacity(value);
}
 

@Override
public void ensureCapacity(long capacity) {
    long cap = lookup.sizeFor(capacity);
    if (cap > bytes.realCapacity()) {
        long value = lookup.sizeFor(capacity);
        BytesStore bytes2 = Jvm.isDebug()
                ? Bytes.wrapForRead(ByteBuffer.allocateDirect(Math.toIntExact(value)))
                : NativeBytesStore.lazyNativeBytesStoreWithFixedCapacity(value);
        bytes2.write(0, bytes);
        bytes.release();
        bytes = bytes2;
    }
}
 

@Override
protected void onRead(@NotNull DocumentContext in, @NotNull WireOut out) {
    for (; ; ) {
        long pos = in.wire().bytes().readPosition();
        if (!reader.readOne())
            return;
        if (pos <= in.wire().bytes().readPosition()) {
            Jvm.warn().on(getClass(), "unable to parse data at the end of message " + in.wire().bytes().toDebugString());
            return;
        }
    }
}
 

private static void logYaml(@NotNull final DocumentContext dc) {
    if (YamlLogging.showServerWrites() || YamlLogging.showServerReads())
        try {
            LOG.info("\nDocumentContext:\n" +
                    Wires.fromSizePrefixedBlobs(dc));

        } catch (Exception e) {
            Jvm.warn().on(WireOutPublisher.class, "\nServer Sends ( corrupted ) :\n" +
                    dc.wire().bytes().toDebugString());
        }
}