Meteor 虽隐藏了底层 WebSocket，但仍可通过 Meteor.server.stream_server.open_sockets 和 Meteor.connection._stream 直接操作原生 socket，实现服务端主动推送（类似 ws.send()）与客户端监听（类似 ws.on()），适用于 Elasticsearch 同步等非响应式数据场景。
meteor 虽隐藏了底层 websocket，但仍可通过 `meteor.server.stream_server.open_sockets` 和 `meteor.connection._stream` 直接操作原生 socket，实现服务端主动推送（类似 `ws.send()`）与客户端监听（类似 `ws.on()`），适用于 elasticsearch 同步等非响应式数据场景。
在 Meteor 应用中，当需要绕过标准的 publish/subscribe 模型、实现服务端主动向客户端广播事件（例如 MongoDB 变更触发 Elasticsearch 实时同步完成通知），原生 WebSocket 通信是最直接的方案。Meteor 底层基于 DDP 协议运行于 WebSocket 之上，因此完全支持对 socket 实例的访问和控制——关键在于正确获取服务端 socket 引用并绑定客户端监听器。
✅ 服务端：获取并发送消息到指定或全部客户端
Meteor 并未暴露公开 API 直接访问 socket，但可通过内部属性安全访问（注意：此为稳定可用的内部接口，自 Meteor 1.8+ 至最新版本均保持一致）：
// server/main.js
import { Meteor } from 'meteor/meteor';
Meteor.startup(() => {
// 监听新连接（可选：用于初始化或日志）
Meteor.onConnection((connection) => {
console.log('Client connected:', connection.id);
});
// 示例：模拟 MongoDB 更改后向所有客户端广播 ES 同步完成事件
const notifyElasticsearchSync = (docId, status = 'success') => {
const sockets = Meteor.server.stream_server.open_sockets;
const message = JSON.stringify({
msg: 'es-sync',
docId,
status,
timestamp: new Date().toISOString()
});
sockets.forEach(socket => {
try {
socket.send(message);
} catch (e) {
// 忽略已断开连接的 socket（如客户端刷新/关闭）
console.debug('Failed to send to socket:', e.message);
}
});
};
// 在你的 MongoDB observe 或 async hook 中调用
// Collection.find().observeChanges({ changed: () => notifyElasticsearchSync(...) });
});
⚠️ 注意事项：
open_sockets是一个实时更新的Array，包含当前所有活跃 DDP 连接对应的 socket 实例；- 发送前建议
try/catch包裹，避免因 socket 已关闭导致异常中断；- 若需定向推送（如仅通知某用户），应结合
connection.id与 socket 的_meteorSession.id进行匹配（见原始示例），但多数通知类场景推荐广播。
✅ 客户端：监听自定义 DDP 消息
客户端通过 Meteor.connection._stream（即底层 SockJS/WebSocket 实例）监听原始 message 事件。注意：该 stream 接收的是原始字符串，需手动解析：
// client/main.js
Meteor.startup(() => {
// 确保连接已建立后再监听
if (Meteor.connection._stream?.readyState === 1) {
setupCustomMessageListener();
} else {
// 等待连接就绪（如页面加载时 DDP 尚未握手完成）
Meteor.connection.onReconnect = setupCustomMessageListener;
}
});
function setupCustomMessageListener() {
Meteor.connection._stream.on('message', (dataStr) => {
let data;
try {
data = JSON.parse(dataStr);
} catch (e) {
console.warn('Invalid custom message format:', dataStr);
return;
}
// 按 msg 字段路由业务逻辑
if (data.msg === 'es-sync') {
console.info('[ES Sync]', `Document ${data.docId} synced: ${data.status}`);
// ✅ 触发 UI 更新、Toast 提示、或触发局部 re-render
// 如使用 React：useTracker 或 Context 更新状态
}
});
}
? 补充说明：
WebSocket 8.18.2下载WebSocket 8.18.2 是该协议规范的一个重要迭代版本，主要优化了连接稳定性与数据传输效率。它通过全双工通信机制，允许客户端与服务器在单一长连接上实时交换数据，大幅降低传统 HTTP 轮询的开销。该版本增强了心跳保活、自动重连及二进制帧传输能力，适用于即时通讯、在线游戏及金融行情推送等低延迟场景，为开发者提供更可靠的实时网络交互基础。
Meteor.connection._stream是 Meteor 内部维护的DDP.Stream实例，在生产环境稳定可用；- 所有自定义消息必须使用
JSON.stringify()发送，且建议统一msg字段作为类型标识，便于解耦与扩展；- 此方式不经过
Mongo.Collection或Minimongo，因此不会触发任何 reactive context，适合轻量级事件通知。
? 替代建议：增强型 Pub/Sub（进阶可选）
若未来需支持响应式查询 + 自定义事件混合场景，可封装一个“事件订阅”模式：
// server
Meteor.publish('esSyncEvents', function() {
const self = this;
// 不返回 cursor，仅维持连接并监听全局事件总线
const handler = (event) => self.write(new DDPCommon.MethodResult(event));
EventEmitter.on('es:sync', handler);
self.ready();
self.onStop(() => EventEmitter.off('es:sync', handler));
});
但对纯通知场景而言，原生 socket 方案更轻量、延迟更低、逻辑更清晰。
综上，Meteor 完全支持类 ws.send() / ws.on() 的底层通信能力——它不是缺失的功能，而是被封装在 DDP 抽象之下。合理利用内部接口，即可在保持 Meteor 生态优势的同时，精准满足实时事件推送需求。
