private void writeTimeout(@Nullable ByteBuffer outBuffer, long writeTime) {
    for (@NotNull Map.Entry<Thread, StackTraceElement[]> entry : Thread.getAllStackTraces().entrySet()) {
        Thread thread = entry.getKey();
        if (thread.getThreadGroup().getName().equals("system"))
            continue;
        @NotNull StringBuilder sb = new StringBuilder();
        sb.append("\n========= THREAD DUMP =========\n");
        sb.append(thread).append(" ").append(thread.getState());
        Jvm.trimStackTrace(sb, entry.getValue());
        sb.append("\n");
        Jvm.warn().on(TcpChannelHub.class, sb.toString());
    }

    closeSocket();

    throw new IORuntimeException("Took " + writeTime + " ms " +
            "to perform a write, remaining= " + outBuffer.remaining());
}
 

/**
 * blocks until there is a connection
 */
public void checkConnection() {
    long start = Time.currentTimeMillis();

    while (clientChannel == null) {

        tcpSocketConsumer.checkNotShuttingDown();

        if (start + timeoutMs > Time.currentTimeMillis())
            try {
                condition.await(1, TimeUnit.MILLISECONDS);

            } catch (InterruptedException e) {
                throw new IORuntimeException("Interrupted");
            }
        else
            throw new IORuntimeException("Not connected to " + socketAddressSupplier);
    }

    if (clientChannel == null)
        throw new IORuntimeException("Not connected to " + socketAddressSupplier);
}
 

@Override
public long index() throws IORuntimeException {
    throwExceptionIfClosed();

    if (this.wire.headerNumber() == Long.MIN_VALUE) {
        try {
            wire.headerNumber(queue.rollCycle().toIndex(cycle, store.lastSequenceNumber(StoreAppender.this)));
            long headerNumber0 = wire.headerNumber();
            assert (((AbstractWire) this.wire).isInsideHeader());
            return isMetaData() ? headerNumber0 : headerNumber0 + 1;
        } catch (IOException e) {
            throw new IORuntimeException(e);
        }
    }

    return isMetaData() ? Long.MIN_VALUE : this.wire.headerNumber() + 1;
}
 

@Override
default void readMarshallable(@NotNull WireIn wire) throws IORuntimeException {
    userId(wire.read(EventId.userId).text());
    domain(wire.read(EventId.domain).text());
    sessionMode(wire.read(EventId.sessionMode).object(SessionMode.class));
    securityToken(wire.read(EventId.securityToken).text());
    @Nullable final String uid = wire.read(EventId.clientId).text();
    if (uid != null)
        clientId(UUID.fromString(uid));
    wireType(wire.read(EventId.wireType).object(WireType.class));
    hostId(wire.read(EventId.hostId).int8());
}
 

public TcpEventHandler(@NotNull final T nc, final TcpHandlerBias bias) {
    this.sc = ISocketChannel.wrapUnsafe(nc.socketChannel().socketChannel());
    this.scToString = sc.toString();
    this.nc = nc;
    this.bias = bias.get();
    try {
        sc.configureBlocking(false);
        Socket sock = sc.socket();
        // TODO: should have a strategy for this like ConnectionNotifier
        if (!DISABLE_TCP_NODELAY)
            sock.setTcpNoDelay(true);

        if (TCP_BUFFER >= 64 << 10) {
            sock.setReceiveBufferSize(TCP_BUFFER);
            sock.setSendBufferSize(TCP_BUFFER);

            checkBufSize(sock.getReceiveBufferSize(), "recv");
            checkBufSize(sock.getSendBufferSize(), "send");
        }
    } catch (IOException e) {
        if (isClosed() || !sc.isOpen())
            throw new IORuntimeException(e);
        Jvm.warn().on(getClass(), e);
    }

    //We have to provide back pressure to restrict the buffer growing beyond,2GB because it reverts to
    // being Native bytes, we should also provide back pressure if we are not able to keep up
    inBBB = Bytes.elasticByteBuffer(TCP_BUFFER + OS.pageSize(), max(TCP_BUFFER + OS.pageSize(), DEFAULT_MAX_MESSAGE_SIZE));
    outBBB = Bytes.elasticByteBuffer(TCP_BUFFER, max(TCP_BUFFER, DEFAULT_MAX_MESSAGE_SIZE));

    // must be set after we take a slice();
    outBBB.underlyingObject().limit(0);
    readLog = new NetworkLog(this.sc, "read");
    writeLog = new NetworkLog(this.sc, "write");
    nbWarningEnabled = Jvm.warn().isEnabled(getClass());
    statusMonitorEventHandler = new StatusMonitorEventHandler(getClass());
    if (FIRST_HANDLER.compareAndSet(false, true))
        warmUp();
}
 

