Delimited messages read/write support

[friendoye]

I work on an Android project that has several protobuf models generated by Google Protobuf plugin. Right now we're almost done with our migration to Wire. Unfortunately, we stumbled upon migration from parseDelimitedFrom()/writeDelimitedTo() API and can't find a compatible Wire API.

AFAICS, a long time ago someone had the same problem (#633), but it's hard to understand whether that issue was disregarded or Wire maintainers implemented such functionality. Also, I see an initiative (protocolbuffers/protobuf#10229) to align different libraries on different language to unify the approach to delimited messages and there is no mentions about Wire either.

Could you shed a light on this topic and suggest what can we do to complete our migration?

Thanks in advance!

[JakeWharton]

Okay so the code to do this today is a bit annoying, but it's doable. Let's go!

First, I haven't thought about length-delimited protos since that comment in 2016 so my memory on how they work is a little hazy. I believe they're quite simply varint32 length-prefixed messages. This is basically the encoding of a nested message on the wire (sans the tag byte). If they're not that, just replace that varint call with whatever the actual type of the length prefix is.

Encoding in this format is trivial with our APIs:

Buffer buffer = new Buffer();
ProtoWriter writer = new ProtoWriter(buffer);

OneField firstWrite = new OneField.Builder().opt_int32(2).build();
writer.writeVarint32(OneField.ADAPTER.encodedSize(firstWrite));
// System.out.println(buffer.copy().readByteString().hex());
OneField.ADAPTER.encode(writer, firstWrite);
// System.out.println(buffer.copy().readByteString().hex());

OneField secondWrite = new OneField.Builder().opt_int32(Integer.MAX_VALUE).build();
writer.writeVarint32(OneField.ADAPTER.encodedSize(secondWrite));
// System.out.println(buffer.copy().readByteString().hex());
OneField.ADAPTER.encode(writer, secondWrite);
// System.out.println(buffer.copy().readByteString().hex());

Write the size, write the message, repeat. If you are wrapped around a BufferedSource instead of a Buffer you would likely want to call either emit() (or flush()) after each message (or batch of messages) was written.

Reading the format can be done today by doing the inverse operation: read the varint, read the message. However, by default, proto message reading will basically loop forever looking for tags (a message is a sequence of tag+value pairs). The only way we know when to stop (such as in a nested message) is when a limit is pushed. The problem is that ProtoReader sets the limit only when you read the tag and then the length, but length-delimited only has a length. Thus, we have no way to push a limit into ProtoReader.

We have some options on how to fix this...

Option 1, which I did not write, is to use a special Source that wraps the real BufferedSource of data and inserts a synthetic tag before each message is read. Then, when we wrap the ProtoReader around this Source (having buffered it) we read a tag, then a length, then the message. We reset the flag to produce the synthetic tag on the wrapper Source and then repeat until EOF.

Option 2, which I did write, is to use a special Source which wraps the real BufferedSource of data and prevents reading past a certain length. Then, after reading the length of the next message, we wrap the Source, read until "EOF" (which is faked by the wrapper), and repeat.

int firstLength = new ProtoReader(buffer).readVarint32();
// System.out.println(buffer.copy().readByteString().hex());
BufferedSource firstSource = Okio.buffer(new FixedLengthSource(buffer, firstLength));
OneField firstRead = OneField.ADAPTER.decode(firstSource);
// System.out.println(buffer.copy().readByteString().hex());
assertEquals(firstWrite, firstRead);

int secondLength = new ProtoReader(buffer).readVarint32();
// System.out.println(buffer.copy().readByteString().hex());
BufferedSource secondSource = Okio.buffer(new FixedLengthSource(buffer, secondLength));
OneField secondRead = OneField.ADAPTER.decode(secondSource);
// System.out.println(buffer.copy().readByteString().hex());
assertEquals(secondWrite, secondRead);

There's two annoying things about this technique:

• You have to make a new ProtoReader only to read a single varint. This suuuucks. We do this because otherwise the ProtoReader loses track of where we are when we separately read from the original BufferedSource. I'd probably reimplement varint-reading logic to a function that operates directly on the BufferedSource to eliminate this.
• There is no built-in, length-delimiting source in Okio. Whoops. Blame me, or Jesse. It's a long-standing issue: Make FixedLengthSource public API okio#1208. We should ship that regardless.

Option 3, which I sorta wrote, involves exposing an API on ProtoReader to externally start a delimited message. This would be a change in Wire, and it's the best option. We should probably also add a symmetric API to write a length-delimited message as well on ProtoWriter.

ProtoReader reader = new ProtoReader(buffer);

reader.nextDelimited(); // returns length, just because it can.
OneField firstRead = OneField.ADAPTER.decode(reader);
// System.out.println(buffer.copy().readByteString().hex());
assertEquals(firstWrite, firstRead);

reader.nextDelimited();
OneField secondRead = OneField.ADAPTER.decode(reader);
// System.out.println(buffer.copy().readByteString().hex());
assertEquals(secondWrite, secondRead);

So no matter what we should probably do some work to make this easier. And we should expose fixed-length sources in Okio because I'm sick of copying the implementation around (you can find one in OkHttp if you need one today).

I'm going to re-open the original issue. It shouldn't have been closed as there's very clearly action to take here! Feel free to keep discussing here, or in that issue.

[friendoye]

Hooray, your suggestions worked out! 🎉

The described approach of reading delimited messages using FixedLenghtSource is pretty neat solution that we might use in our project.

Looking forward to future updates in Wire!

[JakeWharton]

I just sent #2747 which will make things much simpler as well. No messing with multiple readers or limiting sources. Just nextLengthDelimited() and then adapter decode in a loop!