CODEWITHSUNDEEP
pythonsocketsprotocol-buffers
ElSchuer
ElSchuer

Reputation: 68

Receive delimited Protobuf message in python via TCP

I am trying to receive a protobuf message, which was send from a java application with "writeDelmitedTo()" inside my python application.

After some research I already came across this code to read the message from the socket, decode it and parse it.

data = sock.recv()
(size, position) = decoder._DecodeVarint(data, 0)
msg = MessageWrapper_pb2.WrapperMessage().ParseFromString(data[position:position + size])

What I am getting now is a google.protobuf.message.DecodeError: Truncated message Exception.

Has anyone encountered a similar problem or knows how to read delimited data from a socket and parse it correctly?

Edit:

This is the solution that worked for me.

def read_java_varint_delimited_stream(sock):
    buf = []
    data = sock.recv()
    rCount = len(data)
    (size, position) = decoder._DecodeVarint(data, 0)

    buf.append(data)
    while rCount < size+position:
        data = sock.recv(size+position-rCount)
        rCount += len(data)
        buf.append(data)

    return b''.join(buf), size, position

def readMessage(sock):
    data, size, position = read_java_varint_delimited_stream(sock)
    msg = MessageWrapper_pb2.WrapperMessage()
    msg.ParseFromString(data[position:position + size])

    return msg

Upvotes: 3

Views: 4393

Answers (1)

tdelaney
tdelaney

Reputation: 77407

TCP is a stream protocol and there is nothing that says a recv on one end is paired with a single send on the other end. Message based protocols need some way to mark their boundaries so the receiver knows how to find message boundaries.

The writeDelimitedTo docs say that a varint size is sent and then the data. So, read the varint and then read that number of bytes.

Digging deeper, the varint docs describe how its value is encoded by using a byte's high-bit to mark continuation. We we can write up our own decoder

import struct

def read_java_varint_delimited_stream(sock):
    sz = 0
    while True:
        vbyte, = struct.unpack('b', sock.recv(1))
        sz = (vbyte << 7) + (vbyte & 0x7f)
        if not vbyte & 0x80:
            break
    data = []
    while sz:
        buf = sock.recv(sz)
        if not buf:
            raise ValueError("Buffer receive truncated")
        data.append(buf)
        sz -= len(buf)
    return b''.join(buf)

Upvotes: 3

Related Questions