com.google.protobuf.MessageLite#writeTo ( )源码实例Demo

@Override
public void serialize(final Object toSerialize, final Buffer destination) {
    if (!(toSerialize instanceof MessageLite)) {
        throw new SerializationException("Unknown type to serialize (expected MessageLite): " +
                toSerialize.getClass().getName());
    }
    final MessageLite msg = (MessageLite) toSerialize;
    final int size = msg.getSerializedSize();
    // TODO (nkant) : handle compression
    destination.writeByte(0);
    destination.writeInt(size);
    destination.ensureWritable(size);

    final int writerIdx = destination.writerIndex();
    final int writableBytes = destination.writableBytes();
    final CodedOutputStream out = destination.hasArray() ?
            newInstance(destination.array(), destination.arrayOffset() + writerIdx, writableBytes) :
            newInstance(destination.toNioBuffer(writerIdx, writableBytes));
    try {
        msg.writeTo(out);
    } catch (IOException e) {
        throw new SerializationException(e);
    }
    destination.writerIndex(writerIdx + size);
}
 

/**
 * Send a message.
 *
 * @param msg
 *            the message to send
 * @throws CJCommunicationsException
 *             to wrap any occurring IOException
 */
public void send(XMessage message) {
    MessageLite msg = message.getMessage();
    try {
        int type = MessageConstants.getTypeForMessageClass(msg.getClass());
        int size = 1 + msg.getSerializedSize();
        if (this.maxAllowedPacket > 0 && size > this.maxAllowedPacket) {
            throw new CJPacketTooBigException(Messages.getString("PacketTooBigException.1", new Object[] { size, this.maxAllowedPacket }));
        }
        // for debugging
        // System.err.println("Initiating write of message (size=" + size + ", tag=" + ClientMessages.Type.valueOf(type) + ")");
        byte[] sizeHeader = ByteBuffer.allocate(4).order(ByteOrder.LITTLE_ENDIAN).putInt(size).array();
        this.outputStream.write(sizeHeader);
        this.outputStream.write(type);
        msg.writeTo(this.outputStream);
        this.outputStream.flush();
        this.lastPacketSentTime = System.currentTimeMillis();
    } catch (IOException ex) {
        throw new CJCommunicationsException("Unable to write message", ex);
    }
}
 

/**
 * Asynchronously write a message with a notification being delivered to <code>callback</code> upon completion of write of entire message.
 *
 * @param message
 *            message extending {@link XMessage}
 * @param callback
 *            an optional callback to receive notification of when the message is completely written
 */
public void writeAsync(XMessage message, CompletionHandler<Long, Void> callback) {
    MessageLite msg = message.getMessage();
    int type = MessageConstants.getTypeForMessageClass(msg.getClass());
    int size = msg.getSerializedSize();
    int payloadSize = size + 1;
    // we check maxAllowedPacket against payloadSize as that's considered the "packet size" (not including 4 byte size header)
    if (this.maxAllowedPacket > 0 && payloadSize > this.maxAllowedPacket) {
        throw new CJPacketTooBigException(Messages.getString("PacketTooBigException.1", new Object[] { size, this.maxAllowedPacket }));
    }
    // for debugging
    //System.err.println("Initiating write of message (size=" + payloadSize + ", tag=" + com.mysql.cj.mysqlx.protobuf.Mysqlx.ClientMessages.Type.valueOf(type) + ")");
    ByteBuffer messageBuf = ByteBuffer.allocate(HEADER_LEN + size).order(ByteOrder.LITTLE_ENDIAN).putInt(payloadSize);
    messageBuf.put((byte) type);
    try {
        // directly access the ByteBuffer's backing array as protobuf's CodedOutputStream.newInstance(ByteBuffer) is giving a stream that doesn't actually
        // write any data
        msg.writeTo(CodedOutputStream.newInstance(messageBuf.array(), HEADER_LEN, size + HEADER_LEN));
        messageBuf.position(messageBuf.limit());
    } catch (IOException ex) {
        throw new CJCommunicationsException("Unable to write message", ex);
    }
    messageBuf.flip();
    this.bufferWriter.queueBuffer(messageBuf, callback);
}
 

public void send(XMessage message) {
    synchronized (this.waitingAsyncOperationMonitor) {
        MessageLite msg = message.getMessage();
        try {
            int type = MessageConstants.getTypeForMessageClass(msg.getClass());
            int size = 1 + msg.getSerializedSize();
            if (this.maxAllowedPacket > 0 && size > this.maxAllowedPacket) {
                throw new CJPacketTooBigException(Messages.getString("PacketTooBigException.1", new Object[] { size, this.maxAllowedPacket }));
            }
            // for debugging
            // System.err.println("Initiating write of message (size=" + size + ", tag=" + ClientMessages.Type.valueOf(type) + ")");
            byte[] sizeHeader = ByteBuffer.allocate(4).order(ByteOrder.LITTLE_ENDIAN).putInt(size).array();
            this.outputStream.write(sizeHeader);
            this.outputStream.write(type);
            msg.writeTo(this.outputStream);
            this.outputStream.flush();
            this.previousPacketSentTime = this.lastPacketSentTime;
            this.lastPacketSentTime = System.currentTimeMillis();
        } catch (IOException ex) {
            throw new CJCommunicationsException("Unable to write message", ex);
        }
    }
}
 

