import net.openhft.chronicle.core.Jvm; //导入依赖的package包/类
public void enableMarketData(boolean enable) {
    if(!enable)
        isStopped = true;
    else {
        isStopped = false;
        new Thread(() -> {
            while (!isStopped) {
                Random r = new Random();
                int[] fourRandomNumbers = r.ints(4, 0, 1001).toArray();

                MarketData md  = new MarketData("GBP/USD",
                        fourRandomNumbers[0],
                        fourRandomNumbers[1],
                        fourRandomNumbers[2],
                        fourRandomNumbers[3]);

                gatewayPublisher.marketData(md);

                System.out.println(md);
                Jvm.pause(1000);
            }
        }).start();
    }
}
 

import net.openhft.chronicle.core.Jvm; //导入依赖的package包/类
private static void logToStandardOutMessageReceivedInERROR(@NotNull 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);
    }
}
 

import net.openhft.chronicle.core.Jvm; //导入依赖的package包/类
/**
 * used to signal to the server that the client is going to drop the connection, and waits up to
 * one second for the server to acknowledge the receipt of this message
 */
private void sendCloseMessage() {

    this.lock(() -> {

        TcpChannelHub.this.writeMetaDataForKnownTID(0, outWire, null, 0);

        TcpChannelHub.this.outWire.writeDocument(false, w ->
                w.writeEventName(EventId.onClientClosing).text(""));

    }, TryLock.LOCK);

    // wait up to 1 seconds to receive an close request acknowledgment from the server
    try {
        final boolean await = receivedClosedAcknowledgement.await(1, TimeUnit.SECONDS);
        if (!await)
            Jvm.debug().on(getClass(), "SERVER IGNORED CLOSE REQUEST: shutting down the client anyway as the " +
                    "server did not respond to the close() request.");
    } catch (InterruptedException ignore) {
        Thread.currentThread().interrupt();
    }
}
 

import net.openhft.chronicle.core.Jvm; //导入依赖的package包/类
private void logToStandardOutMessageSent(@NotNull WireOut wire, @NotNull 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(getClass(), Bytes.toString(bytes), e);
    }
}
 

import net.openhft.chronicle.core.Jvm; //导入依赖的package包/类
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());
}
 

import net.openhft.chronicle.core.Jvm; //导入依赖的package包/类
/**
 * Delays, via Thread.sleep, for the given millisecond delay, but
 * if the sleep is shorter than specified, may re-sleep or yield
 * until time elapses.
 */
static void delay(long millis) throws InterruptedException {
    long startTime = System.nanoTime();
    long ns = millis * 1000 * 1000;
    for (; ; ) {
        if (millis > 0L)
            Jvm.pause(millis);
        else // too short to sleep
            Thread.yield();
        long d = ns - (System.nanoTime() - startTime);
        if (d > 0L)
            millis = d / (1000 * 1000);
        else
            break;
    }
}
 

