使用TypeScript结合@google-cloud/pubsub的类型匹配问题求助

问题说明 #

使用@google-cloud/pubsub库时遇到两个核心问题：

1. MessageOptions类型需从内部路径@google-cloud/pubsub/build/src/topic导入，希望直接从主包导入
2. 订阅消息的handleMessage函数类型不匹配，触发TS2769错误

现有发布代码 #

import { Message, PubSub, Subscription, Topic } from '@google-cloud/pubsub'
import { MessageOptions } from '@google-cloud/pubsub/build/src/topic' // 希望替换为主包导入

export const instance = new PubSub()
export const psaWorkerTopic = instance.topic(
  MAIN.PSA_WORKER_PUBSUB_TOPIC_NAME,
  {
    batching: PUBSUB_MAX_BATCHING,
    messageOrdering: true,
  }
)

export const publishPsaQueueMessage = async (
  data: PubSubPsaQueue,
  orderingKey?: string
) => {
  try {
    const dataBuffer = Buffer.from(JSON.stringify(data))
    const message: MessageOptions = {
      data: dataBuffer,
      orderingKey,
    }

    const messageId = await psaWorkerTopic.publishMessage(message)
    logger.log(
      `PubSub::PSA::PUBLISHED::msgId: ${messageId}::batchId: ${data.batchId}::orderKey: ${orderingKey}`,
      {
        partnerId: data.partnerId,
        orderingKey,
      }
    )

    return messageId
  } catch (e) {
    logger.error('PubSub::PSA::Could not publish message: ', e)
    if (orderingKey) psaWorkerTopic.resumePublishing(orderingKey)
    throw e
  }
}

现有接收代码 #

export const handleMessage = async (messageBody: {
  message: {
    data: string
    messageId: string
    publishTime: string
    attributes?: Record<string, string>
    orderingKey?: string
  }
  subscription: string
}): Promise<void> => {
  const { message } = messageBody

  const startTime = Date.now()
  const payloadDataStr = Buffer.from(message.data, 'base64').toString('utf8')
  if (!isValidJSONString(payloadDataStr)) {
    throw new Error(`Invalid Json: ${payloadDataStr}`)
  }

  const payloadData: PubSubPsaQueue = JSON.parse(payloadDataStr)

  logger.debug(
    `PubSub::PSA::START::msgId: ${message.messageId}::batchId: ${payloadData.batchId}::orderKey: ${message.orderingKey}::time: ${message.publishTime}::Received payload`,
    payloadDataStr
  )
}

export const psaSubscriptionListener = () => {
  psaWorkerSubscription.on('message', handleMessage) // 此处触发TS2769错误
  psaWorkerSubscription.on('error', (err) => {
    logger.error('PubSub::PSA::Unexpected error with GCP subscription:', err)
  })
}

TypeScript错误信息 #

TS2769: No overload matches this call.
The last overload gave the following error.
Argument of type
(messageBody: {     message: {         data: string;         messageId: string;         publishTime: string;         attributes?: Record<string, string>;         orderingKey?: string;     };     subscription: string; }) => Promise<void>
is not assignable to parameter of type void
subscription.d.ts(68, 5): The last overload is declared here.

真实接收数据结构 #

测试服务器接收的真实数据为包含message和subscription字段的对象：

• message字段包含：base64格式的data字符串、messageId、publishTime、可选的attributes键值对、可选的orderingKey
• subscription字段为订阅名称字符串

解决方案 #

1. 修复MessageOptions导入路径 #

@google-cloud/pubsub主包已导出MessageOptions类型，直接从主包导入即可：

import { Message, PubSub, Subscription, Topic, MessageOptions } from '@google-cloud/pubsub'

2. 修正消息处理函数的类型匹配 #

需要区分本地/服务器端订阅和HTTP云函数接收两种场景的类型差异：

场景1：本地/服务器端订阅（使用subscription.on('message')）

该场景下，message事件的回调参数是库提供的Message实例，而非自定义对象。正确的处理函数如下：

import { Message } from '@google-cloud/pubsub'

export const handleMessage = async (message: Message): Promise<void> => {
  const startTime = Date.now()
  // 直接从Message实例获取Buffer格式的data，无需手动base64解码
  const payloadDataStr = message.data.toString('utf8')
  
  if (!isValidJSONString(payloadDataStr)) {
    message.nack() // 处理失败时调用nack，通知PubSub重新投递消息
    throw new Error(`Invalid Json: ${payloadDataStr}`)
  }

  const payloadData: PubSubPsaQueue = JSON.parse(payloadDataStr)

  logger.debug(
    `PubSub::PSA::START::msgId: ${message.id}::batchId: ${payloadData.batchId}::orderKey: ${message.orderingKey}::time: ${message.publishTime}::Received payload`,
    payloadDataStr
  )

  message.ack() // 处理完成后调用ack确认消息
}

场景2：HTTP云函数接收PubSub消息

云函数接收的是HTTP请求，请求体为自定义的JSON结构，需使用对应的类型定义：

// 定义云函数接收的PubSub消息结构
interface PubSubHttpMessage {
  message: {
    data: string
    messageId: string
    publishTime: string
    attributes?: Record<string, string>
    orderingKey?: string
  }
  subscription: string
}

// Express云函数示例
export const handlePubSubHttpMessage = async (req: Request, res: Response): Promise<void> => {
  const messageBody: PubSubHttpMessage = req.body
  const { message } = messageBody

  const startTime = Date.now()
  const payloadDataStr = Buffer.from(message.data, 'base64').toString('utf8')
  
  if (!isValidJSONString(payloadDataStr)) {
    res.status(400).send(`Invalid Json: ${payloadDataStr}`)
    return
  }

  const payloadData: PubSubPsaQueue = JSON.parse(payloadDataStr)

  logger.debug(
    `PubSub::PSA::START::msgId: ${message.messageId}::batchId: ${payloadData.batchId}::orderKey: ${message.orderingKey}::time: ${message.publishTime}::Received payload`,
    payloadDataStr
  )

  res.status(204).send() // 返回204表示处理成功
}

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