Pub/sub emulator与functions-framework本地联调报资源不存在错误

• 核心原因是Node.js版Pub/Sub客户端默认连接谷歌云公网Pub/Sub服务，没有连上你本地启动的模拟器。你用curl调用接口能成功，是因为手动指定了localhost:8043的模拟器地址，但代码初始化Pub/Sub客户端时没有配置模拟器连接参数，也没有读取对应环境变量，所以请求实际发到了公网的GCP服务，公网项目下不存在你本地创建的reports主题，直接返回5 NOT_FOUND错误。
• 你写的发布函数代码本身存在3个逻辑bug，就算连通模拟器也无法正常运行：
  • publishAllMessages方法中给map传入的回调没有加async关键字，直接在内部写await会触发语法错误；同时map返回的是Promise数组，没有用Promise.all等待所有异步发布任务执行完成，函数会直接返回空数组
  • sendReports方法判断发布结果时用的变量名是publishReports，但实际接收返回值的变量是publishMessages，变量名不匹配会直接抛出引用错误
  • publishMessage接收的参数是已经被序列化为JSON字符串的报告内容，代码里却读取data.title属性，字符串不存在title字段，会导致日志打印内容全为undefined
• 之前创建的推送订阅配置错误，填写的pushEndpoint是Pub/Sub主题的REST路径，不是本地消费函数的监听地址，就算消息发布成功也无法推送到消费端。

1. 配置Pub/Sub客户端连接本地模拟器 #

启动Pub/Sub模拟器之后，在运行本地函数的终端窗口执行环境变量注入命令，客户端会自动识别该变量将请求转发到本地模拟器，不再连接公网服务：

$(gcloud beta emulators pubsub env-init --project=abc)

如果使用Windows PowerShell，执行上述命令后按照终端输出的提示手动设置环境变量即可。
注意：该环境变量仅对当前终端窗口生效，每次新开终端启动本地函数都需要重新执行，也可以将PUBSUB_EMULATOR_HOST=localhost:8043、GOOGLE_CLOUD_PROJECT=abc两个变量写到本地.env文件中，启动函数时自动加载。

2. 修正代码中的逻辑错误 #

补全缺失的常量定义，修正异步写法、变量名和参数传递的问题，修正后的参考代码如下：

const { PubSub } = require('@google-cloud/pubsub');
// 显式指定项目ID，避免客户端从本地gcloud配置中读取线上项目ID导致寻址错误
const pubSubClient = new PubSub({ projectId: 'abc' });
const TOPIC = 'reports'; // 补全缺失的主题名常量

exports.sendReports = async (req, res) => {
  const reports = await getReports(); // 从数据库获取报告数据
  const publishMessages = await publishAllMessages(reports); // 逐条发布报告到主题
  // 修正变量名，和上方接收返回值的变量保持一致
  if (publishMessages.length) {
    console.log('reports published');
    res.status(200).send();
  } else {
    console.error(new Error('reports was not published'));
    res.status(400).send();
  }
};

async function publishAllMessages(reports) {
  try {
    // 给map回调加async关键字，用Promise.all等待所有异步发布任务执行完成
    const results = await Promise.all(reports.map(async (report) => {
      const publishResult = await publishMessage(report);
      console.log(`${report.title} published = ${publishResult.published}`);
      return publishResult.published ? report : null;
    }));
    // 过滤发布失败的空值，只返回发布成功的报告
    return results.filter(Boolean);
  } catch (err) {
    console.error(err);
    return [];
  }
}

async function publishMessage(report) {
  try {
    // 调整序列化逻辑，直接传入报告对象，在方法内部做JSON序列化，避免读取不到title属性
    const dataBuffer = Buffer.from(JSON.stringify(report));
    const messageId = await pubSubClient.topic(TOPIC).publish(dataBuffer);
    console.log(`Message ${messageId} published.`);
    return { name: report.title, published: true };
  } catch (err) {
    console.error(err);
    return { name: report.title, published: false };
  }
}

3. 修正推送订阅配置 #

先删除之前创建的错误订阅：

curl -s -X DELETE 'http://localhost:8043/v1/projects/abc/subscriptions/mysub'

重新创建订阅，将pushEndpoint改为本地Pub/Sub消费函数的实际监听地址。比如你用functions-framework将消费函数启动在8081端口，根路径提供服务，就按如下命令创建：

# 如果模拟器直接在本地宿主机运行，用localhost
curl -s -X PUT 'http://localhost:8043/v1/projects/abc/subscriptions/mysub' \
    -H 'Content-Type: application/json' \
    --data '{"topic":"projects/abc/topics/reports","pushConfig":{"pushEndpoint":"http://localhost:8081/"}}'

注意：如果Pub/Sub模拟器运行在Docker容器中，不能填localhost，需要将地址改为http://host.docker.internal:8081/，保证容器能访问到宿主机上运行的函数服务。

4. 全链路联调验证 #

分别启动两个本地函数：

• HTTP触发的发布函数按你本地的调试配置启动，比如监听8080端口
• Pub/Sub触发的消费函数启动在你配置推送的对应端口（比如前面示例的8081）
  调用本地发布函数的HTTP接口，即可跑通「发布函数拉取数据->发送消息到本地Pub/Sub模拟器->模拟器推送消息到消费函数」的完整链路。

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