RabbitMQ-客户端源码之AMQCommand

/** Current state, used to decide how to handle each incoming frame. */
private enum CAState {
    EXPECTING_METHOD, EXPECTING_CONTENT_HEADER, EXPECTING_CONTENT_BODY, COMPLETE
}
private CAState state;

/** The method for this command */
private Method method;

/** The content header for this command */
private AMQContentHeader contentHeader;

/** The fragments of this command's content body - a list of byte[] */
private final List<byte[]> bodyN;
/** sum of the lengths of all fragments */
private int bodyLength;

/** No bytes of content body not yet accumulated */
private long remainingBodyBytes;

public synchronized boolean handleFrame(Frame f) throws IOException
{
    switch (this.state) {
      case EXPECTING_METHOD:          consumeMethodFrame(f); break;
      case EXPECTING_CONTENT_HEADER:  consumeHeaderFrame(f); break;
      case EXPECTING_CONTENT_BODY:    consumeBodyFrame(f);   break;

      default:
          throw new AssertionError("Bad Command State " + this.state);
    }
    return isComplete();
}

private void consumeMethodFrame(Frame f) throws IOException {
    if (f.type == AMQP.FRAME_METHOD) {
        this.method = AMQImpl.readMethodFrom(f.getInputStream());
        this.state = this.method.hasContent() ? CAState.EXPECTING_CONTENT_HEADER : CAState.COMPLETE;
    } else {
        throw new UnexpectedFrameError(f, AMQP.FRAME_METHOD);
    }
}

public Start(MethodArgumentReader rdr) throws IOException {
    this(rdr.readOctet(), rdr.readOctet(), rdr.readTable(), rdr.readLongstr(), rdr.readLongstr());
}

public final int readOctet()
    throws IOException
{
    clearBits();
    return in.readOctet();//in对象是DataInputStream对象
}

private final CommandAssembler assembler;
public boolean handleFrame(Frame f) throws IOException {
    return this.assembler.handleFrame(f);
}

while (_running) {
    Frame frame = _frameHandler.readFrame();

    if (frame != null) {
        _missedHeartbeats = 0;
        if (frame.type == AMQP.FRAME_HEARTBEAT) {
            // Ignore it: we've already just reset the heartbeat counter.
        } else {
            if (frame.channel == 0) { // the special channel
                _channel0.handleFrame(frame);
            } else {
                if (isOpen()) {
                    // If we're still _running, but not isOpen(), then we
                    // must be quiescing, which means any inbound frames
                    // for non-zero channels (and any inbound commands on
                    // channel zero that aren't Connection.CloseOk) must
                    // be discarded.
                    ChannelManager cm = _channelManager;
                    if (cm != null) {
                        cm.getChannel(frame.channel).handleFrame(frame);
                    }
                }
            }
        }
    } else {
        // Socket timeout waiting for a frame.
        // Maybe missed heartbeat.
        handleSocketTimeout();
    }
}

public void handleFrame(Frame frame) throws IOException {
    AMQCommand command = _command;
    if (command.handleFrame(frame)) { // a complete command has rolled off the assembly line
        _command = new AMQCommand(); // prepare for the next one
        handleCompleteInboundCommand(command);
    }
}

/**
 * Sends this command down the named channel on the channel's
 * connection, possibly in multiple frames.
 * @param channel the channel on which to transmit the command
 * @throws IOException if an error is encountered
 */
public void transmit(AMQChannel channel) throws IOException {
    int channelNumber = channel.getChannelNumber();
    AMQConnection connection = channel.getConnection();

    synchronized (assembler) {
        Method m = this.assembler.getMethod();
        connection.writeFrame(m.toFrame(channelNumber));
        if (m.hasContent()) {
            byte[] body = this.assembler.getContentBody();

            connection.writeFrame(this.assembler.getContentHeader()
                    .toFrame(channelNumber, body.length));

            int frameMax = connection.getFrameMax();
            int bodyPayloadMax = (frameMax == 0) ? body.length : frameMax
                    - EMPTY_FRAME_SIZE;

            for (int offset = 0; offset < body.length; offset += bodyPayloadMax) {
                int remaining = body.length - offset;

                int fragmentLength = (remaining < bodyPayloadMax) ? remaining
                        : bodyPayloadMax;
                Frame frame = Frame.fromBodyFragment(channelNumber, body,
                        offset, fragmentLength);
                connection.writeFrame(frame);
            }
        }
    }

    connection.flush();
}