MQ Consumer Development Guide

Consumption Protocol

MQ currently supports clients to connect and consume data using MQTT, and only supports TLS encryption. Standard MQTT v3.1.1 is supported by default.

SDK Download

A third-party SDK is recommended. Please visit https://github.com/mqtt/mqtt.github.io/wiki/libraries

Development Process

The consumer development process is as follows:

1 Client Connection

The client is connected to the server by sending a MQTT CONNECT message that consists of three elements, as follows:

In addition, there are some limitations on the use of keepalive, will and session fields in a CONNNECT message, as follows:

2 Subscribing to Topics

• Subscribe Message

  In a Subscribe message, topic has the following format, where wildcards are not supported.

  $sys/pb/consume/$instance name/$TOPIC/$SUB

  When you create a subscription, request QoS must be set to be greater than 0, otherwise the subscription fails.

• Subscribe Acknowledged Message

  A subscription is acknowledged using a MQTT SubAck message, and it returns 0x01 (fixed) if successful.

• Unsubscribe Message

  The client cancels a subscription using a MQTT Unsubscribe message.

• Unsubscribe Acknowledgement Message

  The server acknowledges the cancellation using a MQTT UnsubAck message.

3 Message consumption

• After subscribe succeeded, MQ will actively push publish message to clients, depending on message production.
• At the time of consumption, a push notification from MQ will end up as QoS1.
• After data are consumed by the client, it should reply with a puback message according to the order in which data were acquired.

4 Data Parsing

After receiving the publish message, the client parses its payload segments according to the MQTT protocol, and then parses payload data content, following the steps below.

Step 1 Install protobuf

Download link: https://github.com/protocolbuffers/protobuf/releases

Step 2 Download .proto file below

onenet-mq.proto interface file is as follows:

syntax = "proto3";

package mq;

message Msg{
    uint64 msgid          = 1; //MQ中该消息的真实id
    bytes data            = 2; //具体的数据
    uint64 timestamp      = 3; //精确到ms
}

Save onenet-mq.protofile locally

Step 3 Compile proto file

Compile the file by language, taking Java as an example.

protoc --java_out=$DST_DIR $SRC_DIR/onenet-mq.proto

Step 4 Add compiler-generated source files to a project

Taking Java as an example, compile and generate an OnenetMq.java file, and add this source file to a project.

Step 5 Read data

• Call parseFrom() method to create a Msg object.
• Call obj.getMsgid() to get message ID.
• Call obj.getData() to get message data (view data format)
• Call obj.getTimestamp() to get a message timestamp in milliseconds.

An example of Java is as follows:

OnenetMq.Msg obj;
obj = OnenetMq.Msg.parseFrom(mqttPayload);
System.out.println(obj.getMsgid());
System.out.println(new String(obj.getData().toByteArray()));
System.out.println(obj.getTimestamp());