Adding upload progress reporting, issue dojo#285

docs/request.md

@@ -9,3 +9,39 @@ response.
 	* post
 	* delete
 	* put
+
+## Monitoring Download Progress
+
+You can monitor download progress by listening for events on the response object.
+
+```typescript
+request("http://www.example/some-large-file").then(response => {
+	response.on('progress', progressEvent => {
+		console.log(`Total downloaded: ${progressEvent.totalBytesDownloaded}`);
+	});
+});
+```
+
+## Monitoring Upload Progress
+
+You can monitor upload progress by providing an `uploadObserver` in the request options.
+
+```typescript
+const uploader = new UploadObserver();
+uploader.on('upload', uploadEvent => {
+	console.log(`Total uploaded: ${uploadEvent.totalBytesUploaded}`);
+});
+request.post('http://www.example.com/', {
+	body: someLargeString,
+	uploadObserver: uploader
+});
+```
+
+Note that while the node provider will emit a single `upload` event when it is done uploading, it cannot more granular upload events with `string` or `Buffer` body types. To receive more accurate upload events, you can use the `bodyStream` option to provide a `Readable` with the body content.
+
+```typescript
+request.post('http://www.example.com/', {
+	bodyStream: fs.createReadStream('some-large-file'),
+	uploadObserver: uploader
+});
+```

src/request/UploadObserver.ts

@@ -0,0 +1,16 @@
+import { EventedListenerOrArray } from '@dojo/interfaces/bases';
+import { Handle } from '@dojo/interfaces/core';
+import { BaseEventedEvents, Evented } from '../Evented';
+import { UploadEvent } from './interfaces';
+
+export interface UploadObserverEvents extends BaseEventedEvents {
+	(type: 'upload', handler: EventedListenerOrArray<Evented, UploadEvent>): Handle;
+}
+
+export default class UploadObserver extends Evented {
+	constructor() {
+		super();
+	}
+
+	on: UploadObserverEvents;
+}

src/request/interfaces.d.ts

@@ -4,6 +4,7 @@ import { IterableIterator } from '@dojo/shim/iterator';
 import Task from '../async/Task';
 import { BaseEventedEvents, Evented } from '../Evented';
 import UrlSearchParams, { ParamList } from '../UrlSearchParams';
