Spaces:
Sleeping
Sleeping
File size: 4,225 Bytes
cfbbc1a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 | "use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.SseStream = void 0;
const shared_utils_1 = require("@nestjs/common/utils/shared.utils");
const stream_1 = require("stream");
function toDataString(data) {
if ((0, shared_utils_1.isObject)(data)) {
return toDataString(JSON.stringify(data));
}
return data
.split(/\r\n|\r|\n/)
.map(line => `data: ${line}\n`)
.join('');
}
/**
* Adapted from https://raw.githubusercontent.com/EventSource/node-ssestream
* Transforms "messages" to W3C event stream content.
* See https://html.spec.whatwg.org/multipage/server-sent-events.html
* A message is an object with one or more of the following properties:
* - data (String or object, which gets turned into JSON)
* - type
* - id
* - retry
*
* If constructed with a HTTP Request, it will optimise the socket for streaming.
* If this stream is piped to an HTTP Response, it will set appropriate headers.
*/
class SseStream extends stream_1.Transform {
constructor(req) {
super({ objectMode: true });
this.lastEventId = null;
this._headersCommitted = false;
this._destination = null;
this._statusCode = 200;
if (req && req.socket) {
req.socket.setKeepAlive(true);
req.socket.setNoDelay(true);
req.socket.setTimeout(0);
}
}
get headersCommitted() {
return this._headersCommitted;
}
pipe(destination, options) {
this._destination = destination;
this._statusCode = options?.statusCode ?? 200;
this._additionalHeaders = options?.additionalHeaders;
return super.pipe(destination, options);
}
/**
* Writes SSE headers to the destination if they have not been sent yet.
* Headers are deferred until the first message so that, if the observable
* errors before any data is emitted, the HTTP status code can still be
* changed by an exception filter.
*/
commitHeaders() {
if (this._headersCommitted || !this._destination) {
return;
}
if (this._destination.writableEnded) {
return;
}
this._headersCommitted = true;
const statusCode = this._statusCode ?? 200;
const additionalHeaders = this._additionalHeaders;
if (this._destination.writeHead) {
this._destination.writeHead(statusCode, {
...additionalHeaders,
// See https://github.com/dunglas/mercure/blob/master/hub/subscribe.go#L124-L130
'Content-Type': 'text/event-stream',
Connection: 'keep-alive',
// Disable cache, even for old browsers and proxies
'Cache-Control': 'private, no-cache, no-store, must-revalidate, max-age=0, no-transform',
Pragma: 'no-cache',
Expire: '0',
// NGINX support https://www.nginx.com/resources/wiki/start/topics/examples/x-accel/#x-accel-buffering
'X-Accel-Buffering': 'no',
});
this._destination.flushHeaders?.();
}
this._destination.write('\n');
}
_transform(message, encoding, callback) {
this.commitHeaders();
const sanitize = (val) => String(val).replace(/[\r\n]/g, '');
let data = message.type ? `event: ${sanitize(message.type)}\n` : '';
data +=
message.id !== undefined && message.id !== null
? `id: ${sanitize(message.id)}\n`
: '';
data += message.retry ? `retry: ${sanitize(message.retry)}\n` : '';
data += message.data ? toDataString(message.data) : '';
data += '\n';
this.push(data);
callback();
}
/**
* Calls `.write` but handles the drain if needed
*/
writeMessage(message, cb) {
if (message.id === undefined || message.id === null) {
this.lastEventId++;
message.id = this.lastEventId.toString();
}
if (!this.write(message, 'utf-8')) {
this.once('drain', cb);
}
else {
process.nextTick(cb);
}
}
}
exports.SseStream = SseStream;
|