protected long sendEvent(final long startTime,
                         @NotNull final WireKey eventId,
                         @Nullable final WriteValue consumer) {
    long tid;
    if (hub.outBytesLock().isHeldByCurrentThread())
        throw new IllegalStateException("Cannot view map while debugging");

    try {
        final boolean success = hub.outBytesLock().tryLock(10, TimeUnit.SECONDS);
        if (!success)
            throw new IORuntimeException("failed to obtain write lock");
    } catch (InterruptedException e) {
        throw new IORuntimeException(e);
    }

    try {

        tid = writeMetaDataStartTime(startTime);

        Wire wire = hub.outWire();
        try (DocumentContext dc = wire.writingDocument()) {

            @NotNull final ValueOut valueOut = wire.writeEventName(eventId);

            if (consumer == null)
                valueOut.marshallable(WriteMarshallable.EMPTY);
            else
                consumer.writeValue(valueOut);
        }

        hub.writeSocket(wire, true, false);
    } finally {
        hub.outBytesLock().unlock();
    }
    return tid;
}
 

@NotNull
@Override
public byte[] read(@NotNull Bytes in, long size, @Nullable byte[] using) {
    if (size < 0L || size > (long) Integer.MAX_VALUE) {
        throw new IORuntimeException("byte[] size should be non-negative int, " +
                size + " given. Memory corruption?");
    }
    int arrayLength = (int) size;
    if (using == null || arrayLength != using.length)
        using = new byte[arrayLength];
    in.read(using);
    return using;
}
 

@NotNull
@Override
public List<Point> read(@NotNull Bytes in, long size, List<Point> using) {
    if (size % ELEMENT_SIZE != 0) {
        throw new IORuntimeException("Bytes size should be a multiple of " + ELEMENT_SIZE +
                ", " + size + " read");
    }
    long listSizeAsLong = size / ELEMENT_SIZE;
    if (listSizeAsLong > Integer.MAX_VALUE) {
        throw new IORuntimeException("List size couldn't be more than " + Integer.MAX_VALUE +
                ", " + listSizeAsLong + " read");
    }
    int listSize = (int) listSizeAsLong;
    if (using == null) {
        using = new ArrayList<>(listSize);
        for (int i = 0; i < listSize; i++) {
            using.add(null);
        }
    } else if (using.size() < listSize) {
        while (using.size() < listSize) {
            using.add(null);
        }
    } else if (using.size() > listSize) {
        using.subList(listSize, using.size()).clear();
    }
    for (int i = 0; i < listSize; i++) {
        Point point = using.get(i);
        if (point == null)
            using.set(i, point = new Point());
        point.x = in.readDouble();
        point.y = in.readDouble();
    }
    return using;
}
 