import net.openhft.chronicle.core.Jvm; //导入依赖的package包/类
/**
 * Extra checks that get done for all test cases.
 *
 * <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>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();
}
 

import net.openhft.chronicle.core.Jvm; //导入依赖的package包/类
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;
}
 

import net.openhft.chronicle.core.Jvm; //导入依赖的package包/类
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;
        }
    }
 

import net.openhft.chronicle.core.Jvm; //导入依赖的package包/类
public static void main(String[] args) {
    String input = args.length > 0 ? args[0] : OS.TMP + "/input";
    String output = args.length > 1 ? args[1] : OS.TMP + "/output";

    AtomicLong lastUpdate = new AtomicLong(System.currentTimeMillis() + 1000);
    Thread thread = new Thread(() -> {
        ChronicleQueue outputQ = SingleChronicleQueueBuilder.binary(output).build();
        MethodReader reader = outputQ.createTailer().methodReader((HelloReplier) err::println);
        while (!Thread.interrupted()) {
            if (reader.readOne()) {
                lastUpdate.set(System.currentTimeMillis());
            } else {
                Jvm.pause(10);
            }
        }
    });
    thread.setDaemon(true);
    thread.start();

    ChronicleQueue inputQ = SingleChronicleQueueBuilder.binary(input).build();
    HelloWorld helloWorld = inputQ.createAppender().methodWriter(HelloWorld.class);

    Scanner scanner = new Scanner(System.in);
    while (true) {
        while (System.currentTimeMillis() < lastUpdate.get() + 30)
            Thread.yield();

        out.print("Chat ");
        out.flush();
        if (!scanner.hasNextLine())
            break;
        String line = scanner.nextLine();
        helloWorld.hello(line);
        lastUpdate.set(System.currentTimeMillis());
    }
    out.print("Bye");
}
 

import net.openhft.chronicle.core.Jvm; //导入依赖的package包/类
public static void main(String[] args) {
    String path = "queue";
    SingleChronicleQueue queue = SingleChronicleQueueBuilder.binary(path).build();
    ExcerptTailer tailer = queue.createTailer();

    while (true) {
        String text = tailer.readText();
        if (text == null)
            Jvm.pause(10);
        else
            System.out.println(text);

    }
}
 

import net.openhft.chronicle.core.Jvm; //导入依赖的package包/类
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);
}
 

import net.openhft.chronicle.core.Jvm; //导入依赖的package包/类
@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;
    }
}
 

import net.openhft.chronicle.core.Jvm; //导入依赖的package包/类
@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;
        }
    }
}
 

import net.openhft.chronicle.core.Jvm; //导入依赖的package包/类
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());
        }
}
 

import net.openhft.chronicle.core.Jvm; //导入依赖的package包/类
private void resizeInWire(long size) {
    @NotNull final Bytes<?> bytes = inWire.bytes();
    if (size > bytes.realCapacity()) {
        Jvm.debug().on(getClass(), Integer.toHexString(System.identityHashCode(bytes)) + " resized to: " + size);
        bytes.ensureCapacity(size);
    }
}
 

import net.openhft.chronicle.core.Jvm; //导入依赖的package包/类
private synchronized void clean() {

        if (isCleaned)
            return;
        isCleaned = true;
        final long usedDirectMemory = Jvm.usedDirectMemory();
        IOTools.clean(inBBB.underlyingObject());
        IOTools.clean(outBBB.underlyingObject());

        if (usedDirectMemory == Jvm.usedDirectMemory())
            Jvm.warn().on(getClass(), "nothing cleaned");

    }
 

import net.openhft.chronicle.core.Jvm; //导入依赖的package包/类
private void handleIOE(@NotNull IOException e, final boolean clientIntentionallyClosed,
                       @Nullable HeartbeatListener heartbeatListener) {
    try {

        if (clientIntentionallyClosed)
            return;
        if (e.getMessage() != null && e.getMessage().startsWith("Connection reset by peer"))
            LOG.trace("", e.getMessage());
        else if (e.getMessage() != null && e.getMessage().startsWith("An existing connection " +
                "was forcibly closed"))
            Jvm.debug().on(getClass(), e.getMessage());

        else if (!(e instanceof ClosedByInterruptException))
            Jvm.warn().on(getClass(), "", e);

        // The remote server has sent you a RST packet, which indicates an immediate dropping of the connection,
        // rather than the usual handshake. This bypasses the normal half-closed state transition.
        // I like this description: "Connection reset by peer" is the TCP/IP equivalent
        // of slamming the phone back on the hook.

        if (heartbeatListener != null)
            heartbeatListener.onMissedHeartbeat();

    } finally {
        closeSC();
    }
}
 

import net.openhft.chronicle.core.Jvm; //导入依赖的package包/类
public static void closeAllHubs() {
    @NotNull final TcpChannelHub[] hubsArr = hubs.toArray(new TcpChannelHub[hubs.size()]);
    for (@NotNull TcpChannelHub hub : hubsArr) {
        if (hub.isClosed())
            continue;

        Jvm.debug().on(TcpChannelHub.class, "Closing " + hub);
        hub.close();
    }
    hubs.clear();
}
 

import net.openhft.chronicle.core.Jvm; //导入依赖的package包/类
private static void logToStandardOutMessageReceived(@NotNull Wire wire) {
    @NotNull final Bytes<?> bytes = wire.bytes();

    if (!YamlLogging.showClientReads())
        return;

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

            //         LOG.info("Bytes.toString(bytes)=" + Bytes.toString(bytes));
            LOG.info("\nreceives:\n" +
                    "```yaml\n" +
                    Wires.fromSizePrefixedBlobs(wire) +
                    "```\n");
            YamlLogging.title = "";
            YamlLogging.writeMessage("");

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