stream_base: expose `bytesRead` getter

This will provide `bytesRead` data on consumed sockets.

Fix: #3021
PR-URL: https://github.com/nodejs/node/pull/6284
Reviewed-By: Ben Noordhuis <info@bnoordhuis.nl>

5 files changed, 79 insertions, 7 deletions

diff --git a/lib/net.js b/lib/net.js
index e92ec369e5..e3b726a756 100644
--- a/lib/net.js
+++ b/lib/net.js
@@ -97,7 +97,6 @@ exports._normalizeConnectArgs = normalizeConnectArgs;
 // called when creating new Socket, or when re-using a closed Socket
 function initSocketHandle(self) {
   self.destroyed = false;
-  self.bytesRead = 0;
   self._bytesDispatched = 0;
   self._sockname = null;
 
@@ -112,6 +111,10 @@ function initSocketHandle(self) {
   }
 }
 
+
+const BYTES_READ = Symbol('bytesRead');
+
+
 function Socket(options) {
   if (!(this instanceof Socket)) return new Socket(options);
 
@@ -179,6 +182,9 @@ function Socket(options) {
   // Reserve properties
   this.server = null;
   this._server = null;
+
+  // Used after `.destroy()`
+  this[BYTES_READ] = 0;
 }
 util.inherits(Socket, stream.Duplex);
 
@@ -470,6 +476,9 @@ Socket.prototype._destroy = function(exception, cb) {
     if (this !== process.stderr)
       debug('close handle');
     var isException = exception ? true : false;
+    // `bytesRead` should be accessible after `.destroy()`
+    this[BYTES_READ] = this._handle.bytesRead;
+
     this._handle.close(() => {
       debug('emit close');
       this.emit('close', isException);
@@ -521,10 +530,6 @@ function onread(nread, buffer) {
     // will prevent this from being called again until _read() gets
     // called again.
 
-    // if it's not enough data, we'll just call handle.readStart()
-    // again right away.
-    self.bytesRead += nread;
-
     // Optimization: emit the original buffer with end points
     var ret = self.push(buffer);
 
@@ -580,6 +585,9 @@ Socket.prototype._getpeername = function() {
   return this._peername;
 };
 
+Socket.prototype.__defineGetter__('bytesRead', function() {
+  return this._handle ? this._handle.bytesRead : this[BYTES_READ];
+});
 
 Socket.prototype.__defineGetter__('remoteAddress', function() {
   return this._getpeername().address;
diff --git a/src/env.h b/src/env.h
index 9b117e1de0..afbade5dd8 100644
--- a/src/env.h
+++ b/src/env.h
@@ -72,6 +72,7 @@ namespace node {
   V(buffer_string, "buffer")                                                  \
   V(bytes_string, "bytes")                                                    \
   V(bytes_parsed_string, "bytesParsed")                                       \
+  V(bytes_read_string, "bytesRead")                                           \
   V(cached_data_string, "cachedData")                                         \
   V(cached_data_produced_string, "cachedDataProduced")                        \
   V(cached_data_rejected_string, "cachedDataRejected")                        \
diff --git a/src/stream_base-inl.h b/src/stream_base-inl.h
index 81114a265e..099e105334 100644
--- a/src/stream_base-inl.h
+++ b/src/stream_base-inl.h
@@ -43,6 +43,13 @@ void StreamBase::AddMethods(Environment* env,
                                      v8::DEFAULT,
                                      attributes);
 
+  t->InstanceTemplate()->SetAccessor(env->bytes_read_string(),
+                                     GetBytesRead<Base>,
+                                     nullptr,
+                                     env->as_external(),
+                                     v8::DEFAULT,
+                                     attributes);
+
   env->SetProtoMethod(t, "readStart", JSMethod<Base, &StreamBase::ReadStart>);
   env->SetProtoMethod(t, "readStop", JSMethod<Base, &StreamBase::ReadStop>);
   if ((flags & kFlagNoShutdown) == 0)
@@ -80,6 +87,16 @@ void StreamBase::GetFD(Local<String> key,
 
 
 template <class Base>
+void StreamBase::GetBytesRead(Local<String> key,
+                              const PropertyCallbackInfo<Value>& args) {
+  StreamBase* wrap = Unwrap<Base>(args.Holder());
+
+  // uint64_t -> double. 53bits is enough for all real cases.
+  args.GetReturnValue().Set(static_cast<double>(wrap->bytes_read_));
+}
+
+
+template <class Base>
 void StreamBase::GetExternal(Local<String> key,
                              const PropertyCallbackInfo<Value>& args) {
   StreamBase* wrap = Unwrap<Base>(args.Holder());
diff --git a/src/stream_base.h b/src/stream_base.h
index fad2ddd2e0..e722a208a8 100644
--- a/src/stream_base.h
+++ b/src/stream_base.h
@@ -136,7 +136,7 @@ class StreamResource {
                          uv_handle_type pending,
                          void* ctx);
 
-  StreamResource() {
+  StreamResource() : bytes_read_(0) {
   }
   virtual ~StreamResource() = default;
 
@@ -160,9 +160,11 @@ class StreamResource {
       alloc_cb_.fn(size, buf, alloc_cb_.ctx);
   }
 
-  inline void OnRead(size_t nread,
+  inline void OnRead(ssize_t nread,
                      const uv_buf_t* buf,
                      uv_handle_type pending = UV_UNKNOWN_HANDLE) {
+    if (nread > 0)
+      bytes_read_ += static_cast<uint64_t>(nread);
     if (!read_cb_.is_empty())
       read_cb_.fn(nread, buf, pending, read_cb_.ctx);
   }
@@ -182,6 +184,9 @@ class StreamResource {
   Callback<AfterWriteCb> after_write_cb_;
   Callback<AllocCb> alloc_cb_;
   Callback<ReadCb> read_cb_;
+  uint64_t bytes_read_;
+
+  friend class StreamBase;
 };
 
 class StreamBase : public StreamResource {
@@ -249,6 +254,10 @@ class StreamBase : public StreamResource {
   static void GetExternal(v8::Local<v8::String> key,
                           const v8::PropertyCallbackInfo<v8::Value>& args);
 
+  template <class Base>
+  static void GetBytesRead(v8::Local<v8::String> key,
+                           const v8::PropertyCallbackInfo<v8::Value>& args);
+
   template <class Base,
             int (StreamBase::*Method)(  // NOLINT(whitespace/parens)
       const v8::FunctionCallbackInfo<v8::Value>& args)>
diff --git a/test/parallel/test-net-bytes-read.js b/test/parallel/test-net-bytes-read.js
new file mode 100644
index 0000000000..ba2bc160d0
--- /dev/null
+++ b/test/parallel/test-net-bytes-read.js
@@ -0,0 +1,37 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const net = require('net');
+
+const big = Buffer.alloc(1024 * 1024);
+
+const server = net.createServer((socket) => {
+  socket.end(big);
+  server.close();
+}).listen(common.PORT, () => {
+  let prev = 0;
+
+  function checkRaise(value) {
+    assert(value > prev);
+    prev = value;
+  }
+
+  const socket = net.connect(common.PORT, () => {
+    socket.on('data', (chunk) => {
+      checkRaise(socket.bytesRead);
+    });
+
+    socket.on('end', common.mustCall(() => {
+      assert.equal(socket.bytesRead, prev);
+      assert.equal(big.length, prev);
+    }));
+
+    socket.on('close', common.mustCall(() => {
+      assert(!socket._handle);
+      assert.equal(socket.bytesRead, prev);
+      assert.equal(big.length, prev);
+    }));
+  });
+  socket.end();
+});