@Override
public void readMarshallable(BytesIn bytes) throws IORuntimeException {
    int len = Maths.toUInt31(bytes.readStopBit());
    values.clear();
    for (int i = 0; i < len; i++) {
        if (buffer.size() <= values.size()) {
            buffer.add(ObjectUtils.newInstance(MovingAverageCompact.class));
        }
        MovingAverageCompact next = buffer.get(i);
        next.readMarshallable(bytes);
        values.add(next);
    }
}
 

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

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

        String format = newMeta.roll().format();
        if (!format.equals(rollCycle().format())) {
            // roll cycle changed
            overrideRollCycleForFileName(format);
        }

        // if it was overridden - reset
        rollTime = newMeta.roll().rollTime();
        rollTimeZone = newMeta.roll().rollTimeZone();
        epoch = newMeta.roll().epoch();
    } 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.
        Jvm.warn().on(getClass(), "Failback to readonly tablestore", ex);
        metaStore = new ReadonlyTableStore<>(metadata);
    }
}
 

@Override
public void readMarshallable(@NotNull WireIn wire) throws IORuntimeException {
    ValueIn valueIn = wire.getValueIn();
    for (int i = 0; i < NUMBER_OF_LONGS; i++)
        assertEquals(i, valueIn.int64());
    //Jvm.pause(PAUSE_READ_MS);
}
 

@Test
public void testWriteBytes() {
    File dir = DirectoryUtils.tempDir("WriteBytesTest");
    try (ChronicleQueue queue = binary(dir)
            .testBlockSize()
            .build()) {

        ExcerptAppender appender = queue.acquireAppender();
        ExcerptTailer tailer = queue.createTailer();

        outgoingMsgBytes[0] = 'A';
        outgoingBytes.write(outgoingMsgBytes);
        postOneMessage(appender);
        fetchOneMessage(tailer, incomingMsgBytes);
        System.out.println(new String(incomingMsgBytes));

        outgoingBytes.clear();

        outgoingMsgBytes[0] = 'A';
        outgoingMsgBytes[1] = 'B';
        outgoingBytes.write(outgoingMsgBytes);

        postOneMessage(appender);
        fetchOneMessage(tailer, incomingMsgBytes);
        System.out.println(new String(incomingMsgBytes));

    } finally {
        try {
            IOTools.deleteDirWithFiles(dir, 2);
        } catch (IORuntimeException e) {
            // ignored
        }
    }
}
 

@Override
public void readMarshallable(@NotNull BytesIn bytes) throws IORuntimeException {
    _name = bytes.readUtf8();
    _value1 = bytes.readLong();
    _value2 = bytes.readLong();
    _value3 = bytes.readLong();
}
 

@Test
public void testWriteText() {
    File dir = DirectoryUtils.tempDir("testWriteText");
    try (ChronicleQueue queue = binary(dir)
            .testBlockSize()
            .build()) {

        ExcerptAppender appender = queue.acquireAppender();
        ExcerptTailer tailer = queue.createTailer();
        ExcerptTailer tailer2 = queue.createTailer();

        int runs = 1000;
        for (int i = 0; i < runs; i++)
            appender.writeText("" + i);
        for (int i = 0; i < runs; i++)
            assertEquals("" + i, tailer.readText());
        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < runs; i++) {
            assertTrue(tailer2.readText(sb));
            assertEquals("" + i, sb.toString());
        }
    } finally {
        try {
            IOTools.deleteDirWithFiles(dir, 2);
        } catch (IORuntimeException e) {
            // ignored
        }
    }
}
 

@Override
public void readMarshallable(@NotNull WireIn wire) throws IORuntimeException {
    // noop - no fields to unmarshall
}
 

