如何使用Karate DSL测试Protobuf编码的MQTT 5消息？

一、基础项目依赖配置 #

按照Karate标准项目结构搭建即可，测试脚本放在src/test/resources目录，Java工具类放在src/test/java目录，无需额外打包配置。直接将以下依赖抄进pom.xml的dependencies标签内：

• Karate核心测试框架

<dependency>
  <groupId>com.intuit.karate</groupId>
  <artifactId>karate-junit5</artifactId>
  <version>1.4.0.RC5</version>
  <scope>test</scope>
</dependency>

• HiveMQ MQTT5客户端

<dependency>
  <groupId>com.hivemq</groupId>
  <artifactId>hivemq-mqtt-client</artifactId>
  <version>1.3.3</version>
  <exclusions>
    <!-- 排除自带的netty版本，避免和Karate内置netty版本冲突 -->
    <exclusion>
      <groupId>io.netty</groupId>
      <artifactId>netty-all</artifactId>
    </exclusion>
  </exclusions>
</dependency>

• Google动态Protobuf解析库（无需手动编译proto生成Java类）

<dependency>
  <groupId>com.google.protobuf</groupId>
  <artifactId>protobuf-java</artifactId>
  <version>3.25.3</version>
</dependency>
<dependency>
  <groupId>com.google.protobuf</groupId>
  <artifactId>protobuf-dynamic</artifactId>
  <version>1.1.0</version>
</dependency>

二、通用工具类实现（最少Java代码量） #

编写一个无业务耦合的通用工具类，把这个类复制到项目的测试源码目录即可，不需要根据测试场景修改核心逻辑。工具类封装三个核心能力：

1. 初始化HiveMQ MQTT5客户端，提供连接、发布、订阅的基础能力，所有收到的MQTT消息自动存入内存队列
2. 动态Protobuf解析：从用户属性读取.proto全文、二进制payload，输出人类可读的JSON纯文本
3. 动态Protobuf序列化：输入.proto全文、纯文本JSON，输出可直接作为MQTT payload的二进制字节

参考代码示例：

import com.google.protobuf.DescriptorProtos;
import com.google.protobuf.Descriptors;
import com.google.protobuf.DynamicMessage;
import com.google.protobuf.util.JsonFormat;
import com.hivemq.client.mqtt.MqttClient;
import com.hivemq.client.mqtt.datatypes.MqttQos;
import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient;
import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.concurrent.ConcurrentLinkedQueue;

public class MqttProtobufTestUtil {
    // 全局单例MQTT客户端
    private static Mqtt5AsyncClient mqttClient;
    // 存储接收到的MQTT消息，供Karate脚本读取
    public static ConcurrentLinkedQueue<Mqtt5Publish> messageQueue = new ConcurrentLinkedQueue<>();
    // Json转换工具，支持Protobuf和JSON互转
    private static final JsonFormat.Printer jsonPrinter = JsonFormat.printer();
    private static final JsonFormat.Parser jsonParser = JsonFormat.parser();

    // 初始化并连接MQTT broker
    public static void connect(String brokerUrl, String clientId) {
        mqttClient = MqttClient.builder()
                .useMqttVersion5()
                .identifier(clientId)
                .serverHost(brokerUrl.split(":")[0])
                .serverPort(Integer.parseInt(brokerUrl.split(":")[1]))
                .buildAsync();
        // 注册消息回调，收到的消息直接入队
        mqttClient.publishes(mqttPublish -> messageQueue.add(mqttPublish));
        mqttClient.connectWith().cleanStart(true).send().join();
    }

    // 订阅指定topic，QoS传字符串即可
    public static void subscribe(String topic, String qosLevel) {
        MqttQos qos = MqttQos.valueOf(qosLevel);
        mqttClient.subscribeWith().topicFilter(topic).qos(qos).send().join();
    }

    // 解析Protobuf二进制为JSON纯文本
    public static String protobufToJson(String protoContent, byte[] protobufBytes) throws Exception {
        // 从proto文本构造动态Descriptor
        DescriptorProtos.FileDescriptorSet fileSet = DescriptorProtos.FileDescriptorSet.parseFrom(
                protoContent.getBytes(StandardCharsets.UTF_8));
        Descriptors.FileDescriptor fileDescriptor = Descriptors.FileDescriptor.buildFrom(
                fileSet.getFile(0), new Descriptors.FileDescriptor[]{});
        Descriptors.Descriptor messageDescriptor = fileDescriptor.getMessageTypes().get(0);
        // 动态解析二进制
        DynamicMessage message = DynamicMessage.parseFrom(messageDescriptor, protobufBytes);
        // 转成JSON纯文本
        return jsonPrinter.print(message);
    }

