@6river/graphql-google-pubsub 中文文档教程

Question

graphql-google-pubsub

该包实现了 graphql-subscriptions 包中的 PubSubEngine 接口以及新的 AsyncIterator 接口。 它允许您将订阅管理器连接到 Google PubSub 机制以支持 多个订阅管理器实例。

Installation

npm install @axelspringer/graphql-google-pubsub 或者 yarn add @axelspringer/graphql-google-pubsub

Using as AsyncIterator

使用 Subscription 类型定义您的 GraphQL 模式：

schema {
  query: Query
  mutation: Mutation
  subscription: Subscription
}

type Subscription {
    somethingChanged: Result
}

type Result {
    id: String
}

现在，让我们创建一个简单的 GooglePubSub 实例：

import { GooglePubSub } from '@axelspringer/graphql-google-pubsub';
const pubsub = new GooglePubSub();

现在，实现您的订阅类型解析器，使用 pubsub.asyncIterator 映射您需要的事件：

const SOMETHING_CHANGED_TOPIC = 'something_changed';

export const resolvers = {
  Subscription: {
    somethingChanged: {
      subscribe: () => pubsub.asyncIterator(SOMETHING_CHANGED_TOPIC),
    },
  },
}

订阅解析器不是一个函数，而是一个带有 subscribe 方法的对象，它返回 <代码>异步迭代。

调用 GooglePubSub 实例的方法 asyncIterator 将订阅所提供的主题，并将返回绑定到 GooglePubSub 实例并监听任何事件的 AsyncIterator发表在那个话题上。 现在，GraphQL 引擎知道 somethingChanged 是一个订阅，每次我们将对这个主题使用 pubsub.publish 时，GooglePubSub 将 < code>PUBLISH 将事件发送给所有其他订阅的实例，而这些实例又会使用 GraphQL 引擎提供的 next 回调将事件发送给 GraphQL。

pubsub.publish(SOMETHING_CHANGED_TOPIC, { somethingChanged: { id: "123" }});

主题不会自动创建，必须事先创建。

如果您发布非字符串数据，它会被字符串化，您必须解析接收到的消息数据。

Receive Messages

来自 Google PubSub 的收到的消息直接作为有效载荷传递给解析/过滤函数。

您可以在其中提取数据（缓冲区）或使用通用消息处理程序 来转换接收到的消息。

function commonMessageHandler ({attributes = {}, data = ''}) {
  return {
    ...attributes,
    text: data.toString()
  };
}

can use custom message handler 测试说明了通用消息处理程序的灵活性。

Dynamically use a topic based on subscription args passed on the query:

export const resolvers = {
  Subscription: {
    somethingChanged: {
      subscribe: (_, args) => pubsub.asyncIterator(`${SOMETHING_CHANGED_TOPIC}.${args.relevantId}`),
    },
  },
}

Using both arguments and payload to filter events

import { withFilter } from 'graphql-subscriptions';

export const resolvers = {
  Subscription: {
    somethingChanged: {
      subscribe: withFilter(
        (_, args) => pubsub.asyncIterator(`${SOMETHING_CHANGED_TOPIC}.${args.relevantId}`),
        (payload, variables) => payload.somethingChanged.id === variables.relevantId,
      ),
    },
  },
}

Creating the Google PubSub Client

import { GooglePubSub } from '@axelspringer/graphql-google-pubsub';

const pubSub = new GooglePubSub(options, topic2SubName, commonMessageHandler)

Options

这些是传递给内部或通过 Google PubSub 客户端的选项。 如果提供，客户端将从环境变量中提取凭据、项目名称等。 查看身份验证指南了解更多信息。 否则，您可以在选项中提供此详细信息。

const options = {
  projectId: 'project-abc',
  credentials:{
    client_email: 'client@example-email.iam.gserviceaccount.com',
    private_key: '-BEGIN PRIVATE KEY-\nsample\n-END PRIVATE KEY-\n'
  }
};

Subscription Options

订阅选项可以传入subscribe或异步交互器。

注意：google.protobuf.Duration 类型必须作为对象传入具有秒属性 ({ seconds: 123 })。

const dayInSeconds = 60 * 60 * 24;

const subscriptionOptions = {
  messageRetentionDuration: { seconds: dayInSeconds },
  expirationPolicy: {
    ttl: { seconds: dayInSeconds * 2 }, // 2 Days
  },
};

await pubsub.asyncIterator("abc123", subscriptionOptions);

topic2SubName

允许构建不同的工作流程。 如果您在多个服务器实例上侦听同一个订阅，消息将在它们之间分发。 大多数时候，您希望每台服务器有不同的订阅。 这样每个服务器实例都可以通知他们的客户一条新消息。

const topic2SubName = topicName => `${topicName}-${serverName}-subscription`

commonMessageHandler

使用从 Google PubSub 接收到的消息调用公共消息处理程序。 您可以在将消息传递给订阅者的各个过滤器/解析器方法之前对其进行转换。 例如，通过这种方式可以注入 DataLoader 的一个实例，它可以在所有过滤器/解析器方法中使用。

const getDataLoader = () => new DataLoader(...);
const commonMessageHandler = ({attributes: {id}, data}) => ({id, dataLoader: getDataLoader()});

export const resolvers = {
  Subscription: {
    somethingChanged: {
      resolve: ({id, dataLoader}) => dataLoader.load(id)
    },
  },
}

Author

Jonas Hackenberg - jonas-arkulpa

Acknowledgements

这个项目的灵感主要来自 graphql-redis-subscriptions。 非常感谢其作者的工作和灵感。 感谢精益团队（Daniel Vogel、Martin Thomas< /a>、马塞尔·多纳尔、弗洛里安·塔茨基、 Sebastian Herrlinger、Mircea Craculeac 和 蒂姆苏萨）。