Repository: websockets/ws (js)

Issue:
Extend bufferedAmount property
The bufferedAmount property is good for your application to control how much messages you can send (to avoid memory overhead), but it works only if you disable perMessageDeflate.

I propose to extend the property by calculating _sender.queue size as well.

Example:

    get bufferedAmount () {
        var amount = 0;
        var senderQueueSize = 0;
        
        if (this._socket) amount = this._socket.bufferSize || 0;
        
        if (this._sender && this._sender.queue) {
          for (i = 0; i < this._sender.queue.length; i += 1) {
            senderQueueSize += this._sender.queue[i].data.length;
          }
        }
        return amount + senderQueueSize;
    }

The best solution is to avoid cycle at all by adding queueSize variable to _sender and add/subtract length when adding a new element to queue.

    get bufferedAmount () {
        var amount = 0;
        var senderQueueSize = 0;
        
        if (this._socket) amount = this._socket.bufferSize || 0;
        return amount + this._sender.queueSize;
    }

Pull request description:
Take into account the data queued in the sender
This makes the `bufferedAmount` getter take into account the data queued in the sender.

Fixes #970.

cc: @alexions

Patch:
diff --git a/README.md b/README.md
index 15f4ab00..b30735d5 100644
--- a/README.md
+++ b/README.md
@@ -1,12 +1,12 @@
-# ws: a node.js websocket library
+# ws: a node.js WebSocket library
 
 [![Version npm](https://img.shields.io/npm/v/ws.svg)](https://www.npmjs.com/package/ws)
 [![Linux Build](https://img.shields.io/travis/websockets/ws/master.svg)](https://travis-ci.org/websockets/ws)
 [![Windows Build](https://ci.appveyor.com/api/projects/status/github/websockets/ws?branch=master&svg=true)](https://ci.appveyor.com/project/lpinca/ws)
 [![Coverage Status](https://img.shields.io/coveralls/websockets/ws/master.svg)](https://coveralls.io/r/websockets/ws?branch=master)
 
-`ws` is a simple to use WebSocket implementation, up-to-date against RFC-6455,
-and [probably the fastest WebSocket library for node.js][archive].
+`ws` is a simple to use, blazing fast, and thoroughly tested WebSocket client
+and server implementation.
 
 Passes the quite extensive Autobahn test suite. See http://websockets.github.com/ws
 for the full reports.
@@ -16,7 +16,7 @@ for the full reports.
 * **HyBi drafts 07-12** (Use the option `protocolVersion: 8`)
 * **HyBi drafts 13-17** (Current default, alternatively option `protocolVersion: 13`)
 
-### Installing
+## Installing
 
 ```
 npm install --save ws
@@ -39,6 +39,46 @@ compiler is installed on the host system.
   validation. But if you want to be 100% spec-conforming and have fast
   validation of UTF-8 then this module is a must.
 
+## API Docs
+
+See [`/doc/ws.md`](https://github.com/websockets/ws/blob/master/doc/ws.md)
+for Node.js-like docs for the ws classes.
+
+## WebSocket compression
+
+`ws` supports the [permessage-deflate extension][permessage-deflate] extension
+which enables the client and server to negotiate a compression algorithm and
+its parameters, and then selectively apply it to the data payloads of each
+WebSocket message.
+
+The extension is enabled by default but adds a significant overhead in terms of
+performance and memory comsumption. We suggest to use WebSocket compression
+only if it is really needed.
+
+To disable the extension you can set the `perMessageDeflate` option to `false`.
+On the server:
+
+```js
+const WebSocket = require('ws');
+
+const wss = new WebSocket.Server({
+  perMessageDeflate: false,
+  port: 8080
+});
+```
+
+On the client:
+
+```js
+const WebSocket = require('ws');
+
+const ws = new WebSocket('ws://www.host.com/path', {
+  perMessageDeflate: false
+});
+```
+
+## Usage examples
+
 ### Sending and receiving text data
 
 ```js
@@ -90,41 +130,7 @@ wss.on('connection', function connection(ws) {
 });
 ```
 
-### ExpressJS example
-
-```js
-const express = require('express');
-const http = require('http');
-const url = require('url');
-const WebSocket = require('ws');
-
-const app = express();
-
-app.use(function (req, res) {
-  res.send({ msg: "hello" });
-});
-
-const server = http.createServer(app);
-const wss = new WebSocket.Server({ server });
-
-wss.on('connection', function connection(ws) {
-  const location = url.parse(ws.upgradeReq.url, true);
-  // You might use location.query.access_token to authenticate or share sessions
-  // or ws.upgradeReq.headers.cookie (see http://stackoverflow.com/a/16395220/151312)
-
-  ws.on('message', function incoming(message) {
-    console.log('received: %s', message);
-  });
-
-  ws.send('something');
-});
-
-server.listen(8080, function listening() {
-  console.log('Listening on %d', server.address().port);
-});
-```
-
-### Server sending broadcast data
+### Broadcast example
 
 ```js
 const WebSocket = require('ws');
@@ -152,25 +158,38 @@ wss.on('connection', function connection(ws) {
 });
 ```
 
-### Error handling best practices
+### ExpressJS example
 
 ```js
-// If the WebSocket is closed before the following send is attempted
-ws.send('something');
+const express = require('express');
+const http = require('http');
+const url = require('url');
+const WebSocket = require('ws');
 
-// Errors (both immediate and async write errors) can be detected in an optional
-// callback. The callback is also the only way of being notified that data has
-// actually been sent.
-ws.send('something', function ack(error) {
-  // If error is not defined, the send has been completed, otherwise the error
-  // object will indicate what failed.
+const app = express();
+
+app.use(function (req, res) {
+  res.send({ msg: "hello" });
 });
 
-// Immediate errors can also be handled with `try...catch`, but **note** that
-// since sends are inherently asynchronous, socket write failures will *not* be
-// captured when this technique is used.
-try { ws.send('something'); }
-catch (e) { /* handle error */ }
+const server = http.createServer(app);
+const wss = new WebSocket.Server({ server });
+
+wss.on('connection', function connection(ws) {
+  const location = url.parse(ws.upgradeReq.url, true);
+  // You might use location.query.access_token to authenticate or share sessions
+  // or ws.upgradeReq.headers.cookie (see http://stackoverflow.com/a/16395220/151312)
+
+  ws.on('message', function incoming(message) {
+    console.log('received: %s', message);
+  });
+
+  ws.send('something');
+});
+
+server.listen(8080, function listening() {
+  console.log('Listening on %d', server.address().port);
+});
 ```
 
 ### echo.websocket.org demo
@@ -207,10 +226,26 @@ examples folder.
 
 Otherwise, see the test cases.
 
-## API Docs
+## Error handling best practices
 
-See [`/doc/ws.md`](https://github.com/websockets/ws/blob/master/doc/ws.md)
-for Node.js-like docs for the ws classes.
+```js
+// If the WebSocket is closed before the following send is attempted
+ws.send('something');
+
+// Errors (both immediate and async write errors) can be detected in an optional
+// callback. The callback is also the only way of being notified that data has
+// actually been sent.
+ws.send('something', function ack(error) {
+  // If error is not defined, the send has been completed, otherwise the error
+  // object will indicate what failed.
+});
+
+// Immediate errors can also be handled with `try...catch`, but **note** that
+// since sends are inherently asynchronous, socket write failures will *not* be
+// captured when this technique is used.
+try { ws.send('something'); }
+catch (e) { /* handle error */ }
+```
 
 ## Changelog
 
@@ -221,4 +256,4 @@ for changelog entries.
 
 [MIT](LICENSE)
 
-[archive]: http://web.archive.org/web/20130314230536/http://hobbycoding.posterous.com/the-fastest-websocket-module-for-nodejs
+[permessage-deflate]: https://tools.ietf.org/html/rfc7692
diff --git a/lib/Sender.js b/lib/Sender.js
index 74f6e85d..3913a813 100644
--- a/lib/Sender.js
+++ b/lib/Sender.js
@@ -24,12 +24,16 @@ class Sender {
    */
   constructor (socket, extensions) {
     this.perMessageDeflate = (extensions || {})[PerMessageDeflate.extensionName];
+    this._socket = socket;
+
     this.firstFragment = true;
-    this.processing = false;
     this.compress = false;
-    this._socket = socket;
-    this.onerror = null;
+
+    this.processing = false;
+    this.bufferedBytes = 0;
     this.queue = [];
+
+    this.onerror = null;
   }
 
   /**
@@ -86,10 +90,23 @@ class Sender {
    * @public
    */
   ping (data, mask) {
+    var readOnly = true;
+
+    if (data && !Buffer.isBuffer(data)) {
+      if (data instanceof ArrayBuffer) {
+        data = Buffer.from(data);
+      } else if (ArrayBuffer.isView(data)) {
+        data = viewToBuffer(data);
+      } else {
+        data = Buffer.from(data);
+        readOnly = false;
+      }
+    }
+
     if (this.perMessageDeflate) {
-      this.enqueue([this.doPing, data, mask]);
+      this.enqueue([this.doPing, data, mask, readOnly]);
     } else {
-      this.doPing(data, mask);
+      this.doPing(data, mask, readOnly);
     }
   }
 
@@ -98,14 +115,15 @@ class Sender {
    *
    * @param {*} data The message to send
    * @param {Boolean} mask Specifies whether or not to mask `data`
+   * @param {Boolean} readOnly Specifies whether `data` can be modified
    * @private
    */
-  doPing (data, mask) {
+  doPing (data, mask, readOnly) {
     this.frameAndSend(data, {
-      readOnly: true,
       opcode: 0x09,
       rsv1: false,
       fin: true,
+      readOnly,
       mask
     });
 
@@ -120,10 +138,23 @@ class Sender {
    * @public
    */
   pong (data, mask) {
+    var readOnly = true;
+
+    if (data && !Buffer.isBuffer(data)) {
+      if (data instanceof ArrayBuffer) {
+        data = Buffer.from(data);
+      } else if (ArrayBuffer.isView(data)) {
+        data = viewToBuffer(data);
+      } else {
+        data = Buffer.from(data);
+        readOnly = false;
+      }
+    }
+
     if (this.perMessageDeflate) {
-      this.enqueue([this.doPong, data, mask]);
+      this.enqueue([this.doPong, data, mask, readOnly]);
     } else {
-      this.doPong(data, mask);
+      this.doPong(data, mask, readOnly);
     }
   }
 
@@ -132,14 +163,15 @@ class Sender {
    *
    * @param {*} data The message to send
    * @param {Boolean} mask Specifies whether or not to mask `data`
+   * @param {Boolean} readOnly Specifies whether `data` can be modified
    * @private
    */
-  doPong (data, mask) {
+  doPong (data, mask, readOnly) {
     this.frameAndSend(data, {
-      readOnly: true,
       opcode: 0x0a,
       rsv1: false,
       fin: true,
+      readOnly,
       mask
     });
 
@@ -243,7 +275,7 @@ class Sender {
   /**
    * Frames and sends a piece of data according to the HyBi WebSocket protocol.
    *
-   * @param {*} data The data to send
+   * @param {Buffer} data The data to send
    * @param {Object} options Options object
    * @param {Number} options.opcode The opcode
    * @param {Boolean} options.readOnly Specifies whether `data` can be modified
@@ -267,17 +299,6 @@ class Sender {
       return;
     }
 
-    if (!Buffer.isBuffer(data)) {
-      if (data instanceof ArrayBuffer) {
-        data = Buffer.from(data);
-      } else if (ArrayBuffer.isView(data)) {
-        data = viewToBuffer(data);
-      } else {
-        data = Buffer.from(data);
-        options.readOnly = false;
-      }
-    }
-
     const mergeBuffers = data.length < 1024 || options.mask && options.readOnly;
     var dataOffset = options.mask ? 6 : 2;
     var payloadLength = data.length;
@@ -334,12 +355,13 @@ class Sender {
   dequeue () {
     if (this.processing) return;
 
-    const handler = this.queue.shift();
-    if (!handler) return;
+    const params = this.queue.shift();
+    if (!params) return;
 
+    if (params[1]) this.bufferedBytes -= params[1].length;
     this.processing = true;
 
-    handler[0].apply(this, handler.slice(1));
+    params[0].apply(this, params.slice(1));
   }
 
   /**
@@ -361,6 +383,7 @@ class Sender {
    * @private
    */
   enqueue (params) {
+    if (params[1]) this.bufferedBytes += params[1].length;
     this.queue.push(params);
     this.dequeue();
   }
diff --git a/lib/WebSocket.js b/lib/WebSocket.js
index ea351355..ec7490bd 100644
--- a/lib/WebSocket.js
+++ b/lib/WebSocket.js
@@ -82,7 +82,9 @@ class WebSocket extends EventEmitter {
   get bufferedAmount () {
     var amount = 0;
 
-    if (this._socket) amount = this._socket.bufferSize || 0;
+    if (this._socket) {
+      amount = this._socket.bufferSize + this._sender.bufferedBytes;
+    }
     return amount;
   }
 
diff --git a/package.json b/package.json
index 628cd215..597951a6 100644
--- a/package.json
+++ b/package.json
@@ -30,7 +30,7 @@
   "devDependencies": {
     "benchmark": "~2.1.2",
     "bufferutil": "~1.3.0",
-    "eslint": "~3.13.0",
+    "eslint": "~3.14.0",
     "eslint-config-semistandard": "~7.0.0",
     "eslint-config-standard": "~6.2.1",
     "eslint-plugin-promise": "~3.4.0",


Test patch:
diff --git a/test/Sender.test.js b/test/Sender.test.js
index 48c12696..d6454b47 100644
--- a/test/Sender.test.js
+++ b/test/Sender.test.js
@@ -73,6 +73,24 @@ describe('Sender', function () {
     });
   });
 
+  describe('#pong', function () {
+    it('works with multiple types of data', function (done) {
+      let count = 0;
+      const sender = new Sender({
+        write: (data) => {
+          assert.ok(data.equals(Buffer.from([0x8a, 0x02, 0x68, 0x69])));
+          if (++count === 3) done();
+        }
+      });
+
+      const array = new Uint8Array([0x68, 0x69]);
+
+      sender.pong(array.buffer, false);
+      sender.pong(array, false);
+      sender.pong('hi', false);
+    });
+  });
+
   describe('#send', function () {
     it('compresses data if compress option is enabled', function (done) {
       const perMessageDeflate = new PerMessageDeflate({ threshold: 0 });
diff --git a/test/WebSocket.test.js b/test/WebSocket.test.js
index 7ac77dd9..13b9d4bb 100644
--- a/test/WebSocket.test.js
+++ b/test/WebSocket.test.js
@@ -131,19 +131,34 @@ describe('WebSocket', function () {
       });
 
       it('defaults to zero upon "open"', function (done) {
-        server.createServer(++port, (srv) => {
+        const wss = new WebSocketServer({ port: ++port }, () => {
           const ws = new WebSocket(`ws://localhost:${port}`);
 
           ws.onopen = () => {
             assert.strictEqual(ws.bufferedAmount, 0);
-
-            ws.on('close', () => srv.close(done));
-            ws.close();
+            wss.close(done);
           };
         });
       });
 
-      it('stress kernel write buffer', function (done) {
+      it('takes into account the data in the sender queue', function (done) {
+        const wss = new WebSocketServer({ port: ++port }, () => {
+          const ws = new WebSocket(`ws://localhost:${port}`);
+
+          ws.on('open', () => {
+            ws.send('foo');
+            ws.send('bar', (err) => {
+              assert.ifError(err);
+              assert.strictEqual(ws.bufferedAmount, 0);
+              wss.close(done);
+            });
+
+            assert.strictEqual(ws.bufferedAmount, 3);
+          });
+        });
+      });
+
+      it('takes into account the data in the socket queue', function (done) {
         const wss = new WebSocketServer({ port: ++port }, () => {
           const ws = new WebSocket(`ws://localhost:${port}`, {
             perMessageDeflate: false
@@ -152,7 +167,10 @@ describe('WebSocket', function () {
 
         wss.on('connection', (ws) => {
           while (true) {
-            if (ws.bufferedAmount > 0) break;
+            if (ws._socket.bufferSize > 0) {
+              assert.strictEqual(ws.bufferedAmount, ws._socket.bufferSize);
+              break;
+            }
             ws.send('hello'.repeat(1e4));
           }
           wss.close(done);