Как использовать сервер-отправленные события в express.js

Question

Я устанавливаю свой сервер REST с помощью express.js. Теперь я хочу добавить sse на этот сервер. После того как я реализовал пакет sse this, я получаю сообщение об ошибке. Я знаю, что получаю эту ошибку, когда попытаюсь дважды использовать 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('...'); 
}); 

Кроме того, я нашел this раствор и this. Можно ли использовать sse с функцией res.write или другим способом без использования другого пакета?

Answer 1

Вы можете достичь этого без других пакетов.

Я написал сообщение в блоге об этом, part 1 изложил основы.

Answer 2

Из документации по используемой библиотеке вы должны использовать 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

Answer 3

Самореклама: Я написал пакет ExpreSSE, который обеспечивает промежуточное программные для работы с SSE в экспрессе, вы можете найти его на НОМ: @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()); 
    }); 
}); 

Существует также другой промежуточный слой, чтобы подтолкнуть то же событие для нескольких клиентов.

Answer 4

Кто-то (пользователь: https://stackoverflow.com/users/451634/benny-neugebauer) из этой статьи: addEventListener on custom object) буквально дал мне подсказку о том, как реализовать это без какой-либо другой упаковки, кроме выражения! У меня это работает!

Во-первых, EventEmitter импортировать в Node:

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, тип контента - это текст/поток событий, без кеша и для продолжения ,

Вы также будете называть метод .on вашего экземпляра EventEmitter, который принимает 2 параметра: строку события для прослушивания и функцию для обработки этого события (эта функция может принимать столько же параметров, сколько и данный)

Теперь все, что вам нужно сделать, чтобы отправить событие сервера, - это позвонить.испускают метод вашего экземпляра EventEmitter:

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({}) }); 
}); 

в someModule:

function someMethod(request, Stream) { 
    return new Promise((resolve, reject) => { 
    Stream.emit("push", "test", { data: 'some data' }); 
    return resolve(); 
    }) 
} 

Это легко! Никакой другой пакет не требуется!

Вот ссылка на EventEmitter класса узла: 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)