37648054681

Committed 07 Oct 2026 03:55PM UTC coverage: 89.174% (-2.1%) from 91.29%

Build # 37648054681

Build Type

push

github

Committed by

scottf

Commit Message

NatsConnectionV3 (default) with OutboundBuffer, classic fixes, build-variants CI

- NatsConnectionV3 extends NatsConnection, selected by ConnectionImplementation (V3 default, Classic available via option, property or -PconnectionImplementation)
- OutboundBuffer: control lane + bounded data blocks; SUB control line limit check; kept-block limits from the first server's max payload
- Classic fixes: protocol buffer sizing, tryingToConnect CAS, lock-free pending getters, sendPing pong future cleanup, per-run reader/writer flag
- build-variants.yml replaces build-windows.yml
- Plans: connection V3, active/passive V3, testing.java overview, V2 bugs from review

Coverage Stats

4596 of 5421 branches covered (84.78%)

Branch coverage included in aggregate %.

1027 of 1185 new or added lines in 12 files covered. (86.67%)

161 existing lines in 4 files now uncovered.

11532 of 12665 relevant lines covered (91.05%)

4.03 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

92.68

/core/src/main/java/io/synadia/client/impl/NatsConnectionReader.java

package io.synadia.client.impl;

import io.synadia.client.Message;
import io.synadia.client.ReadListener;
import io.synadia.client.utils.IncomingHeadersProcessor;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;

import static io.synadia.client.utils.NatsConstants.*;

/**
 * The read loop for a connection. It runs on its own thread, pulls bytes off the socket and
 * advances a small state machine that gathers a protocol op, then its protocol line, then any
 * headers and payload, handing each completed message to the connection for delivery.
 */
public class NatsConnectionReader implements Runnable {

    enum Mode {
        GATHER_OP,
        GATHER_PROTO,
        GATHER_MSG_HMSG_PROTO,
        PARSE_PROTO,
        GATHER_HEADERS,
        GATHER_DATA
    }

    // volatile (was final) so a live reader can be repointed at a different connection - the reader
    // thread reads this on every delivered message and must see the new target. Repointing is done
    // through setConnection(...) rather than direct field access, so the field stays private; see
    // NatsConnection.setReaderConnection(...) for the entry point
    private volatile NatsConnection connection;

    private ByteBuffer protocolBuffer; // use a byte buffer to assist character decoding

    private boolean gotCR;
    
    private String op;
    private final char[] opArray;
    private int opPos;

    private final char[] msgLineChars;
    private int msgLinePosition;

    private Mode mode;

    private IncomingMessageFactory incoming;
    private byte[] msgHeaders;
    private byte[] msgData;
    private int msgHeadersPosition;
    private int msgDataPosition;

    private final byte[] buffer;
    private int bufferPosition;

    private Future<Boolean> stopped;
    private Future<DataPort> dataPortFuture;
    private DataPort dataPort;
    // A new flag per start(). The thread started by start() checks and clears only the flag it was
    // started with, so a late thread from a previous socket cannot stop or disturb the current one.
    private volatile AtomicBoolean running;

    private final boolean utf8Mode;
    // The listener the reader thread invokes: either a no-op (no ReadListener configured) or a wrapper
    // that dispatches on the CURRENT connection (read from the volatile connection field at call time).
    // volatile because the reader thread reads it and the setConnection caller thread may replace it.
    private volatile ReadListener readListener;
    // The raw ReadListener (from Options) the current wrapper was built from - null when the no-op is in
    // place. Used by refreshReadListener to skip a needless rebuild when a repoint doesn't change it.
    // Only touched by the constructing / repointing thread, never the reader thread.
    private ReadListener currentUserRl;