public boolean action() {
    final int read;
    boolean busy = false;
    bufferHandler.handleDecryptedData(inboundApplicationData, outboundApplicationData);
    try {
        if (outboundApplicationData.position() != 0) {

            outboundApplicationData.flip();

            if (engine.wrap(precomputedWrapArray, outboundEncodedData).
                    getStatus() == SSLEngineResult.Status.CLOSED) {
                LOGGER.warn("Socket closed");
                return false;
            }
            busy = outboundApplicationData.hasRemaining();
            outboundApplicationData.compact();
        }
        if (outboundEncodedData.position() != 0) {
            outboundEncodedData.flip();
            bufferHandler.writeData(outboundEncodedData);
            busy |= outboundEncodedData.hasRemaining();
            outboundEncodedData.compact();
        }

        read = bufferHandler.readData(inboundEncodedData);
        if (read == -1) {
            throw new IORuntimeException("Socket closed");
        }
        busy |= read != 0;

        if (inboundEncodedData.position() != 0) {
            inboundEncodedData.flip();
            engine.unwrap(inboundEncodedData, precomputedUnwrapArray);
            busy |= inboundEncodedData.hasRemaining();
            inboundEncodedData.compact();
        }

        if (inboundApplicationData.position() != 0) {
            inboundApplicationData.flip();
            bufferHandler.handleDecryptedData(inboundApplicationData, outboundApplicationData);
            busy |= inboundApplicationData.hasRemaining();
            inboundApplicationData.compact();
        }
    } catch (IOException e) {
        throw new UncheckedIOException(e);
    }

    return busy;
}
 

public void checkNotShuttingDown() {
    if (isShuttingdown())
        throw new IORuntimeException("Called after shutdown", shutdownHere);
}
 

@Override
public void readMarshallable(@NotNull final WireIn wire) throws IORuntimeException {
    throwExceptionIfClosed();

    tid = CoreFields.tid(wire);
}
 

@Override
public void readMarshallable(@NotNull WireIn wire) throws IORuntimeException {
    wire.read("networkStatsListenerFactory").object(networkStatsListenerFactory, Function.class);
    defaults();
    super.readMarshallable(wire);
}
 

@Override
public void readMarshallable(@NotNull final WireIn wire) throws IORuntimeException {
}
 

@NotNull
@Override
public CharSequence read(Bytes in, @Nullable CharSequence using) {
    long csLengthAsLong = in.readStopBit();
    if (csLengthAsLong > Integer.MAX_VALUE) {
        throw new IORuntimeException("cs len shouldn't be more than " + Integer.MAX_VALUE +
                ", " + csLengthAsLong + " read");
    }
    int csLength = (int) csLengthAsLong;
    StringBuilder sb;
    if (using instanceof StringBuilder) {
        sb = (StringBuilder) using;
        sb.setLength(0);
        sb.ensureCapacity(csLength);
    } else {
        sb = new StringBuilder(csLength);
    }

    int remainingBytes = in.readInt();
    charsetDecoder.reset();
    inputBuffer.clear();
    outputBuffer.clear();
    boolean endOfInput = false;
    // this loop inspired by the CharsetDecoder.decode(ByteBuffer) implementation
    while (true) {
        if (!endOfInput) {
            int inputChunkSize = Math.min(inputBuffer.remaining(), remainingBytes);
            inputBuffer.limit(inputBuffer.position() + inputChunkSize);
            in.read(inputBuffer);
            inputBuffer.flip();
            remainingBytes -= inputChunkSize;
            endOfInput = remainingBytes == 0;
        }

        CoderResult cr = inputBuffer.hasRemaining() ?
                charsetDecoder.decode(inputBuffer, outputBuffer, endOfInput) :
                CoderResult.UNDERFLOW;

        if (cr.isUnderflow() && endOfInput)
            cr = charsetDecoder.flush(outputBuffer);

        if (cr.isUnderflow()) {
            if (endOfInput) {
                break;
            } else {
                inputBuffer.compact();
                continue;
            }
        }

        if (cr.isOverflow()) {
            outputBuffer.flip();
            sb.append(outputBuffer);
            outputBuffer.clear();
            continue;
        }

        try {
            cr.throwException();
        } catch (CharacterCodingException e) {
            throw new IORuntimeException(e);
        }
    }
    outputBuffer.flip();
    sb.append(outputBuffer);

    return sb;
}
 

@Override
public void readMarshallable(@NotNull WireIn wireIn) throws IORuntimeException {
    charset = (Charset) wireIn.read(() -> "charset").object();
    inputBufferSize = wireIn.read(() -> "inputBufferSize").int32();
    initTransients();
}
 

