c++ - 是否可以使用 Varint32 大小前缀的 Protocol Buffer 消息实现类似 'FileInputStream::BackUp()' 的功能？

Question

我正在尝试使用 following implementation 在 C++ 中解析带分隔符的 protobuf 消息(来自文件) readDelimitedFrom() - 也复制如下:

bool readDelimitedFrom(
    google::protobuf::io::ZeroCopyInputStream* rawInput,
    google::protobuf::MessageLite* message) {
  // We create a new coded stream for each message.  Don't worry, this is fast,
  // and it makes sure the 64MB total size limit is imposed per-message rather
  // than on the whole stream.  (See the CodedInputStream interface for more
  // info on this limit.)
  google::protobuf::io::CodedInputStream input(rawInput);

  // Read the size.
  uint32_t size;
  if (!input.ReadVarint32(&size)) return false;

  // Tell the stream not to read beyond that size.
  google::protobuf::io::CodedInputStream::Limit limit =
      input.PushLimit(size);

  // Parse the message.
  if (!message->MergeFromCodedStream(&input)) return false;
  if (!input.ConsumedEntireMessage()) return false;

  // Release the limit.
  input.PopLimit(limit);

  return true;
}

我的问题是我需要根据消息中包含的 uint32_t 字段对消息进行分组并分批处理它们 - 我们称它为 id。

目前，我的主循环中有以下代码:

...
int infd = -1;
_sopen_s(&infd, argv[1], _O_RDONLY | _O_BINARY, _SH_DENYWR, _S_IREAD);

google::protobuf::io::ZeroCopyInputStream *input = 
    new google::protobuf::io::FileInputStream(infd);

std::vector<ProtoMessage> msgList;
bool readMore = true;

do {
    ProtoMessage msg;
    readMore = readNextMessage(input, msg, msgList);

    if (!msgList.empty()) {
        std::cout << "Processing Message Batch - ID: " << msgList[0].id();
        /* some processing done here */
    }
} while (readMore);

readNextMessage()的实现如下:

bool readNextMessage(
    google::protobuf::io::ZeroCopyInputStream* rawInput,
    ProtoMessage& nextMsg,
    std::vector<ProtoMessage>& batchList) {

    bool sameBatch = false;
    uint32_t msgID = 0;
    do {
        if (readDelimitedFrom(rawInput, &scan) == -1)
            return false;
        if (nextMsg.id() == 0)
            msgID = nextMsg.id();    // guaranteed to be non-zero
        if (sameBatch = (msgID == nextMsg.id()))
            batchList.push_back(nextMsg); 
    } while (sameBatch); 

    // need a way to roll-back here as nextMsg is now the first new
    // ProtoMessage belonging to a new batch.

    return true;
}

这个函数的逻辑相当简单:获取一个 ZeroCopyInputStream 并使用 readDelimitedFrom() 解析它以将 ProtoMessage 消息分组到一个 vector 中基于他们的 id 字段。如果遇到具有新 ID 的消息，则停止并将控制返回给 main 以处理消息批处理。

这导致了非期望的要求，即必须使用/读取第一条消息(包括其 Varint32 编码的大小)不属于前一批，而没有办法“备份”溪流。我希望能够将 ZeroCopyInputStream 指向最后一个 readDelimitedFrom() 之前的位置。

有什么方法可以修改readDelimitedFrom()使其也返回调用时消耗的字节数，然后在ZeroCopyInputStream上使用指针运算来实现所需的功能？

提供的函数 ZeroCopyInputStream::Backup() 有一个先决条件，即 ZeroCopyInputStream::Next() 是最后一个方法调用。显然，使用 CodedInputStream 包装器解析分隔消息时情况并非如此。

Answer 1

ZeroCopyInputStream::Backup() 只能备份接收到的最后一个缓冲区。一条消息可能跨越多个缓冲区，因此在给定 ZeroCopyInputStream 接口(interface)的情况下，没有通用的方法来执行您想要的操作。

一些选项:

• 在解析每条消息之前调用rawInput->ByteCount()，以便准确确定消息开始的字节位置。如果您需要回滚，请向后查找底层文件并在其之上重新创建 ZeroCopyInputStream。当然，这只有在您从文件中读取时才有效。
• 当您在新的批处理中遇到一条消息时，将其保存到一边，然后在来电者要求开始阅读下一批处理时将其取回。