    NatsConnectionReader(NatsConnection connection) {
        this.connection = connection;

        this.running = new AtomicBoolean(false);
        this.stopped = new CompletableFuture<>();
        ((CompletableFuture<Boolean>)this.stopped).complete(Boolean.TRUE); // we are stopped on creation

        this.protocolBuffer = ByteBuffer.allocate(this.connection.getOptions().getMaxControlLine());
        this.msgLineChars = new char[this.connection.getOptions().getMaxControlLine()];
        this.opArray = new char[MAX_PROTOCOL_RECEIVE_OP_LENGTH];
        this.buffer = new byte[connection.getOptions().getBufferSize()];
        this.bufferPosition = 0;

        this.utf8Mode = connection.getOptions().supportUTF8Subjects();

        refreshReadListener(connection);
    }

    void setReadListener(ReadListener rl) {
        // keep currentUserRl in step with what is actually installed, so a later repoint compares against
        // this listener rather than the previous connection's Options (see refreshReadListener)
        currentUserRl = rl;
        if (rl == null) {
            readListener = new ReadListener() {};
        }
        else {
            readListener = new ReadListener() {
                @Override
                public void protocol(String op, String text) {
                    connection.makeCallback(() -> rl.protocol(op, text));
                }

                @Override
                public void message(String op, Message message) {
                    connection.makeCallback(() -> rl.message(op, message));
                }
            };
        }
    }

    // (Re)derive the readListener from the connection's Options, but replace it ONLY when the configured
    // ReadListener actually changed - so a repoint that doesn't change it costs nothing. The wrapper reads
    // the current connection from the volatile field at dispatch time (not a captured reference), so when
    // the source listener is unchanged the existing wrapper (or no-op) is already correct for the new
    // connection and is kept as-is.
    //   - first call (construction): readListener is still null, so we always build.
    //   - repoint, old null + new null: same (no-op already installed) - keep it.
    //   - repoint, old == new (same instance): the wrapper is fine - keep it.
    //   - otherwise: build the matching wrapper (or no-op) and remember the new source.
    private void refreshReadListener(NatsConnection connection) {
        final ReadListener newSuppliedUserRl = connection.getOptions().getReadListener();
        if (readListener != null && newSuppliedUserRl == currentUserRl) {
            return;
        }
        currentUserRl = newSuppliedUserRl;
        if (newSuppliedUserRl == null) {
            readListener = new ReadListener() {};
        }
        else {
            readListener = new ReadListener() {
                @Override
                public void protocol(String op, String text) {
                    NatsConnectionReader.this.connection.makeCallback(() -> newSuppliedUserRl.protocol(op, text));
                }

                @Override
                public void message(String op, Message message) {
                    NatsConnectionReader.this.connection.makeCallback(() -> newSuppliedUserRl.message(op, message));
                }
            };
        }
    }

    // test access only - the listener the reader thread would invoke (no-op or wrapper)
    ReadListener readListener() {
        return readListener;
    }

    // Repoint this reader at a different connection. The connection field is volatile, so the reader
    // thread sees the new target the next time it reads it, and the ReadListener is re-derived from the
    // new connection's Options only when it actually changed (see refreshReadListener).
    //
    // What follows the repoint: everything read from the connection field on use - message delivery,
    // statistics, protocol handling (OK/ERR/PONG/INFO), and the re-derived ReadListener. What does NOT
    // follow: the wire-parse parameters fixed at construction - the read buffer size, the max-control-line
    // buffers (msgLineChars / protocolBuffer), and utf8 subject mode. Those hold in-flight parse state and
    // are used in tight loops by the reader thread; swapping them from another thread mid-parse would
    // corrupt the message being decoded, so they are intentionally left as constructed. A repoint is
    // therefore only fully safe across connections whose parse-affecting Options (buffer size, max control
    // line, utf8 subjects) match; otherwise the reader keeps parsing per its original Options.
    //
    // INVARIANT: volatile gives visibility, NOT atomicity across a multi-step message parse. This method
    // does NOT require the reader to be stopped first - a caller may repoint a running reader - so if a
    // repoint lands mid-message it can straddle both connections (e.g. read stats registered on the
    // previous connection, delivery on the new one). A caller needing a clean handoff must ensure the
    // reader is between messages, or accept the split. When the two connections share the same underlying
    // socket, delivering to the new connection is correct and only a read-byte stat is split.
    protected void setConnection(NatsConnection connection) {
        this.connection = connection;
        refreshReadListener(connection);
    }

