@dxv-系统/实时

实时更改信号结束 AWS App同步事件:订阅-token mintting,SigV4发布,以及一个带有投票倒计时的平反订阅钩.

npm i @dxv-systems/realtime

原则:信号而不是数据

发布一个小的"Stale"有效载荷,让客户端通过正常的,许可检查的路径重新读取(What went stale)router.refresh(), (中文(简体) ), invalidateQueries) (中文(简体) ). 实时层不需要您的数据模型 。 套接字只会缩短延迟时间;投票仍然会限制延迟时间.

所需资源

您需要 AppSync 事件 API 。 配有 HS256 JWT 的订阅者 mintSubscribeToken。该符号是作为(否)发送的 Bearer 前缀)作为 Authorization 在 WebSocket 连接页眉和每个页眉中的值 subscribe 留言。 您的 API 授权方必须用同样的签名秘密来验证该符号, 并只允许在 matchChannel(claims.ch, channel) 这是真的。 出版使用IAM(SigV4).

环境变量

设定所有这些可以实时启用 。 每个服务器功能都读取它们 懒散,在呼叫时间,所以你的应用构建 没有它们,未配置的环境继续工作: publish 变成一个不,并且 mintSubscribeToken 返回时 null。 。 。 .

服务器( U)./server,仅指节点)

import { mintSubscribeToken, realtimeClientConfig, publish } from "@dxv-systems/realtime/server";

• mintSubscribeToken({ sub, channel, ttlSeconds? }): HS256 JWT 与 node:crypto (无) jsonwebtoken (b) 依赖性。 索赔: sub, (中文(简体) ), ns (从) REALTIME_NAMESPACE, (中文(简体) ), ch (单位:千美元) channel 你通过,这可能是一种模式,例如 /myapp/*, (中文(简体) ), iat, (中文(简体) ), exp (最多为 iat + 900,默认值900). 如果第一个部分 channel 没有 ns: 这是一个呼叫器错误, 不是运行时的条件 。 回返 null 当阴道被关闭时.
• realtimeClientConfig({ sub, channel }): 满分 { token, expiresAt, httpEndpoint, wsEndpoint } 您的符号路由返回浏览器, 或 null。 。 。 . expiresAt 实值 时间( 毫秒)而不是JWT的第二精度 exp,因此客户端可以将其与 Date.now() 直接说.
• publish(channel, payload)页:1 POST /event2s超时,永远不要扔。 回返 { ok: true, skipped: true } 不发送 当发布env vars 的未设置。 车身 : { channel, events: [JSON.stringify(payload)] }。 。 。 .
• publishMany(channels, payload):粉丝出道方便. 返回第一个失败,或 { ok: true, skipped: true } 只有每个频道都跳过.

./server 还有再出口 matchChannel。 。 。 .

Token 路由( Next.js App 路由器)

// app/api/realtime/token/route.ts
import { NextResponse } from "next/server";
import { realtimeClientConfig } from "@dxv-systems/realtime/server";
import { getSession } from "@/lib/session";

export async function GET() {
  const session = await getSession();
  if (!session) return new NextResponse(null, { status: 401 });

  const config = realtimeClientConfig({ sub: session.userId, channel: "/myapp/*" });
  return NextResponse.json(config); // null when realtime isn't configured: the client just doesn't connect
}

承诺后公布( 任何节点服务器)

import { publish } from "@dxv-systems/realtime/server";

await db.transaction(async (tx) => { /* write */ });
const result = await publish("/myapp/changes", { type: "changed", paths: ["/dashboard/planning"] });
if (!result.ok) console.error("realtime publish failed", result.error); // never fails the request

异态性( N).(中文(简体) )

import { matchChannel } from "@dxv-systems/realtime";