@Override
public void write(Bytes out, @NotNull CharSequence cs) {
    // Write the actual cs length for accurate StringBuilder.ensureCapacity() while reading
    out.writeStopBit(cs.length());
    long encodedSizePos = out.writePosition();
    out.writeSkip(4);
    charsetEncoder.reset();
    inputBuffer.clear();
    outputBuffer.clear();
    int csPos = 0;
    boolean endOfInput = false;
    // this loop inspired by the CharsetEncoder.encode(CharBuffer) implementation
    while (true) {
        if (!endOfInput) {
            int nextCsPos = Math.min(csPos + inputBuffer.remaining(), cs.length());
            append(inputBuffer, cs, csPos, nextCsPos);
            inputBuffer.flip();
            endOfInput = nextCsPos == cs.length();
            csPos = nextCsPos;
        }

        CoderResult cr = inputBuffer.hasRemaining() ?
                charsetEncoder.encode(inputBuffer, outputBuffer, endOfInput) :
                CoderResult.UNDERFLOW;

        if (cr.isUnderflow() && endOfInput)
            cr = charsetEncoder.flush(outputBuffer);

        if (cr.isUnderflow()) {
            if (endOfInput) {
                break;
            } else {
                inputBuffer.compact();
                continue;
            }
        }

        if (cr.isOverflow()) {
            outputBuffer.flip();
            writeOutputBuffer(out);
            outputBuffer.clear();
            continue;
        }

        try {
            cr.throwException();
        } catch (CharacterCodingException e) {
            throw new IORuntimeException(e);
        }
    }
    outputBuffer.flip();
    writeOutputBuffer(out);

    out.writeInt(encodedSizePos, (int) (out.writePosition() - encodedSizePos - 4));
}
 

@Override
public void readMarshallable(@NotNull WireIn wire) throws IORuntimeException {
    super.readMarshallable(wire);
    assert rollCycle != null;
}
 

@Test
    @Ignore("TODO FIX")
    public void testWithAsQueueService() {
        // acts as three processes in one test
        // process A writes to the HelloWorld interface.
        // process B read fromt he HelloWorld interface and writes to the
        String input = OS.TARGET + "/input-" + System.nanoTime();
        String output = OS.TARGET + "/output-" + System.nanoTime();

        HelloReplier replier = createMock(HelloReplier.class);
        replier.reply("Hello April");
        replier.reply("Hello June");
        replay(replier);

        ServiceWrapperBuilder<HelloReplier> builder = ServiceWrapperBuilder
                .serviceBuilder(input, output, HelloReplier.class, HelloWorldImpl::new)
                .inputSourceId(1).outputSourceId(2);

        try (CloseableHelloWorld helloWorld = builder.inputWriter(CloseableHelloWorld.class);
             MethodReader replyReader = builder.outputReader(replier);
             ServiceWrapper helloWorldService = builder.get()) {

            helloWorld.hello("April");
            helloWorld.hello("June");

//            System.out.println(helloWorldService.inputQueues()[0].dump());
            for (int i = 0; i < 2; i++) {
                while (!replyReader.readOne()) {
                    Thread.yield();
                }
            }
//            System.out.println(helloWorldService.outputQueue().dump());
            verify(replier);
        } finally {
            builder.closeQueues();
            try {
                IOTools.deleteDirWithFiles(new File(input), 2);
                IOTools.deleteDirWithFiles(new File(output), 2);
            } catch (IORuntimeException e) {
                e.printStackTrace();
            }
        }
    }
 

@Override
public void readMarshallable(BytesIn bytes) throws IORuntimeException {
    ts = bytes.readLong();
    bytes.read(filler);
}
 

/**
 * Winds this Tailer to after the last entry which wrote an entry to the queue.
 *
 * @param queue which was written to.
 * @return this ExcerptTailer
 *
 * @throws IORuntimeException if the provided {@code queue} couldn't be wound to the last index.
 * @throws NullPointerException if the provided {@code queue} is {@code null}
 */
@NotNull
ExcerptTailer afterLastWritten(ChronicleQueue queue) throws IORuntimeException;