从Google Cloud PubSub获取Protocol Buffer Schema解析报错求助

我熟悉Google Cloud PubSub、Topic及对应Schema，在为GC PubSub的Topic定义Protocol Buffer类型Schema（采用proto2语法）后，通过Node.js编写的Cloud Function从Firestore读取数据，尝试发布到关联该Schema的PubSub Topic。但调用protobuf.parse解析从PubSub获取的Schema定义时，出现“illegal token 'string'”错误。以下是相关信息及解决思路：

相关代码与信息 #

Schema内容 #

syntax = "proto2";

message ProtocolBuffer {
  string event_name = 1;
  string user_id = 2;
}

Cloud Function代码 #

const Firestore = require('@google-cloud/firestore');
const { PubSub } = require('@google-cloud/pubsub');
const protobuf = require('protobufjs');

const firestore = new Firestore();
const pubsub = new PubSub();

exports.publishToBigQuery = async (event, context) => {
    console.log("event", JSON.stringify(event, 2, null));
    console.log("context", JSON.stringify(context, 2, null));
    const affectedDoc = firestore.doc(`messages/${context.params.documentId}`);

    try {
        const documentSnapshot = await affectedDoc.get();
        if (documentSnapshot.exists) {
            const firestoreData = documentSnapshot.data();

            const topic = pubsub.topic('firestore-document-created-with-proto-schema');

            const schema = pubsub.schema('event-pb-bq');
            const info = await schema.get();
            console.log('info', info);

            let root = new protobuf.Root();
            const type = protobuf.parse(info.definition);
            console.log('type', type);
            const ProtocolBuffer = type.root.lookupType('ProtocolBuffer');
            console.log('ProtocolBuffer', ProtocolBuffer);
            const message = ProtocolBuffer.create(firestoreData);
            console.log('message', message);
            const data = Buffer.from(message.toJSON());
            console.log('data', data);

            const value = await topic.publishMessage({data});
            console.log("Message published", value);
        } else {
            console.log("Document doesn't exist", JSON.stringify(affectedDoc));
        }
    } catch (error) {
        console.error("Error when fetching document", error);
    };
};

错误栈 #

event {"oldValue":{},"updateMask":{},"value":{"createTime":"2022-09-09T16:08:59.107887Z","fields":{"event_name":{"stringValue":"fridayeventname"},"user_id":{"stringValue":"fridayuserid"}},"name":"projects/myproject/databases/(default)/documents/messages/jH9W7SQj2aLh7eK8lRCl","updateTime":"2022-09-09T16:08:59.107887Z"}
context {"eventId":"a00ecca0-0740-4cf8-94bf-15828af8e180-0","eventType":"providers/cloud.firestore/eventTypes/document.create","notSupported":{},"params":{"documentId":"jH9W7SQj2aLh7eK8lRCl"},"resource":"projects/myproject/databases/(default)/documents/messages/jH9W7SQj2aLh7eK8lRCl","timestamp":"2022-09-09T16:08:59.107887Z"}
info {
 name: 'projects/myproject/schemas/event-pb-bq',
 type: 'PROTOCOL_BUFFER',
 definition: 'syntax = "proto2";;\n' +
 '\n' +
 'message ProtocolBuffer {\n' +
 ' string event_name = 1;\n' +
 ' string user_id = 2;\n' +
 '}\n'
}
Error when fetching document Error: illegal token 'string' (line 4)
 at illegal (/workspace/node_modules/protobufjs/src/parse.js:96:16)
 at parseType_block (/workspace/node_modules/protobufjs/src/parse.js:347:31)
 at ifBlock (/workspace/node_modules/protobufjs/src/parse.js:290:17)
 at parseType (/workspace/node_modules/protobufjs/src/parse.js:308:9)
 at parseCommon (/workspace/node_modules/protobufjs/src/parse.js:261:17)
 at Object.parse (/workspace/node_modules/protobufjs/src/parse.js:829:21)
 at exports.publishToBigQuery (/workspace/index.js:26:35)

解决思路 #

1. 修复Schema定义中的语法错误 #

从日志输出的info.definition可以看到，第一行syntax = "proto2";;多了一个分号，这是导致protobufjs解析失败的直接原因：

• 重新在PubSub中更新Schema，将第一行修正为syntax = "proto2";
• 若无法直接修改已创建的Schema，需删除现有Schema后重新创建。

2. 修正Firestore数据处理逻辑 #

从event日志可见，Firestore返回的数据是嵌套结构（如event_name对应{stringValue: "fridayeventname"}），直接用documentSnapshot.data()获取的数据不符合protobuf期望的扁平结构，需先转换：

// 转换Firestore数据为扁平结构
const firestoreData = documentSnapshot.data();
const formattedData = {};
for (const key in firestoreData) {
  if (firestoreData[key].stringValue) {
    formattedData[key] = firestoreData[key].stringValue;
  }
  // 可根据实际字段类型扩展其他处理，比如numberValue等
}

3. 修正消息发布的编码方式 #

当前代码用Buffer.from(message.toJSON())是错误的，protobuf消息需编码为二进制格式才能正确发布到PubSub：

// 替换原有的data生成逻辑
const message = ProtocolBuffer.create(formattedData);
// 验证消息格式是否符合Schema
const errMsg = ProtocolBuffer.verify(message);
if (errMsg) throw new Error(errMsg);
// 编码为二进制Buffer
const data = ProtocolBuffer.encode(message).finish();

4. 可选：预加载本地Schema文件 #

若不想每次从PubSub拉取Schema解析，可直接在代码中引入本地proto文件，避免因PubSub返回的Schema格式异常导致问题：

const root = await protobuf.load('path/to/your/schema.proto');
const ProtocolBuffer = root.lookupType('ProtocolBuffer');

内容的提问来源于stack exchange，提问作者T D