    // Should only be called if the current thread has exited.
    // Use the Future from stop() to determine if it is ok to call this.
    // This method resets that future so mistiming can result in badness.
    void start(Future<DataPort> dataPortFuture) {
        this.dataPortFuture = dataPortFuture;
        AtomicBoolean runFlag = new AtomicBoolean(true);
        this.running = runFlag;
        this.stopped = connection.getReaderExecutor().submit(() -> run(runFlag), Boolean.TRUE);
    }

    Future<Boolean> stop() {
        return stop(true);
    }

    // May be called several times on an error.
    // Returns a future that is completed when the thread completes, not when this
    // method does.
    Future<Boolean> stop(boolean shutdownDataPort) {
        AtomicBoolean runFlag = running;
        if (runFlag.get()) {
            runFlag.set(false);
            if (shutdownDataPort && dataPort != null) {
                try {
                    dataPort.shutdownInput();
                }
                catch (IOException e) {
                    // we don't care, we are shutting down anyway
                }
            }
        }
        return stopped;
    }

    boolean isRunning() {
        return running.get();
    }

    @Override
    public void run() {
        run(running);
    }

    private void run(AtomicBoolean runFlag) {
        try {
            dataPort = this.dataPortFuture.get(); // Will wait for the future to complete
            this.mode = Mode.GATHER_OP;
            this.gotCR = false;
            this.opPos = 0;

            while (runFlag.get() && !Thread.interrupted()) {
                this.bufferPosition = 0;
                int bytesRead = dataPort.read(this.buffer, 0, this.buffer.length);

                if (bytesRead > 0) {
                    connection.getStatisticsCollector().registerRead(bytesRead);

                    while (this.bufferPosition < bytesRead) {
                        if (this.mode == Mode.GATHER_OP) {
                            this.gatherOp(bytesRead);
                        }
                        else if (this.mode == Mode.GATHER_MSG_HMSG_PROTO) {
                            if (this.utf8Mode) {
                                this.gatherProtocol(bytesRead);
                            } else {
                                this.gatherMessageProtocol(bytesRead);
                            }
                        }
                        else if (this.mode == Mode.GATHER_PROTO) {
                            this.gatherProtocol(bytesRead);
                        }
                        else if (this.mode == Mode.GATHER_HEADERS) {
                            this.gatherHeaders(bytesRead);
                        }
                        else {  // Mode.GATHER_DATA
                            this.gatherMessageData(bytesRead);
                        }

                        if (this.mode == Mode.PARSE_PROTO) { // Could be the end of the read
                            this.parseProtocolMessage();
                            this.protocolBuffer.clear();
                        }
                    }
                } else if (bytesRead < 0) {
                    throw new IOException("Read channel closed.");
                } else {
                    this.connection.getStatisticsCollector().registerRead(bytesRead); // track the 0
                }
            }
        } catch (IOException io) {
            // if already not running, an IOE is not unreasonable in a transition state
            if (runFlag.get()) {
                this.connection.handleCommunicationIssue(this, io);
            }
        } catch (CancellationException | ExecutionException ex) {
            // Exit
        } catch (InterruptedException ex) {
            // Exit
            Thread.currentThread().interrupt();
        } finally {
            runFlag.set(false);
            // Clear the buffers, since they are only used inside this try/catch
            // We will reuse later
            this.protocolBuffer.clear();
        }
    }

