feat: enable bytes read tracking

src/file.ts

@@ -35,7 +35,7 @@ import * as once from 'onetime';
 import * as os from 'os';
 const pumpify = require('pumpify');
 import * as resumableUpload from 'gcs-resumable-upload';
-import {Duplex, Writable, Readable} from 'stream';
+import {Duplex, Writable, Readable, Transform} from 'stream';
 import * as streamEvents from 'stream-events';
 import * as through from 'through2';
 import * as xdgBasedir from 'xdg-basedir';
@@ -198,6 +198,8 @@ export interface CreateWriteStreamOptions extends CreateResumableUploadOptions {
   gzip?: string | boolean;
   resumable?: boolean;
   validation?: string | boolean;
+  // tslint:disable-next-line:no-any
+  onUploadProgress?: (progressEvent: any) => void;
 }
 
 export interface MakeFilePrivateOptions {
@@ -1752,12 +1754,20 @@ class File extends ServiceObject<File> {
       md5,
     });
 
+    const progressStream = new ProgressStream();
+    if (options.onUploadProgress) {
+      progressStream.on('progress', bytesRead => {
+        options.onUploadProgress!({bytesRead});
+      });
+    }
+
     const fileWriteStream = duplexify();
 
     const stream = streamEvents(
       pumpify([
         gzip ? zlib.createGzip() : through(),
         validateStream,
+        progressStream,
         fileWriteStream,
       ])
     ) as Duplex;
@@ -3489,6 +3499,25 @@ class File extends ServiceObject<File> {
   }
 }
 
+/**
+ * Basic Passthrough Stream that records the number of bytes read
+ * every time the cursor is moved.
+ */
+class ProgressStream extends Transform {
+  bytesRead: number;
+  constructor() {
+    super(...arguments);
+    this.bytesRead = 0;
+  }
+  // tslint:disable-next-line: no-any
+  _transform(chunk: any, encoding: string, callback: Function) {
+    this.bytesRead += chunk.length;
+    this.emit('progress', this.bytesRead);
+    this.push(chunk);
+    callback();
+  }
+}
+
 /*! Developer Documentation
  *
  * All async methods (except for streams) will return a Promise in the event

system-test/storage.ts

@@ -93,6 +93,7 @@ import {
   Iam,
 } from '../src';
 import * as nock from 'nock';
+import * as readline from 'readline';
 
 interface ErrorCallbackFunction {
   (err: Error | null): void;
@@ -2791,6 +2792,21 @@ describe('storage', () => {
     });
   });
 
+  describe('bucket upload with progress', () => {
+    it('show bytes sent', async () => {
+      const fileSize = fs.statSync(FILES.big.path).size;
+      await bucket.upload(FILES.big.path, {
+        onUploadProgress: evt => {
+          const progress = (evt.bytesRead / fileSize) * 100;
+
+          readline.clearLine(process.stdout, 0);
+          readline.cursorTo(process.stdout, 0, 0);
+          process.stdout.write(`${Math.round(progress)}% complete`);
+        },
+      });
+    });
+  });
+
   describe('channels', () => {
     it('should create a channel', done => {
       const config = {