azure-web-pubsub-ts

Azure Web PubSub SDKs for TypeScript

通过 WebSocket 连接和发布/订阅模式实现实时消息传递。

安装

# Server-side management
npm install @azure/web-pubsub @azure/identity

# Client-side real-time messaging
npm install @azure/web-pubsub-client

# Express middleware for event handlers
npm install @azure/web-pubsub-express

环境变量

WEBPUBSUB_CONNECTION_STRING=Endpoint=https://<resource>.webpubsub.azure.com;AccessKey=<key>;Version=1.0;
WEBPUBSUB_ENDPOINT=https://<resource>.webpubsub.azure.com

服务端：WebPubSubServiceClient

身份验证

import { WebPubSubServiceClient, AzureKeyCredential } from "@azure/web-pubsub";
import { DefaultAzureCredential } from "@azure/identity";

// Connection string
const client = new WebPubSubServiceClient(
  process.env.WEBPUBSUB_CONNECTION_STRING!,
  "chat"  // hub name
);

// DefaultAzureCredential (recommended)
const client2 = new WebPubSubServiceClient(
  process.env.WEBPUBSUB_ENDPOINT!,
  new DefaultAzureCredential(),
  "chat"
);

// AzureKeyCredential
const client3 = new WebPubSubServiceClient(
  process.env.WEBPUBSUB_ENDPOINT!,
  new AzureKeyCredential("<access-key>"),
  "chat"
);

生成客户端访问令牌

// Basic token
const token = await client.getClientAccessToken();
// token.url contains a bearer credential: return only to the authorized client, never log it.

// Token with user ID
const userToken = await client.getClientAccessToken({
  userId: "user123",
});

// Token with permissions
const permToken = await client.getClientAccessToken({
  userId: "user123",
  roles: [
    "webpubsub.joinLeaveGroup.chat-room",
    "webpubsub.sendToGroup.chat-room",  // specific group
  ],
  groups: ["chat-room"],  // auto-join on connect
  expirationTimeInMinutes: 60,
});

发送消息

// Broadcast to all connections in hub
await client.sendToAll({ message: "Hello everyone!" });
await client.sendToAll("Plain text", { contentType: "text/plain" });

// Send to specific user (all their connections)
await client.sendToUser("user123", { message: "Hello!" });

// Send to specific connection
await client.sendToConnection("connectionId", { data: "Direct message" });

// Send with filter (OData syntax)
await client.sendToAll({ message: "Filtered" }, {
  filter: "userId ne 'admin'",
});

群组管理

const group = client.group("chat-room");

// Add user/connection to group
await group.addUser("user123");
await group.addConnection("connectionId");

// Remove from group
await group.removeUser("user123");

// Send to group
await group.sendToAll({ message: "Group message" });

// Close all connections in group
await group.closeAllConnections({ reason: "Maintenance" });

连接管理

// Check existence
const userExists = await client.userExists("user123");
const connExists = await client.connectionExists("connectionId");

// Close connections
await client.closeConnection("connectionId", { reason: "Kicked" });
await client.closeUserConnections("user123");
await client.closeAllConnections();

// Permissions
await client.grantPermission("connectionId", "sendToGroup", { targetName: "chat" });
await client.revokePermission("connectionId", "sendToGroup", { targetName: "chat" });

客户端：WebPubSubClient

连接

import { WebPubSubClient } from "@azure/web-pubsub-client";

// Direct URL
const client = new WebPubSubClient("<client-access-url>");

// Dynamic URL from negotiate endpoint
const client2 = new WebPubSubClient({
  getClientAccessUrl: async () => {
    const response = await fetch("/negotiate");
    const { url } = await response.json();
    return url;
  },
});

// Register handlers BEFORE starting
client.on("connected", (e) => {
  console.log(`Connected: ${e.connectionId}`);
});

client.on("group-message", (e) => {
  console.log(`${e.message.group}: ${e.message.data}`);
});

await client.start();

发送消息

// Join group first
await client.joinGroup("chat-room");

// Send to group
await client.sendToGroup("chat-room", "Hello!", "text");
await client.sendToGroup("chat-room", { type: "message", content: "Hi" }, "json");