    // Gather the op, either up to the first space or the first carriage return.
    void gatherOp(int maxPos) throws IOException {
        try {
            while(this.bufferPosition < maxPos) {
                byte b = this.buffer[this.bufferPosition];
                this.bufferPosition++;

                if (gotCR) {
                    if (b == LF) { // Got CRLF, jump to parsing
                        this.op = opFor(opArray, opPos);
                        this.gotCR = false;
                        this.opPos = 0;
                        this.mode = Mode.PARSE_PROTO;
                        break;
                    } else {
                        throw new IllegalStateException("Bad socket data, no LF after CR");
                    }
                } else if (b == SP || b == TAB) { // Got a space, get the rest of the protocol line
                    this.op = opFor(opArray, opPos);
                    this.opPos = 0;
                    if (this.op.equals(OP_MSG) || this.op.equals(OP_HMSG)) {
                        this.msgLinePosition = 0;
                        this.mode = Mode.GATHER_MSG_HMSG_PROTO;
                    } else {
                        this.mode = Mode.GATHER_PROTO;
                    }
                    break;
                } else if (b == CR) {
                    this.gotCR = true;
                } else {
                    this.opArray[opPos] = (char) b;
                    this.opPos++;
                }
            }
        }
        catch (ArrayIndexOutOfBoundsException | IllegalStateException | NumberFormatException | NullPointerException ex) {
            throw new IOException("Gather Operation", ex);
        }
    }

    // Stores the message protocol line in a char buffer that will be read for subject, reply
    void gatherMessageProtocol(int maxPos) throws IOException {
        try {
            while(this.bufferPosition < maxPos) {
                byte b = this.buffer[this.bufferPosition];
                this.bufferPosition++;

                if (gotCR) {
                    if (b == LF) {
                        this.mode = Mode.PARSE_PROTO;
                        this.gotCR = false;
                        break;
                    } else {
                        throw new IllegalStateException("Bad socket data, no LF after CR");
                    }
                } else if (b == CR) {
                    this.gotCR = true;
                } else {
                    if (this.msgLinePosition >= this.msgLineChars.length) {
                        throw new IllegalStateException("Protocol line is too long");
                    }
                    this.msgLineChars[this.msgLinePosition] = (char) b; // Assumes ascii, as per protocol doc
                    this.msgLinePosition++;
                }
            }
        }
        catch (IllegalStateException | NumberFormatException | NullPointerException ex) {
            throw new IOException("Gather Message", ex);
        }
    }

    // Gather bytes for a protocol line
    void gatherProtocol(int maxPos) throws IOException {
        // protocol buffer has max capacity, shouldn't need resizing
        try {
            while(this.bufferPosition < maxPos) {
                byte b = this.buffer[this.bufferPosition];
                this.bufferPosition++;

                if (gotCR) {
                    if (b == LF) {
                        this.protocolBuffer.flip();
                        this.mode = Mode.PARSE_PROTO;
                        this.gotCR = false;
                        break;
                    } else {
                        throw new IllegalStateException("Bad socket data, no LF after CR");
                    }
                } else if (b == CR) {
                    this.gotCR = true;
                } else {
                    if (!protocolBuffer.hasRemaining()) {
                        this.protocolBuffer = this.connection.enlargeBuffer(this.protocolBuffer); // just double it
                    }
                    this.protocolBuffer.put(b);
                }
            }
        }
        catch (IllegalStateException | NumberFormatException | NullPointerException ex) {
            throw new IOException("Gather Protocol", ex);
        }
    }

    void gatherHeaders(int maxPos) throws IOException {
        try {
            while(this.bufferPosition < maxPos) {
                int possible = maxPos - this.bufferPosition;
                int want = msgHeaders.length - msgHeadersPosition;

                // Grab all we can, until we get the necessary number of bytes
                if (want > 0 && want <= possible) {
                    System.arraycopy(this.buffer, this.bufferPosition, this.msgHeaders, this.msgHeadersPosition, want);
                    msgHeadersPosition += want;
                    this.bufferPosition += want;
                    continue;
                } else if (want > 0) {
                    System.arraycopy(this.buffer, this.bufferPosition, this.msgHeaders, this.msgHeadersPosition, possible);
                    msgHeadersPosition += possible;
                    this.bufferPosition += possible;
                    continue;
                }

                if (msgHeadersPosition == msgHeaders.length) {
                    incoming.setHeaders(new IncomingHeadersProcessor(msgHeaders));
                    msgHeaders = null;
                    msgHeadersPosition = -1;
                    this.mode = Mode.GATHER_DATA;
                    break;
                } else {
                    throw new IllegalStateException("Bad socket data, headers do not match expected length");
                }
            }
        }
        catch (IllegalStateException | NullPointerException ex) {
            throw new IOException("Gather Header", ex);
        }
    }

