图表ql- google- pubsub
此软件包执行 PubSubEngine 接口 图ql 订阅 软件包和新的 AsyncIterator 接口. 它允许您将您的订阅管理器连接到 Google PubSub 机制,以支持 多个订阅管理器实例 .
安装
npm install @axelspringer/graphql-google-pubsub
或者说
yarn add @axelspringer/graphql-google-pubsub
使用自动同步器
定义您的图形QL 方案 Subscription 类型 :
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方法,该返回AsyncIterable。 。 。 .
调用方法 asyncIterator 会 议 日 程 和 议 程 GooglePubSub 实例将订阅提供的专题,并将返回 AsyncIterator 与GooglePubSub实例捆绑在一起,并听取有关该主题的任何活动.
现在 GraphQL 引擎知道 somethingChanged 是一个订阅, 每次我们将使用 pubsub.publish 关于这一主题 GooglePubSub 将 PUBLISH 将事件发送到GraphQL next graphQL 引擎给出的回调 .
pubsub.publish(SOMETHING_CHANGED_TOPIC, { somethingChanged: { id: "123" }});
话题不会自动创建,必须事先创建.
如果您发布非字符串数据, 它会被字符串化, 你必须 解析收到的信件数据。 。 。 .
接收信件
这个 收到消息 从Google PubSub作为有效载荷直接传递到分辨率/过滤器函数.
您可以提取那里的数据( Buffer) 或使用 普通信件处理器 以转换收到的消息.
function commonMessageHandler ({attributes = {}, data = ''}) {
return {
...attributes,
text: data.toString()
};
}
这个 can use custom message handler 测试可以显示普通信件处理器的灵活性.
动态使用基于订阅参数的查询主题 :
export const resolvers = {
Subscription: {
somethingChanged: {
subscribe: (_, args) => pubsub.asyncIterator(`${SOMETHING_CHANGED_TOPIC}.${args.relevantId}`),
},
},
}
使用参数和有效载荷过滤事件
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,
),
},
},
}
创建 Google PubSub 客户端
import { GooglePubSub } from '@axelspringer/graphql-google-pubsub';
const pubSub = new GooglePubSub(options, topic2SubName, commonMessageHandler)
选项
这些是 选项 被传递给内部或通过Google PubSub客户端. 如果提供,客户端将从环境变量中提取证书,项目名称等. 看看这个 认证指南 更多信息. 否则您可以在选项中提供此细节 .
const options = {
projectId: 'project-abc',
credentials:{
client_email: '[email protected]',
private_key: '-BEGIN PRIVATE KEY-\nsample\n-END PRIVATE KEY-\n'
}
};
订阅选项
订阅选项 可以传入 subscribe 或者说 asyncInterator。 。 。 .
说明: (原始内容存档于2017-10-29). google.protobuf. 会期 类型必须作为具有一秒属性的对象相传({ seconds: 123 }) (中文(简体) ).
const dayInSeconds = 60 * 60 * 24;
const subscriptionOptions = {
messageRetentionDuration: { seconds: dayInSeconds },
expirationPolicy: {
ttl: { seconds: dayInSeconds * 2 }, // 2 Days
},
};
await pubsub.asyncIterator("abc123", subscriptionOptions);
专题2SubName
允许构建不同的工作流程. 如果您在多个服务器实例中听到同一订阅, 信件将会在它们之间分发 . 大多数时候,您想要每个服务器不同的订阅. 这样每个服务器实例都可以告知客户端新的消息.
const topic2SubName = topicName => `${topicName}-${serverName}-subscription`
普通信使汉德勒
普通消息处理器随从Google PubSub收到的消息被调用. 您可以在消息传递给订阅者的个人过滤/解析方法之前将消息转换. 举例来说,这样可以注射一种 数据记录器 用于所有过滤/解析方法.
const getDataLoader = () => new DataLoader(...);
const commonMessageHandler = ({attributes: {id}, data}) => ({id, dataLoader: getDataLoader()});
export const resolvers = {
Subscription: {
somethingChanged: {
resolve: ({id, dataLoader}) => dataLoader.load(id)
},
},
}
作者
鸣谢
这个项目的灵感主要来自 图克l- redis 订阅。 。 。 . 非常感谢其作者的作品和灵感。 多亏了精益队丹尼尔·沃格尔, (中文(简体) ), 马丁·托马斯, (中文(简体) ), 马塞尔多纳, (中文(简体) ), 弗洛里安·塔兹克, (中文(简体) ), 塞巴斯蒂安·赫林格, (中文(简体) ), Mircea Craculeac 维基百科 财务报告和审定财务报表 蒂姆·苏萨) (中文(简体) ).