public void send(XMessage message, CompletionHandler<Long, Void> callback) {
    MessageLite msg = message.getMessage();
    int type = MessageConstants.getTypeForMessageClass(msg.getClass());
    int size = msg.getSerializedSize();
    int payloadSize = size + 1;
    // we check maxAllowedPacket against payloadSize as that's considered the "packet size" (not including 4 byte size header)
    if (this.maxAllowedPacket > 0 && payloadSize > this.maxAllowedPacket) {
        throw new CJPacketTooBigException(Messages.getString("PacketTooBigException.1", new Object[] { size, this.maxAllowedPacket }));
    }
    // for debugging
    //System.err.println("Initiating write of message (size=" + payloadSize + ", tag=" + com.mysql.cj.mysqlx.protobuf.Mysqlx.ClientMessages.Type.valueOf(type) + ")");
    ByteBuffer messageBuf = ByteBuffer.allocate(HEADER_LEN + size).order(ByteOrder.LITTLE_ENDIAN).putInt(payloadSize);
    messageBuf.put((byte) type);
    try {
        // directly access the ByteBuffer's backing array as protobuf's CodedOutputStream.newInstance(ByteBuffer) is giving a stream that doesn't actually
        // write any data
        msg.writeTo(CodedOutputStream.newInstance(messageBuf.array(), HEADER_LEN, size));
        messageBuf.position(messageBuf.limit());
    } catch (IOException ex) {
        throw new CJCommunicationsException("Unable to write message", ex);
    }
    messageBuf.flip();
    this.bufferWriter.queueBuffer(messageBuf, callback);
}
 

@Override
public void sendProtoMessage(MessageLite message) throws IOException {
  // Write message
  if (delimited) {
    try {
      message.writeDelimitedTo(out);
      out.flush();
    } catch (IOException e) {
      // Cannot write anymore, just close socket
      socket.close();
      throw e;
    }
  } else {
    message.writeTo(out);
    out.flush();
    socket.shutdownOutput();
  }
}
 

/**
 * Encode a response object to protobuf format for the HttpResponse.
 *
 * @param buf target byte buffer for encoding
 * @param acceptCharset Accept-Charset header
 * @param object object to encode
 * @return ByteBuf representing protobuf formatted bytes
 */
@Override
public ByteBuf encode(ByteBuf buf, CharSequence acceptCharset, Object object) throws IOException {
  if (!(object instanceof MessageLite)) {
    throw new IllegalArgumentException(
        String.format("%s does not extend from MessageLite", object.getClass().getName()));
  }
  MessageLite msg = (MessageLite) object;
  try (ByteBufOutputStream stream = new ByteBufOutputStream(buf)) {
    msg.writeTo(stream);
    return buf;
  }
}
 

public static ByteBuf protoByteBuf(MessageLite msg) {
    final ByteBuf buf = Unpooled.buffer();
    try (ByteBufOutputStream os = new ByteBufOutputStream(buf)) {
        msg.writeTo(os);
    } catch (IOException e) {
        throw new UncheckedIOException(e);
    }
    return buf;
}
 

public void testMessageLiteInterface() throws Exception {
  ExtensionRegistryLite registry = ExtensionRegistryLite.newInstance();
  TypicalData data = TypicalData.newBuilder().build();
  MessageLite messageLite = data;
  MessageLite.Builder builderLite = messageLite.newBuilderForType();
  messageLite.writeTo(new ByteArrayOutputStream());
  messageLite.writeDelimitedTo(new ByteArrayOutputStream());
  builderLite.mergeFrom(new ByteArrayInputStream(new byte[0]));
  builderLite.mergeFrom(new ByteArrayInputStream(new byte[0]), registry);
  builderLite.mergeDelimitedFrom(new ByteArrayInputStream(new byte[0]));
  builderLite.mergeDelimitedFrom(new ByteArrayInputStream(new byte[0]), registry);
  assertEquals(0, messageLite.getSerializedSize());
}
 

private void writeMessage(ByteBuf out, int messageType, MessageLite message) throws IOException {
    ByteArrayOutputStream byteOutput = new ByteArrayOutputStream();
    message.writeTo(byteOutput);
    out.writeInt(messageType);
    this.writeBytes(out, byteOutput.toByteArray());
}