    // Gather bytes for a message body into a byte array that is then
    // given to the message object
    void gatherMessageData(int maxPos) throws IOException {
        try {
            while(this.bufferPosition < maxPos) {
                int possible = maxPos - this.bufferPosition;
                int want = msgData.length - msgDataPosition;

                // Grab all we can, until we get to the CR/LF
                if (want > 0 && want <= possible) {
                    System.arraycopy(this.buffer, this.bufferPosition, this.msgData, this.msgDataPosition, want);
                    msgDataPosition += want;
                    this.bufferPosition += want;
                    continue;
                } else if (want > 0) {
                    System.arraycopy(this.buffer, this.bufferPosition, this.msgData, this.msgDataPosition, possible);
                    msgDataPosition += possible;
                    this.bufferPosition += possible;
                    continue;
                }

                byte b = this.buffer[this.bufferPosition];
                this.bufferPosition++;

                if (gotCR) {
                    if (b == LF) {
                        incoming.setData(msgData);
                        NatsMessage m = incoming.getMessage();
                        this.connection.deliverMessage(m);
                        readListener.message(op, m);
                        msgData = null;
                        msgDataPosition = 0;
                        incoming = null;
                        gotCR = false;
                        this.op = UNKNOWN_OP;
                        this.mode = Mode.GATHER_OP;
                        break;
                    } else {
                        throw new IllegalStateException("Bad socket data, no LF after CR");
                    }
                } else if (b == CR) {
                    gotCR = true;
                } else {
                    throw new IllegalStateException("Bad socket data, no CRLF after data");
                }
            }
        }
        catch (IllegalStateException | NullPointerException ex) {
            throw new IOException("Gather Message Data", ex);
        }
    }

    /**
     * Take the next whitespace delimited element from the protocol line currently being parsed,
     * advancing the line position past it and past the delimiter.
     * @param max the position in the line to stop at, normally the length of the gathered line
     * @return the element, or null if the line position is already at or past max
     */
    public String grabNextMessageLineElement(int max) {
        if (this.msgLinePosition >= max) {
            return null;
        }

        int start = this.msgLinePosition;

        while (this.msgLinePosition < max) {
            char c = this.msgLineChars[this.msgLinePosition];
            this.msgLinePosition++;

            if (c == SP || c == TAB) {
                return new String(this.msgLineChars, start, this.msgLinePosition - start -1); //don't grab the space, avoid an intermediate char sequence
            }
        }

        return new String(this.msgLineChars, start, this.msgLinePosition-start);
    }

