如何用 WebSockets 实现 AWS AppSync 实时更新

AWS AppSync 是一个无服务器 GraphQL 服务器，内置缓存和身份验证。GraphQL 是 REST 的现代替代方案，在许多很酷的功能中，其中包括严格的模式，允许客户端准确选择 api 应返回的值。

GraphQL 有三种操作：查询（Query）检索数据；变更（Mutation）更改数据；订阅（Subscription）让客户端在数据更新时获得更新。在本文中，我将探讨如何使用 Mutation 和订阅创建 AppSync API，使客户端能够使用 WebSockets 实时接收来自应用程序的更新。

设置 AppSync Api

GraphQL 模式定义了服务器的操作和数据类型。它是用 SDL（模式定义语言）编写的。

type Mutation {
  sendMessage(msg: String): Message!
}
type Query {
  dummy(param: String): String
}
type Subscription {
  subscribeToMessages: Message
  @aws_subscribe(mutations: ["sendMessage"])
}
type Message {
  message: String
}
schema {
  query: Query
  mutation: Mutation
  subscription: Subscription
}

在上述模式中，我们有一个Mutation sendMessage。它接收一个字符串输入，并向客户端返回一个 Message 对象。每当客户端调用该Mutation时，就会向订阅者发送 Message 对象。AppSync 要求名为 dummy 的查询使模式有效，但在本例中没有用处。

订阅 subscribeToMessages 用于希望从 api 获取实时更新的客户端。更新的数据类型为消息。该订阅还有一个 AWS AppSync 特定指令，由 @aws_subscribe 分配（mutations：[“sendMessage”]）。指令是 GraphQL 的一项功能，但这个指令、这个指令和其他指令只能与 AppSync 服务器配合使用。此指令指示 AppSync 将此订阅与 sendMessage mutations挂钩，每次有人调用 sendMessage 时，响应就会发送给订阅者。

当客户端执行mutation sendMessage 时，输入是字符串。要将其转换为消息对象，我们需要一个 “解析器”。

sendMessageResolver: 
    Type: AWS::AppSync::Resolver
    Properties:
      ApiId: !GetAtt graphQLApi.ApiId
      TypeName: Mutation
      FieldName: sendMessage
      DataSourceName: !GetAtt noneDataSource.Name
      RequestMappingTemplate: |
          {
            "version": "2017-02-28",
            "payload": {
              "message": $util.toJson($context.arguments.msg)
            }
          }
      ResponseMappingTemplate: "$util.toJson($ctx.result)"

上述解析器是使用 CloudFormation 模板定义的，该定义说明了解析器应适用于哪些字段和类型，然后以一种称为 VTL 的格式定义请求和响应模板。解析器也可以用 JavaScript 定义。请求映射模板接收输入并将其映射到消息类型。响应映射只是返回结果，这是将结果发送给订阅者所需要的。解析器非常强大，可以包含大量应用逻辑。

AppSync api 可以有一个或多个数据源，如 DynamoDB、Lambda 或 OpenSearch。不过，本文章使用的是 NoneDataSource。这是一种特殊的数据源，不存储任何数据。我们需要的只是服务器转发数据。例如，如果我们想将所有消息存储在 DynamoDB 表中，我们可以使用 DynamoDB 数据源，其余的都是一样的。

通过部署此 CloudFormation 堆栈，您将在 AWS 控制台中看到 AppSync API。要测试它，请导航到菜单中的查询，然后在两个单独的浏览器窗口中打开它。在第一个窗口中，添加以下订阅并按播放（可能看起来什么都没发生，但这只是因为还没有更新）。

subscription MySubscription {
  subscribeToMessages {
    message
  }
}

在第二个窗口中，添加这一mutation。

mutation MyMutation {
  sendMessage(msg: "Hello world") {
    message
  }
}

现在，如果运行mutation，该消息将出现在订阅窗口中！

使用 JavaScript 订阅实时更新

现在，让我们创建一个可以订阅此 feed 的 JS 客户端。有很多不错的 GraphQL 包，但它们很快就会变得非常庞大。没有必要为了订阅 AppSync API 而拖入大量代码。

在 CloudFormation 堆栈的输出部分可以找到以下值。AppSync 有一个用于常规 GraphQL 的 url 和一个用于实时的 url，不幸的是，要使其正常工作，需要同时使用这两个 url。

const APPSYNC_HOST = "xxxx.appsync-api.<region>.amazonaws.com";
const APPSYNC_REALTIME_HOST = "xxx.appsync-realtime-api.<region>.amazonaws.com";
const APPSYNC_API_KEY = "xxxxxxxxx";

api 设置为使用 api 密钥进行身份验证，因此在打开 WebSocket 时必须包含这些密钥。遗憾的是，WebSocket 不接受标头，因此需要对凭据进行编码并作为查询参数传递。包含主机和 api 密钥的对象会进行 base64 编码和字符串化。该字符串作为查询参数header包含在内，而 Base64 编码的空负载包含在payload参数中。

function encodeAppSyncCredentials() {
  const creds = {
    host: APPSYNC_HOST,
    "x-api-key": APPSYNC_API_KEY,
  };
  const b64Creds = window.btoa(JSON.stringify(creds));

  return b64Creds;
}

function getWebsocketUrl() {
  const header = encodeAppSyncCredentials(APPSYNC_HOST, APPSYNC_API_KEY);
  const payload = window.btoa(JSON.stringify({}));

  const url = `wss://${APPSYNC_REALTIME_HOST}/graphql?header=${header}&payload=${payload}`;

  return url;
}

WebSocket 连接是使用graphql-ws协议建立的。

const url = getWebsocketUrl (); 

const websocket = new  WebSocket (url,[ “graphql-ws” ]);

在开始订阅之前需要执行一个小的握手协议。当连接打开时，客户端发送connection_init. 然后服务器回复connection_ack，然后我们可以发送订阅调用。

创建 WebSocket 打开时的监听器，当它被触发时，消息connection_init将发送到服务器。

websocket.addEventListener("open", () => {
  websocket.send(
    JSON.stringify({
      type: "connection_init",
    })
  );
});

接下来添加一个message侦听器。每次服务器向客户端发送消息时都会触发此事件。

websocket.addEventListener("message", (event) => {
  message = JSON.parse(event.data);
  console.log(message);

  switch (message.type) {
    case "connection_ack":
      startSubscription(websocket);
      break;
    case "start_ack":
      console.log("start_ack");
      break;
    case "error":
      console.error(message);
      break;
    case "data":
      console.log(message.payload.data.subscribeToMessages.message);
      break;
  }
});

如果消息类型是connection_ack，我们就可以发送startSubscription 调用。它有一个随机 uuid、 start类型和包含 GraphQL 订阅查询的有效负载。它还包含一个extensions对象，其中包含我们上面使用的相同授权信息。

function startSubscription(websocket) {
  const subscribeMessage = {
    id: window.crypto.randomUUID(),
    type: "start",
    payload: {
      data: JSON.stringify({
        query: `subscription GetMessagesSub {
                  subscribeToMessages {
                      message
                  }
              }`,
      }),
      extensions: {
        authorization: {
          "x-api-key": APPSYNC_API_KEY,
          host: APPSYNC_HOST,
        },
      },
    },
  };
  websocket.send(JSON.stringify(subscribeMessage));
}

如果服务器的响应是 start_ack，则表示启动信息已被接受，客户端已连接。

如果服务器发送了错误信息，则会将其打印到控制台。

如果类型为 data，则表示客户端已收到一条信息。

有关完整的代码示例，请参阅https://github.com/mindejulian/appsync-demo。

作者：Julian Minde