Protobuf Java method parseFrom() hangs when receiving message from MQTT Server

Question

I'm currently trying to publish and receive Protobuf messages via a Mosquitto MQQT Server. I am succesfully publishing the proper to the Server. However, when a client receives it, the method parseFrom() hangs and never returns. This is a very similar issue to this one, occuring when a Protobuf message is sent via a Socket that's never closed.

Publisher :

MqttClient adapterClient = new MqttClient(broker, clientID);
SpecsMessage.Specs protoNotifyMessage = SpecsMessage.Specs.newBuilder()
                    .setNodeType("basic")
                    .setAddress(serverSocket.getInetAddress().getHostName())
                    .setPort(serverSocket.getLocalPort())
                    .build();
MqttMessage notifyMessage = new MqttMessage(protoNotifyMessage.toString().getBytes());
adapterClient.publish("availableNodes", notifyMessage);

Subscriber :

public class TestController implements MqttCallback {
  public void messageArrived(String topic, MqttMessage message){
                System.out.println("New node connected");     
                System.out.println("Payload: \n" + new String(message.getPayload()));                           
                SpecsMessage.Specs protoMessage = SpecsMessage.Specs.parseFrom(message.getPayload());
  }
}

I could not find a way to specify to the MQQT Server a correct way to send the message.

I also tried using the writeDelimitedFrom() method.

MqttClient adapterClient = new MqttClient(broker, clientID);
SpecsMessage.Specs protoNotifyMessage = SpecsMessage.Specs.newBuilder()
                    .setNodeType("basic")
                    .setAddress(serverSocket.getInetAddress().getHostName())
                    .setPort(serverSocket.getLocalPort())
                    .build();
ByteArrayOutputStream output = new ByteArrayOutputStream();
protoNotifyMessage.writeDelimitedTo(output);
MqttMessage notifyMessage = new MqttMessage(output.toByteArray());
adapterClient.publish("availableNodes", notifyMessage);

However, the message is not correctly converted in byte[], here's what it should look like :

nodeType: "basic"
address: "0.0.0.0"
port: 43101

And here's what I get :

basic0.0.0.0��

Is there a way to make this work, either by correcting the sending method or by solving the byte[] conversion problem?

I doubt this is the issue, but I would be surprised if parseFrom itself hangs, rather than the method providing the data. new String(payload) works, though. Does it work if you store the payload in a variable, and then use in both the println and parseFrom calls? — Andy Turner
Thank you for your remark. I tried it but parseFrom() still hangs on me. However, as you suggested, this does point out that it is not a communication problem. — joss-enet
You should also be careful about using the default charset with getBytes() and new String: you'll get unexpected results if the publisher and subscriber JVMs use different default charsets. — Andy Turner

Andy Turner Andy Turner · Accepted Answer · 2020-04-02T13:34:19

You're trying to use parseFrom to parse the text-formatted proto. parseFrom is for parsing the wire format.

Send the proto in the wire format instead - message.toByteArray().

(If you want to parse from the text format, use TextFormat).

Protobuf Java method parseFrom() hangs when receiving message from MQTT Server

1 Answers