    static String opFor(char[] chars, int length) {
        if (length == 3) {
            if ((chars[0] == 'M' || chars[0] == 'm') &&
                        (chars[1] == 'S' || chars[1] == 's') && 
                        (chars[2] == 'G' || chars[2] == 'g')) {
                return OP_MSG;
            } else if (chars[0] == '+' && 
                (chars[1] == 'O' || chars[1] == 'o') && 
                (chars[2] == 'K' || chars[2] == 'k')) {
                return OP_OK;
            } else {
                return UNKNOWN_OP;
            }
        } else if (length == 4) { // do them in a unique order for uniqueness when possible to branch asap
            if ((chars[1] == 'I' || chars[1] == 'i') && 
                    (chars[0] == 'P' || chars[0] == 'p') && 
                    (chars[2] == 'N' || chars[2] == 'n') &&
                    (chars[3] == 'G' || chars[3] == 'g')) {
                return OP_PING;
            } else if ((chars[1] == 'O' || chars[1] == 'o') && 
                        (chars[0] == 'P' || chars[0] == 'p') && 
                        (chars[2] == 'N' || chars[2] == 'n') &&
                        (chars[3] == 'G' || chars[3] == 'g')) {
                return OP_PONG;
            } else if (chars[0] == '-' && 
                        (chars[1] == 'E' || chars[1] == 'e') &&
                        (chars[2] == 'R' || chars[2] == 'r') && 
                        (chars[3] == 'R' || chars[3] == 'r')) {
                return OP_ERR;
            } else if ((chars[0] == 'I' || chars[0] == 'i') &&
                    (chars[1] == 'N' || chars[1] == 'n') &&
                    (chars[2] == 'F' || chars[2] == 'f') &&
                    (chars[3] == 'O' || chars[3] == 'o')) {
                return OP_INFO;
            } else if ((chars[0] == 'H' || chars[0] == 'h') &&
                    (chars[1] == 'M' || chars[1] == 'm') &&
                    (chars[2] == 'S' || chars[2] == 's') &&
                    (chars[3] == 'G' || chars[3] == 'g')) {
                return OP_HMSG;
            }  else {
                return UNKNOWN_OP;
            }
        } else {
            return UNKNOWN_OP;
        }
    }

    private static final int[] TENS = new int[] { 1, 10, 100, 1_000, 10_000, 100_000, 1_000_000, 10_000_000, 100_000_000, 1_000_000_000};

    /**
     * Parse an unsigned decimal length from a protocol line. Faster than Integer.parseInt because it
     * accepts only digits, no sign and no whitespace, and it uses a lookup table of powers of ten.
     * @param s the digits to parse
     * @return the value
     * @throws NumberFormatException if a character is not a digit, or there are more digits than an int can hold
     */
    public static int parseLength(String s) throws NumberFormatException {
        int length = s.length();
        int retVal = 0;

        if (length > TENS.length) {
            throw new NumberFormatException("Long in message length \"" + s + "\" "+length+" > "+TENS.length);
        }
        
        for (int i=length-1;i>=0;i--) {
            char c = s.charAt(i);
            int d = (c - '0');

            if (d>9) {
                throw new NumberFormatException("Invalid char in message length '" + c + "'");
            }

            retVal += d * TENS[length - i - 1];
        }

        return retVal;
    }

