[fix] Do not swallow errors emitted while destroying (#1672)

Refs: nodejs/node#29197

Author: lpinca

lib/stream.js

@@ -78,6 +78,8 @@ function createWebSocketStream(ws, options) {
  });
 
  ws.once('error', function error(err) {
+ if (duplex.destroyed) return;
+
  duplex.destroy(err);
  });
 
@@ -94,8 +96,15 @@ function createWebSocketStream(ws, options) {
  return;
  }
 
- ws.once('close', function close() {
+ let called = false;
+
+ ws.once('error', function error(err) {
+ called = true;
  callback(err);
+ });
+
+ ws.once('close', function close() {
+ if (!called) callback(err);
  process.nextTick(emitClose, duplex);
  });
  ws.terminate();

test/create-websocket-stream.test.js

@@ -225,6 +225,52 @@ describe('createWebSocketStream', () => {
  });
  });
 
+ it('does not swallow errors that may occur while destroying', (done) => {
+ const frame = Buffer.concat(
+ Sender.frame(Buffer.from([0x22, 0xfa, 0xec, 0x78]), {
+ fin: true,
+ rsv1: true,
+ opcode: 0x02,
+ mask: false,
+ readOnly: false
+ })
+ );
+
+ const wss = new WebSocket.Server(
+ {
+ perMessageDeflate: true,
+ port: 0
+ },
+ () => {
+ const ws = new WebSocket(`ws://localhost:${wss.address().port}`);
+ const duplex = createWebSocketStream(ws);
+
+ duplex.on('error', (err) => {
+ assert.ok(err instanceof Error);
+ assert.strictEqual(err.code, 'Z_DATA_ERROR');
+ assert.strictEqual(err.errno, -3);
+
+ duplex.on('close', () => {
+ wss.close(done);
+ });
+ });
+
+ let bytesRead = 0;
+
+ ws.on('open', () => {
+ ws._socket.on('data', (chunk) => {
+ bytesRead += chunk.length;
+ if (bytesRead === frame.length) duplex.destroy();
+ });
+ });
+ }
+ );
+
+ wss.on('connection', (ws) => {
+ ws._socket.write(frame);
+ });
+ });
+
  it("does not suppress the throwing behavior of 'error' events", (done) => {
  const wss = new WebSocket.Server({ port: 0 }, () => {
  const ws = new WebSocket(`ws://localhost:${wss.address().port}`);

test/permessage-deflate.test.js

@@ -589,6 +589,7 @@ describe('PerMessageDeflate', () => {
  perMessageDeflate.accept([{}]);
  perMessageDeflate.decompress(data, true, (err) => {
  assert.ok(err instanceof Error);
+ assert.strictEqual(err.code, 'Z_DATA_ERROR');
  assert.strictEqual(err.errno, -3);
  done();
  });