+import UploadObserver from './UploadObserver';
 
 export interface Body {
 	readonly bodyUsed: boolean;
@@ -51,6 +52,11 @@ export interface StartEvent extends ResponseEvent {
 	type: 'start';
 }
 
+export interface UploadEvent extends ResponseEvent {
+	type: 'upload';
+	totalBytesUploaded: number;
+}
+
 export type Provider = (url: string, options?: RequestOptions) => Task<Response>;
 
 export type ProviderTest = (url: string, options?: RequestOptions) => boolean | null;
@@ -65,6 +71,7 @@ export interface RequestOptions {
 	timeout?: number;
 	user?: string;
 	query?: string | ParamList;
+	uploadObserver?: UploadObserver;
 }
 
 export interface ResponseEvents extends BaseEventedEvents {

src/request/providers/node.ts

@@ -13,12 +13,14 @@ import Headers from '../Headers';
 import { RequestOptions } from '../interfaces';
 import Response from '../Response';
 import TimeoutError from '../TimeoutError';
+import { Readable } from 'stream';
 
 /**
  * Request options specific to a node request
  */
 export interface NodeRequestOptions extends RequestOptions {
 	agent?: any;
+	bodyStream?: Readable;
 	ca?: any;
 	cert?: string;
 	ciphers?: string;
@@ -533,13 +535,36 @@ export default function node(url: string, options: NodeRequestOptions = {}): Tas
 
 		request.once('error', reject);
 
-		if (options.body) {
-			if (options.body instanceof Buffer) {
-				request.end(options.body.toString());
+		if (options.bodyStream) {
+			options.bodyStream.pipe(request);
+			if (options.uploadObserver) {
+				let uploadedSize = 0;
+
+				options.bodyStream.on('data', (chunk: any) => {
+					uploadedSize += chunk.length;
+					(options.uploadObserver!).emit({
+						type: 'upload',
+						totalBytesUploaded: uploadedSize
+					});
+				});
 			}
-			else {
-				request.end(options.body.toString());
+			options.bodyStream.on('end', () => {
+				request.end();
+			});
+		}
+		else if (options.body) {
+			const body = options.body.toString();
+
+			if (options.uploadObserver) {
+				request.on('response', () => {
+					(options.uploadObserver!).emit({
+						type: 'upload',
+						totalBytesUploaded: body.length
+					});
+				});
 			}
+
+			request.end(body);
 		}
 		else {
 			request.end();

src/request/providers/xhr.ts

@@ -298,6 +298,17 @@ export default function xhr(url: string, options: XhrRequestOptions = {}): Task<
 		}
 	});
 
+	if (options.uploadObserver) {
+		const observer = options.uploadObserver;
+
+		request.upload.addEventListener('progress', event => {
+			observer.emit({
+				type: 'upload',
+				totalBytesUploaded: event.loaded
+			});
+		});
+	}
+
 	request.send(options.body || null);
 
 	return task;

tests/unit/request/node.ts

@@ -7,6 +7,7 @@ import * as zlib from 'zlib';
 import { Response } from '../../../src/request/interfaces';
 import { default as nodeRequest, NodeResponse } from '../../../src/request/providers/node';
 import TimeoutError from '../../../src/request/TimeoutError';
+import UploadObserver from '../../../src/request/UploadObserver';
 
 const serverPort = 8124;
 const serverUrl = 'http://localhost:' + serverPort;
@@ -270,6 +271,15 @@ function buildRedirectTests(methods: RedirectTestData[]) {
 
 function getResponseData(request: any): DummyResponse {
 	const urlInfo = parse(request.url, true);
+
+	if (urlInfo.query.dataKey === 'echo') {
+		return {
+			body: JSON.stringify({
+				headers: request.headers
+			})
+		};
+	}
+
 	return responseData[ urlInfo.query.dataKey ] || {};
 }
 
@@ -375,6 +385,17 @@ registerSuite({
 			}
 		},
 
+		bodyStream: {
+			'stream is read'(this: any) {
+				return nodeRequest(getRequestUrl('echo'), {
+					method: 'POST',
+					bodyStream: fs.createReadStream('tests/support/data/foo.json')
+				}).then(res => res.json()).then(json => {
+					assert.deepEqual(requestData, { foo: 'bar' });
+				});
+			}
+		},
+
 		'content encoding': (function (compressionTypes) {
 			const suites: { [key: string]: any } = {};
 
@@ -523,6 +544,42 @@ registerSuite({
 						assert.strictEqual(error.name, 'TimeoutError');
 					})
 				);
+		},
+		'upload monitoriting': {
+			'with a stream'(this: any) {
+				let events: number[] = [];
+
+				const uploadMonitor = new UploadObserver();
+				uploadMonitor.on('upload', (event) => {
+					events.push(event.totalBytesUploaded);
+				});
+
+				return nodeRequest(getRequestUrl('foo.json'), {
+					method: 'POST',
+					bodyStream: fs.createReadStream('tests/support/data/foo.json'),
+					uploadObserver: uploadMonitor
+				}).then(res => {
+					assert.isTrue(events.length > 0, 'was expecting at least one monitor event');
+					assert.equal(events[events.length - 1], 17);
+				});
+			},
+			'without a stream'(this: any) {
+				let events: number[] = [];
+
+				const uploadMonitor = new UploadObserver();
+				uploadMonitor.on('upload', (event) => {
+					events.push(event.totalBytesUploaded);
+				});
+
+				return nodeRequest(getRequestUrl('foo.json'), {
+					method: 'POST',
+					body: '{ "foo": "bar" }\n',
+					uploadObserver: uploadMonitor
+				}).then(res => {
+					assert.isTrue(events.length > 0, 'was expecting at least one monitor event');
+					assert.equal(events[events.length - 1], 17);
+				});
+			}
 		}
 	},
 

tests/unit/request/xhr.ts

@@ -9,6 +9,7 @@ import UrlSearchParams from '../../../src/UrlSearchParams';
 import has from '@dojo/has/has';
 import { XhrResponse } from '../../../src/request/providers/xhr';
 import Promise from '@dojo/shim/Promise';
+import UploadObserver from '../../../src/request/UploadObserver';
 
 let echoServerAvailable = false;
 registerSuite({
@@ -118,6 +119,28 @@ registerSuite({
 			});
 		},
 
+		'upload monitoring'(this: any) {
+			if (!echoServerAvailable) {
+				this.skip('No echo server available');
+			}
+
+			let events: number[] = [];
+
+			const uploadMonitor = new UploadObserver();
+			uploadMonitor.on('upload', (event) => {
+				events.push(event.totalBytesUploaded);
+			});
+
+			return xhrRequest('/__echo/post', {
+				method: 'POST',
+				body: '12345',
+				uploadObserver: uploadMonitor
+			}).then(res => {
+				assert.isTrue(events.length > 0, 'was expecting at least one monitor event');
+				assert.equal(events[events.length - 1], 5);
+			});
+		},
+
 		'query': {
 			'.get with query URL and query option string'(this: any) {
 				if (!echoServerAvailable) {