    void parseProtocolMessage() throws IOException {
        try {
            switch (this.op) {
                case OP_MSG:
                    int protocolLength = this.msgLinePosition; //This is just after the last character
                    int protocolLineLength = protocolLength + 4; // 4 for the "MSG "

                    if (this.utf8Mode) {
                        protocolLineLength = protocolBuffer.remaining() + 4;

                        CharBuffer buff = StandardCharsets.UTF_8.decode(protocolBuffer);
                        protocolLength = buff.remaining();
                        buff.get(this.msgLineChars, 0, protocolLength);
                    }

                    this.msgLinePosition = 0;
                    String subject = grabNextMessageLineElement(protocolLength);
                    String sid = grabNextMessageLineElement(protocolLength);
                    String replyTo = grabNextMessageLineElement(protocolLength);
                    String lengthChars;

                    if (this.msgLinePosition < protocolLength) {
                        lengthChars = grabNextMessageLineElement(protocolLength);
                    } else {
                        lengthChars = replyTo;
                        replyTo = null;
                    }

                    if (subject == null || subject.isEmpty() || sid == null || sid.isEmpty() || lengthChars == null) {
                        throw new IllegalStateException("Bad MSG control line, missing required fields");
                    }

                    int incomingLength = parseLength(lengthChars);

                    this.incoming = new IncomingMessageFactory(sid, subject, replyTo, protocolLineLength);
                    this.mode = Mode.GATHER_DATA;
                    this.msgData = new byte[incomingLength];
                    this.msgDataPosition = 0;
                    this.msgLinePosition = 0;
                    break;
                case OP_HMSG:
                    int hProtocolLength = this.msgLinePosition; //This is just after the last character
                    int hProtocolLineLength = hProtocolLength + 5; // 5 for the "HMSG "

                    if (this.utf8Mode) {
                        hProtocolLineLength = protocolBuffer.remaining() + 5;

                        CharBuffer buff = StandardCharsets.UTF_8.decode(protocolBuffer);
                        hProtocolLength = buff.remaining();
                        buff.get(this.msgLineChars, 0, hProtocolLength);
                    }

                    this.msgLinePosition = 0;
                    String hSubject = grabNextMessageLineElement(hProtocolLength);
                    String hSid = grabNextMessageLineElement(hProtocolLength);
                    String replyToOrHdrLen = grabNextMessageLineElement(hProtocolLength);
                    String hdrLenOrTotLen = grabNextMessageLineElement(hProtocolLength);

                    String hReplyTo = null;
                    int hdrLen;
                    int totLen;

                    // if there is more it must be replyTo hdrLen totLen instead of just hdrLen totLen
                    if (this.msgLinePosition < hProtocolLength) {
                        hReplyTo = replyToOrHdrLen;
                        hdrLen = parseLength(hdrLenOrTotLen);
                        totLen = parseLength(grabNextMessageLineElement(hProtocolLength));
                    }
                    else {
                        hdrLen = parseLength(replyToOrHdrLen);
                        totLen = parseLength(hdrLenOrTotLen);
                    }

                    if(hSubject==null || hSubject.isEmpty() || hSid==null || hSid.isEmpty()) {
                        throw new IllegalStateException("Bad HMSG control line, missing required fields");
                    }

                    this.incoming = new IncomingMessageFactory(hSid, hSubject, hReplyTo, hProtocolLineLength);
                    this.msgHeaders = new byte[hdrLen];
                    this.msgData = new byte[totLen - hdrLen];
                    this.mode = Mode.GATHER_HEADERS;
                    this.msgHeadersPosition = 0;
                    this.msgDataPosition = 0;
                    this.msgLinePosition = 0;
                    break;
                case OP_OK:
                    this.connection.processOK();
                    readListener.protocol(op, null);
                    this.op = UNKNOWN_OP;
                    this.mode = Mode.GATHER_OP;
                    break;
                case OP_ERR:
                    String errorText = StandardCharsets.UTF_8.decode(protocolBuffer).toString().replace("'", "");
                    this.connection.processError(errorText);
                    readListener.protocol(op, errorText);
                    this.op = UNKNOWN_OP;
                    this.mode = Mode.GATHER_OP;
                    break;
                case OP_PING:
                    readListener.protocol(op, null);
                    this.connection.sendPong();
                    this.op = UNKNOWN_OP;
                    this.mode = Mode.GATHER_OP;
                    break;
                case OP_PONG:
                    readListener.protocol(op, null);
                    this.connection.handlePong();
                    this.op = UNKNOWN_OP;
                    this.mode = Mode.GATHER_OP;
                    break;
                case OP_INFO:
                    String info = StandardCharsets.UTF_8.decode(protocolBuffer).toString();
                    this.connection.handleInfo(info);
                    readListener.protocol(op, null);
                    this.op = UNKNOWN_OP;
                    this.mode = Mode.GATHER_OP;
                    break;
                default:
                    throw new IllegalStateException("Unknown protocol operation "+op);
            }
        }
        catch (IllegalStateException | NumberFormatException | NullPointerException ex) {
            throw new IOException("Parse Protocol OP_" + op, ex);
        }
    }

    //For testing
    void fakeReadForTest(byte[] bytes) {
        for (int x = 0; x < this.buffer.length; x++) {
            this.buffer[x] = 0;
        }
        System.arraycopy(bytes, 0, this.buffer, 0, bytes.length);
        this.bufferPosition = 0;
        this.op = UNKNOWN_OP;
        this.mode = Mode.GATHER_OP;
    }

    String currentOp() {
        return this.op;
    }
}

synadia-io / nats.java.v3 / 37648054681

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous