如何在 express.js 中使用服务器发送的事件
How to use server-sent-events in express.js
我用express.js设置我的REST服务器。现在我想将 sse 添加到此服务器。实现此 sse 包后,出现错误。我知道我收到此错误,何时会尝试使用res.send
两次,但我不是。
ERROR: Error: Can't set headers after they are sent.
at ServerResponse.OutgoingMessage.setHeader (http.js:690:11)
at ServerResponse.header (/home/root/node_modules/express/lib/response.js:718:10)
at ServerResponse.send (/home/root/node_modules/express/lib/response.js:163:12)
at app.get.str (/home/root/.node_app_slot/main.js:1330:25)
at Layer.handle [as handle_request] (/home/root/node_modules/express/lib/router/layer.js:95:5)
at next (/home/root/node_modules/express/lib/router/route.js:131:13)
at sse (/home/root/node_modules/server-sent-events/index.js:35:2)
at Layer.handle [as handle_request] (/home/root/node_modules/express/lib/router/layer.js:95:5)
at next (/home/root/node_modules/express/lib/router/route.js:131:13)
at Route.dispatch (/home/root/node_modules/express/lib/router/route.js:112:3)
我是否有可能不能再在 sse 函数中使用表达式方法?例如:
app.get('/events', sse, function(req, res) {
res.send('...');
});
此外,我找到了这个解决方案和这个。是否可以使用 res.write
函数或其他方式制作 sse,而无需使用其他包?
我不同意使用 Socket.IO 来实现基本的服务器发送事件。浏览器 API 非常简单,Express 中的实现只需要对正常的响应路由进行一些更改:
app.get('/streaming', (req, res) => {
res.setHeader('Cache-Control', 'no-cache');
res.setHeader('Content-Type', 'text/event-stream');
res.setHeader('Access-Control-Allow-Origin', '*');
res.setHeader('Connection', 'keep-alive');
res.flushHeaders(); // flush the headers to establish SSE with client
let counter = 0;
let interValID = setInterval(() => {
counter++;
if (counter >= 10) {
clearInterval(interValID);
res.end(); // terminates SSE session
return;
}
res.write(`data: ${JSON.stringify({num: counter})}'n'n`); // res.write() instead of res.send()
}, 1000);
// If client closes connection, stop sending events
res.on('close', () => {
console.log('client dropped me');
clearInterval(interValID);
res.end();
});
});
- 根据规范设置适当的标头
- 使用 res.flushHeaders(( 建立 SSE 连接
- 使用 res.write(( 而不是 res.send(( 发送数据
- 要结束来自服务器的流,请使用 res.end((
上面的代码片段使用 setInterval(( 模拟向客户端发送数据 10 秒,然后结束连接。客户端将收到丢失连接的错误,并自动尝试重新建立连接。若要避免这种情况,可以在出错时关闭客户端,或者让浏览器发送客户端理解为正常关闭的特定事件消息。如果客户端关闭连接,我们可以捕获 'close' 事件以正常结束服务器上的连接并停止发送事件。
快递:4.17.1node: 10.16.3
您绝对可以在没有其他软件包的情况下实现这一目标。
我写了一篇关于这个的博客文章,第 1 部分列出了基础知识。
您不能关闭 SSE,因为这会破坏功能。重点是它是一个开放的HTTP连接。这允许在任何时候将新事件推送到客户端。
这为 John 的出色答案添加了一个完整的、可运行的示例,并进行了调整,添加了Connection: keep-alive
标题。还包括一个客户端来读取流并处理多个块同时到达的可能性,这似乎是fetch
的特征。
JSON 并非绝对必要,但可用于将数据有效负载与 SSE 元数据分开。
server.js
:
const express = require("express");
const app = express();
app.use(express.static("public"));
app.get("/stream", (req, res) => {
res.writeHead(200, {
"Connection": "keep-alive",
"Cache-Control": "no-cache",
"Content-Type": "text/event-stream",
});
let counter = 0;
const interval = setInterval(() => {
const chunk = JSON.stringify({chunk: counter++});
res.write(`data: ${chunk}'n'n`);
}, 100);
res.on("close", () => {
clearInterval(interval);
res.end();
});
});
const listener = app.listen(process.env.PORT || 3000, () =>
console.log(`Your app is listening on port ${listener.address().port}`)
);
public/index.html
:
<!DOCTYPE html>
<html lang="en">
<head><title>SSE POC</title></head>
<body>
<script>
(async () => {
const response = await fetch("/stream", {
headers: {
"Accept": "text/event-stream",
},
});
if (!response.ok) {
throw Error(response.statusText());
}
for (const reader = response.body.getReader(); ; ) {
const {value, done} = await reader.read();
if (done) {
break;
}
const chunk = new TextDecoder().decode(value);
const subChunks = chunk.split(/(?<=})'n'ndata: (?={)/);
for (const subChunk of subChunks) {
const payload = subChunk.replace(/^data: /, "");
document.body.innerText = JSON.parse(payload).chunk;
}
}
})();
</script>
</body>
</html>
node server.js
后,将浏览器导航到localhost:3000
。您也可以直接使用 curl localhost:3000/stream
测试流。
我不会重复 John 回答中的注释,但简而言之,我们设置了必要的标头并刷新它们以开始连接,然后使用 res.write
发送大量数据。调用 res.end()
以终止服务器上的连接或侦听关闭连接的客户端的res.on("close", ...)
。
客户端使用fetch
和response.body.getReader()
,可以用const {value, done} = await reader.read()
读取,用TextDecoder().decode(value)
解码。
另请参阅 https://masteringjs.io/tutorials/express/server-sent-events
Express 4.18.2,节点 18.16.0,Chrome 版本 114.0.5735.110(官方版本((64 位(
自我推销:我编写了 ExpreSSE 包,它提供了在 express 中使用 SSE 的中间件,你可以在 npm: @toverux/expresse 上找到它。
一个简单的例子:
router.get('/events', sse(/* options */), (req, res) => {
let messageId = parseInt(req.header('Last-Event-ID'), 10) || 0;
someModule.on('someEvent', (event) => {
//=> Data messages (no event name, but defaults to 'message' in the browser).
res.sse.data(event);
//=> Named event + data (data is mandatory)
res.sse.event('someEvent', event);
//=> Comment, not interpreted by EventSource on the browser - useful for debugging/self-documenting purposes.
res.sse.comment('debug: someModule emitted someEvent!');
//=> In data() and event() you can also pass an ID - useful for replay with Last-Event-ID header.
res.sse.data(event, (messageId++).toString());
});
});
还有另一个中间件可以将相同的事件推送到多个客户端。
从您正在使用的库的文档来看,在将其用作函数的中间件时,您应该使用res.sse
。看:https://www.npmjs.com/package/server-sent-events
但是,正如您提到的,所有这些实际上是从他们的代码中完成的包装res.write
。看:https://github.com/zacbarton/node-server-sent-events/blob/master/index.js#L11
新答案:
只需使用 socket.io,它就更容易更好!https://www.npmjs.com/package/socket.io#in-conjunction-with-express
基本设置:
const express = require('express');
const PORT = process.env.PORT || 5000;
const app = express();
const server = require('http').createServer(app);
const io = require('socket.io')(server);
// listen to socket connections
io.on('connection', function(socket){
// get that socket and listen to events
socket.on('chat message', function(msg){
// emit data from the server
io.emit('chat message', msg);
});
});
// Tip: add the `io` reference to the request object through a middleware like so:
app.use(function(request, response, next){
request.io = io;
next();
});
server.listen(PORT);
console.log(`Listening on port ${PORT}...`);
在任何路由处理程序中,您可以使用 socket.io:
app.post('/post/:post_id/like/:user_id', function likePost(request, response) {
//...
request.io.emit('action', 'user liked your post');
})
客户端:
<script src="/socket.io/socket.io.js"></script>
<script src="https://code.jquery.com/jquery-1.11.1.js"></script>
<script>
$(function () {
var socket = io();
$('form').submit(function(e){
e.preventDefault(); // prevents page reloading
socket.emit('chat message', $('#m').val());
$('#m').val('');
return false;
});
socket.on('chat message', function(msg){
$('#messages').append($('<li>').text(msg));
});
});
</script>
完整示例:https://socket.io/get-started/chat/
原答案:
有人(用户:https://stackoverflow.com/users/451634/benny-neugebauer | 来自本文:自定义对象上的 addEventListener(从字面上给了我一个提示,告诉我如何在没有任何其他包的情况下实现这一点,除了 express!我有它的工作!
首先,导入节点的事件发射器:
const EventEmitter = require('events');
然后创建一个实例:
const Stream = new EventEmitter();
然后为事件流创建 GET 路由:
app.get('/stream', function(request, response){
response.writeHead(200, {
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
'Connection': 'keep-alive'
});
Stream.on("push", function(event, data) {
response.write("event: " + String(event) + "'n" + "data: " + JSON.stringify(data) + "'n'n");
});
});
在此 GET 路由中,您回写请求为 200 OK,内容类型为文本/事件流,无缓存,并保持活动状态。
您还将调用 EventEmitter 实例的 .on 方法,该方法需要 2 个参数:要侦听的事件字符串和处理该事件的函数(该函数可以接受给定的参数(
现在。。。。发送服务器事件所要做的就是调用 EventEmitter 实例的 .emit 方法:
Stream.emit("push", "test", { msg: "admit one" });
第一个参数是要触发的事件的字符串(确保它与 GET 路由中的参数相同(。.emit 方法的每个后续参数都将传递给侦听器的回调!
就是这样!
由于您的实例是在路由定义之上的范围内定义的,因此您可以从任何其他路由调用 .emit 方法:
app.get('/', function(request, response){
Stream.emit("push", "test", { msg: "admit one" });
response.render("welcome.html", {});
});
由于 JavaScript 的作用域如何工作,您甚至可以将该 EventEmitter 实例传递给其他函数,甚至可以从其他模块传递:
const someModule = require('./someModule');
app.get('/', function(request, response){
someModule.someMethod(request, Stream)
.then(obj => { return response.json({}) });
});
在某个模块中:
function someMethod(request, Stream) {
return new Promise((resolve, reject) => {
Stream.emit("push", "test", { data: 'some data' });
return resolve();
})
}
就这么简单!无需其他套餐!
以下是指向 Node 的事件发射器类的链接: https://nodejs.org/api/events.html#events_class_eventemitter
我的例子:
const EventEmitter = require('events');
const express = require('express');
const app = express();
const Stream = new EventEmitter(); // my event emitter instance
app.get('/stream', function(request, response){
response.writeHead(200, {
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
'Connection': 'keep-alive'
});
Stream.on("push", function(event, data) {
response.write("event: " + String(event) + "'n" + "data: " + JSON.stringify(data) + "'n'n");
});
});
setInterval(function(){
Stream.emit("push", "test", { msg: "admit one" });
}, 10000)
- 提示使用服务器端事件处理程序激活JavaScript
- 未命中服务器发送的事件侦听器
- 服务器上的Meteor客户端断开连接事件
- 服务器端事件+客户端事件与Websocket
- 服务器发送的事件直到脚本完成才更新
- 事件循环的MEAN.JS setInterval进程(从另一个服务器获取数据)
- 为什么我没有得到对Comet服务器流事件的实时响应
- ServiceStack JavaScript服务器事件客户端的身份验证
- 服务器用Ruby Grape发送事件
- 销毁窗口.重新调整事件订阅服务器的大小
- 通过将文件上载到ASHX处理程序来实现SSE(服务器发送的事件)
- 从javascript强制服务器TextChanged事件
- 用于显示服务器端验证结果的 Jqgrid 编辑事件
- asp.net javascript .click() 事件未在服务器上触发
- 服务器触发客户端事件两次
- 套接字发生了什么's在'断开连接'事件(服务器端)
- jQueryUI FullCalendar:从服务器获取事件后的回调
- 从服务器获取数据,并使用$http服务和ng-change在select事件上显示它
- 如何在 Elm 中捕获服务器事件
- 在 HTML 按钮单击上的 JavaScript 函数之后调用服务器事件