matchChannel(pattern, channel) 是您授权方应该执行的频道规则 。 无图案 /* 只匹配准确的频道。 结束于 /* 匹配前缀下的每个频道。 节点或DOM依赖,所以在任何捆绑的目标中都是安全的.

反应( E)./react, (中文(简体) ), "use client",对等依赖 react >=18(中文(简体) )

import { useRealtimeChannel, useLiveRefresh, defaultIsBusy } from "@dxv-systems/realtime/react";

useRealtimeChannel(channel, onEvent, { getConfig, onResync? })

执行 AppSync 事件 WebSocket 协议它连接到 wss://<wsEndpoint>/event/realtime 与 aws-appsync-event-ws 和基数64url header-... 子议定书,发送 connection_init,订阅于 connection_ack并重新武装一个守住的监视器 在每一个 ka (它关闭和重新连接 如果没有人到达内部) connectionTimeoutMs) (中文(简体) ). 它召唤 onEvent 与每枚已解析的有效载荷 data 留言.

connected 反映a 实时订阅 (单位:千美元)subscribe_success),而不只是打开的套接字. 重新连接使用被封顶的指数回放与焦急(1s至30s). onEvent, (中文(简体) ), getConfig 财务报告和审定财务报表 onResync 你不必记起他们.

断开与下一个之间发布的事件 subscribe_success 已丢失 : 服务器无法重新发送 . onResync 每起火灾 subscribe_success 在第一个钩子实例之后(每一次重新连接),你可以重新读取状态并关闭缺口。 把它线接到同样的刷新 onEvent数字 :

useRealtimeChannel(channel, onEvent, { getConfig, onResync: () => router.refresh() });

套接字被旋转后 一小时 (单位:千美元)MAX_CONNECTION_AGE_MS),有计划的再连接,没有后退. 这不是一个符号刷新:授权方只运行在连接和订阅上,从不开放订阅,所以一个符号达到15分钟 exp 不需要新的套接字。 一个小时的上限限定了订阅能活过旋转的签名秘密的时间. getConfig 在每个连接上再次被调用,所以每个新连接都带有当前符号.

getConfig 返回时 null 表示"这里没有配置实时". 钩子一直被暂停connected: false,没有套接字,没有出错)并重试 getConfig 每30岁 A 类 subscribe_error 或者说 broadcast_error 日志一次通过 console.warn (与错误的有效载荷)在关闭和后退之前,所以一个错误的信道模式出现在控制台上.

共享连接 : 一个WebSocket每个应用程序,而不是每个钩子。 每个钩子在其频道上注册一个订阅者. 频道的第一个订阅者发送 subscribe,最后一个发送 unsubscribe,套接字只在频道没有订阅者时关闭。 同一频道上的两个钩子共用一个订阅,两个都接受它的活动. 第二个是 connected 立即得到没有 onResync因为它没有错过任何东西。 A 类 subscribe_error 仍然关闭整个套接字(通常的原因是一个已过期的符号,只有再连接才能获取一个新的符号),因此每个信道重新一起订阅,每个已经订阅的钩子都被再同步. 按登记顺序为每个连接指定订阅代号(s1, (中文(简体) ), s2(中文(简体) ). __resetRealtimeConnection() 仅导出用于测试 : 它放弃了模块级连接, 所以每次测试都开始干净.

结合两个钩子

useLiveRefresh 需求 connected 从 useRealtimeChannel时 useRealtimeChannel调用 requestRefresh 从 useLiveRefresh。通过调试打破循环 useRealtimeChannel 首先,通过一个回调,通过一个回调,并填补该回调从 useLiveRefresh返回值 .

"use client";
import { useRef } from "react";
import { useRouter, usePathname } from "next/navigation";
import { useRealtimeChannel, useLiveRefresh } from "@dxv-systems/realtime/react";

export function LiveRefresh() {
  const router = useRouter();
  const pathname = usePathname();
  const requestRefreshRef = useRef<() => void>(() => {});

  function matches(payload: unknown) {
    const paths = (payload as { paths?: string[] })?.paths ?? [];
    return paths.some((p) => pathname.startsWith(p));
  }

  const { connected } = useRealtimeChannel(
    "/myapp/*",
    (e) => {
      if (matches(e)) requestRefreshRef.current();
    },
    {
      getConfig: () => fetch("/api/realtime/token").then((r) => (r.ok ? r.json() : null)),
      onResync: () => requestRefreshRef.current(), // a reconnect may have missed events
    },
  );

  const { requestRefresh } = useLiveRefresh({
    refresh: () => router.refresh(),
    intervalMs: 2 * 60_000,
    connectedIntervalMs: 10 * 60_000,
    connected,
    // isBusy omitted: the built-in heuristic is used
  });
  requestRefreshRef.current = requestRefresh;

  return null;
}

Vite / TanStack 查询

import { useRef } from "react";
import { useQueryClient } from "@tanstack/react-query";
import { useRealtimeChannel, useLiveRefresh } from "@dxv-systems/realtime/react";

function useOrderEvents(orgId: string, orderId: string) {
  const queryClient = useQueryClient();
  const invalidate = () => queryClient.invalidateQueries({ queryKey: ["order", orderId] });

  const requestRefreshRef = useRef<() => void>(() => {});

  const { connected } = useRealtimeChannel(
    `/myapp/org-${orgId}/order-${orderId}`,
    () => requestRefreshRef.current(),
    {
      getConfig: () => fetch("/api/realtime/token").then((r) => (r.ok ? r.json() : null)),
      onResync: () => requestRefreshRef.current(),
    },
  );

  const { requestRefresh } = useLiveRefresh({
    refresh: invalidate,
    intervalMs: 30_000,
    connectedIntervalMs: 2 * 60_000,
    connected,
  });
  requestRefreshRef.current = requestRefresh;
}

useLiveRefresh({ refresh, intervalMs, connectedIntervalMs, connected, isBusy? }) 返回时 { requestRefresh }

电话 refresh() 每个 connected ? connectedIntervalMs : intervalMs,仅在标签可见时。 当标签再次可见时,如果上次刷新时间超过30s,则立即刷新. 当用户忙的时候,它跳过一个勾选,并在15s后安排一个单次重试(不是一叠重试).

requestRefresh() 按要求运行同一门道(可见度检查,忙于检查其单行15s重试,最后再更新),所以实时事件可以在不绕过繁忙门的情况下触发刷新. 它的特性很稳定,可以直接通过 onEvent 或者说 onResync。在相距500米以内调用 组合成一个后继的刷新,所以连续数个事件导致一个刷新。 隐藏标签时的呼叫被推迟,当用户返回标签时运行,无论30s Staleness规则如何.

默认忙碌检查 (目标) isBusy:一个被扣倒的指针总是算作繁忙(5分钟后被打得僵硬,被用安全网来对付被卡住的旗帜). 否则,有重点的输入、文本区域、可选择或可满足的,或开放对话框([role="dialog"] 或者说 [aria-modal="true"],它也匹配 Radix 对话框),只算忙 在后一个的60个之内 keydown, (中文(简体) ), input, (中文(简体) ), pointerdown 或者说 wheel。一个在搜索框中以 Stale 焦点为焦点的闲置墙面显示会不断刷新。 纯决定函数导出为 shouldSkipRefresh(facts: BusyFacts) 测试. useLiveRefresh 在挂载时注册其交互听器,并在卸载时将其删除;在导入时没有附加.

过 isBusy 替换为您自己的 () => boolean。 。 。 . defaultIsBusy() 以无国籍检查方式导出 外部 useLiveRefresh。它没有交互窗口跟踪(一个普通函数不能拥有听众),因此倾向于省略 isBusy 过 defaultIsBusy 改为 useLiveRefresh。 。 。 .

模块格式

输出为普通JS. 这个 "use client" 指令在 ./react 幸存下来的建筑,但土地 之后 页:1 "use strict" 普通JS输出会添加。 环绕 ./react 在你身边 "use client" 组件;不依赖软件包的指令来设定边界.

议定书说明

• data 消息 event 字段. AppSync 协议参考显示 data 留言的 event 作为 JSON 字符串的一元数组,而出版的docs 则描述了一组简单的字符串事件。 钩子既接受空弦又接受阵列 JSON.parse每个项目.
• 出版头条. publish() 发送 content-type: application/json 加上SigV4 标题 aws4fetch 添加(host, (中文(简体) ), x-amz-date, (中文(简体) ), Authorization,匹配 事件 HTTP 发布文件。 。 。 .
• 没有出版重试. aws4fetch 默认情况下,重复5xx和429次答复最多10次,超过2s公布的预算. publish() 禁用这些重试.
• 没有符号驱动重接 . 套接字仅在网络滴入、 持续超时、 a subscribe_error或者一个小时的帽子 提前关闭以刷新符号只会创建一个事件丢失的窗口 .