// Send options
await client.sendToGroup("chat-room", "Hello", "text", {
  noEcho: true,        // Don't echo back to sender
  fireAndForget: true, // Don't wait for ack
});

// Send event to server
await client.sendEvent("userAction", { action: "typing" }, "json");

事件处理程序

// Connection lifecycle
client.on("connected", (e) => {
  console.log(`Connected: ${e.connectionId}, User: ${e.userId}`);
});

client.on("disconnected", (e) => {
  console.log(`Disconnected: ${e.message}`);
});

client.on("stopped", () => {
  console.log("Client stopped");
});

// Messages
client.on("group-message", (e) => {
  console.log(`[${e.message.group}] ${e.message.fromUserId}: ${e.message.data}`);
});

client.on("server-message", (e) => {
  console.log(`Server: ${e.message.data}`);
});

// Rejoin failure
client.on("rejoin-group-failed", (e) => {
  console.log(`Failed to rejoin ${e.group}: ${e.error}`);
});

Express 事件处理程序

import express from "express";
import { WebPubSubEventHandler } from "@azure/web-pubsub-express";

const app = express();

const handler = new WebPubSubEventHandler("chat", {
  path: "/api/webpubsub/hubs/chat/",
  
  // Blocking: approve/reject connection
  handleConnect: (req, res) => {
    if (!req.claims?.sub) {
      res.fail(401, "Authentication required");
      return;
    }
    res.success({
      userId: req.claims.sub[0],
      groups: ["general"],
      roles: ["webpubsub.sendToGroup.general"],
    });
  },
  
  // Blocking: handle custom events
  handleUserEvent: (req, res) => {
    console.log(`Event from ${req.context.userId}:`, req.data);
    res.success(`Received: ${req.data}`, "text");
  },
  
  // Non-blocking
  onConnected: (req) => {
    console.log(`Client connected: ${req.context.connectionId}`);
  },
  
  onDisconnected: (req) => {
    console.log(`Client disconnected: ${req.context.connectionId}`);
  },
});

app.use(handler.getMiddleware());

// Negotiate endpoint
app.get("/negotiate", async (req, res) => {
  if (!req.user?.id) return res.status(401).end();
  const token = await serviceClient.getClientAccessToken({
    userId: req.user.id, // prerequisite: verified authenticated application session
  });
  res.json({ url: token.url });
});

app.listen(8080);

主要类型

// Server
import {
  WebPubSubServiceClient,
  WebPubSubGroup,
  GenerateClientTokenOptions,
  HubSendToAllOptions,
} from "@azure/web-pubsub";

// Client
import {
  WebPubSubClient,
  WebPubSubClientOptions,
  OnConnectedArgs,
  OnGroupDataMessageArgs,
} from "@azure/web-pubsub-client";

// Express
import {
  WebPubSubEventHandler,
  ConnectRequest,
  UserEventRequest,
  ConnectResponseHandler,
} from "@azure/web-pubsub-express";

最佳实践

1. 使用 Entra ID 身份验证 - 在生产环境中使用 DefaultAzureCredential
2. 在启动前注册处理程序 - 避免错过初始事件
3. 使用群组作为频道 - 按主题/房间组织消息
4. 处理重新连接 - 客户端默认会自动重新连接
5. 在 handleConnect 中验证 - 尽早拒绝未授权的连接
6. 使用 noEcho - 必要时防止消息回传给发送者

适用场景

当需要执行概览中所述的工作流或操作时，可使用此技能。

限制

• 仅当任务明确符合上述范围时，才使用此技能。
• 请勿将输出视为针对特定环境进行验证、测试或专家审查的替代品。
• 如果缺少必需的输入、权限、安全边界或成功标准，请停止并请求澄清。

negotiate 示例需要真实的应用程序身份验证中间件；如果缺少身份信息，必须默认拒绝。根据可信的服务器会话解析用户/群组成员关系，授权每个请求的群组，并仅签发有效期较短的特定群组角色。不得在声称权限限定于群组的同时授予通配符角色。令牌/连接 URL 属于持有者密钥；应在响应/日志中保护这些信息，且不得将已连接的套接字视为对所有群组均已授权。将广播/群组变更限定为用户获准接收的对象和内容。