    // 把JSON纯文本序列化为Protobuf二进制
    public static byte[] jsonToProtobuf(String protoContent, String jsonContent) throws Exception {
        DescriptorProtos.FileDescriptorSet fileSet = DescriptorProtos.FileDescriptorSet.parseFrom(
                protoContent.getBytes(StandardCharsets.UTF_8));
        Descriptors.FileDescriptor fileDescriptor = Descriptors.FileDescriptor.buildFrom(
                fileSet.getFile(0), new Descriptors.FileDescriptor[]{});
        Descriptors.Descriptor messageDescriptor = fileDescriptor.getMessageTypes().get(0);
        DynamicMessage.Builder builder = DynamicMessage.newBuilder(messageDescriptor);
        jsonParser.merge(jsonContent, builder);
        return builder.build().toByteArray();
    }

    // 发布消息，支持携带用户属性
    public static void publish(String topic, byte[] payload, Map<String, String> userProperties, String qosLevel) {
        MqttQos qos = MqttQos.valueOf(qosLevel);
        var publishBuilder = mqttClient.publishWith()
                .topic(topic)
                .payload(payload)
                .qos(qos);
        // 批量设置用户属性
        userProperties.forEach(publishBuilder::userProperty);
        publishBuilder.send().join();
    }
}

三、Karate测试用例编写（无需额外Java代码） #

写完通用工具类之后，所有测试逻辑都可以用Karate脚本实现，不需要再编写Java代码。示例feature文件如下：

Feature: MQTT5 Protobuf消息测试
  Background:
    * def util = Java.type('com.yourpackage.MqttProtobufTestUtil')
    # 连接测试broker
    * util.connect('127.0.0.1:1883', 'karate-test-client')
    # 订阅测试topic
    * util.subscribe('test/topic', 'AT_LEAST_ONCE')

  Scenario: 验证Protobuf消息双向转换
    # 1. 准备测试数据
    * def testProto = '''syntax = "proto3"; message TestMsg { int32 id = 1; string name = 2; }'''
    * def testJson = '{"id":123,"name":"karate-test"}'
    # 2. 把JSON转成Protobuf二进制，连同proto内容通过用户属性一起发布
    * def protoPayload = util.jsonToProtobuf(testProto, testJson)
    * def userProps = { "proto-content": testProto }
    * util.publish('test/topic', protoPayload, userProps, 'AT_LEAST_ONCE')

    # 3. 等待接收到自己发布的消息
    * def receivedMsg
    * retry(5, 1000) {
      * def msg = util.messageQueue.poll()
      * if (msg != null) { receivedMsg = msg }
      * match msg != null
    }

    # 4. 从消息用户属性提取proto内容，把二进制payload转成纯文本验证
    * def receivedProto = receivedMsg.getUserProperties().asMap().get("proto-content").toString()
    * def receivedText = util.protobufToJson(receivedProto, receivedMsg.getPayloadAsBytes().toByteArray())
    * match receivedText == testJson

四、Java基础薄弱用户实用提示 #

• 不要手动编译.proto生成Java类：动态Protobuf解析方案完全省去了protoc编译、类加载的复杂步骤，不管.proto内容怎么变更，都不需要修改Java代码。
• 如果启动时报类加载、版本冲突错误，直接检查HiveMQ依赖中的netty排除配置是否生效，不需要自行调整其他依赖版本，Karate内置的netty版本已经足够支撑MQTT客户端运行。
• 传输.proto内容时，如果遇到特殊字符导致MQTT用户属性解析失败，直接把.proto内容做Base64编码再存入用户属性，接收时解码即可，避免MQTT协议对用户属性字符串的特殊字符限制。
• 不要在Karate脚本中编写异步回调逻辑：所有MQTT消息都存在内存队列中，测试脚本用轮询方式从队列取消息即可，不需要处理复杂的多线程逻辑，减少脚本出错概率。
• 动态解析Protobuf时，默认取.proto文件中的第一个顶层message做解析，如果.proto内包含多个message，可小幅修改工具类方法，传入需要解析的message全限定名即可，不需要重构核心逻辑。
• 不需要把工具类打包成单独Jar包，直接放在项目的测试源码目录下，Karate运行时会自动加载，完全省去复杂的项目构建配置。

内容的提问来源于stack exchange，提问作者Drunda Nibel