index.js

var Readable = require('stream').Readable
var nodeFetch		= (...args) => import('node-fetch').then(({default: fetch}) => fetch(...args));
var EventEmitter	= require('events').EventEmitter;
var util		= require('util');

function resumableUpload() {
	this.byteCount	= 0; //init variables
	this.tokens	= {};
	this.filepath	= '';
	this.retry	= -1;
	this.host	= 'www.googleapis.com';
	this.metadata = {};
};

util.inherits(resumableUpload, EventEmitter);

//Init the upload by POSTing google for an upload URL (saved to self.location)
resumableUpload.prototype.upload = async function() {
	var self = this;
	var options = {
		url:	'https://' + self.host + self.api + '?uploadType=resumable',
		headers: {
		  'Host':			self.host,
		  'Authorization':		'Bearer ' + self.tokens.access_token,
		  'Content-Length':		JSON.stringify(self.metadata).length,
		  'Content-Type':		'application/json',
		  'X-Upload-Content-Length':	self.content.length,
		  'X-Upload-Content-Type': 	'message/rfc822'
		},
		body: JSON.stringify(self.metadata)
	};
	//Send request and start upload if success
	const callback = function(err, res, body) {
		if (err) {
			self.emit('error', err instanceof Error ? err : new Error(JSON.stringify(err)));
			self.emit('progress', 'Retrying ...');
			if ((self.retry > 0) || (self.retry <= -1)) {
				self.retry--;
				self.upload(); // retry
			} else {
				return;
			}
		}
		self.location = res.headers.get('location');
		self.send();
	}
	options.method = self.method;
	try {
		const result = await nodeFetch(options.url, options);
		if (result.status !== 200) {
			callback({status: result.status, statusText: result.statusText}, null);
		} else {
			callback(null, result, await result.text());
		}
	} catch (error) {
		callback(error)
	}
}

//Pipes uploadPipe to self.location (Google's Location header)
resumableUpload.prototype.send = async function() {
	var self = this;
	var options = {
		url: self.location, //self.location becomes the Google-provided URL to PUT to
		headers: {
		  'Authorization':	'Bearer ' + self.tokens.access_token,
		  'Content-Length':	self.content.length - self.byteCount,
		  'Content-Type':	'message/rfc822'
		}
	};
	try {
		//creates file stream, pipes it to self.location
		var uploadPipe = new Readable
		uploadPipe.push(self.content)    // the string you want
		uploadPipe.push(null)
	} catch (e) {
		self.emit('error', new Error(e));
		return;
	}
	var health = setInterval(function(){
		self.getProgress(function(err, res, body) {
			if (!err && typeof res.headers.range !== 'undefined') {
				self.emit('progress', res.headers.range.substring(8));
			}
		});
	}, 5000);
	options.body = uploadPipe;
	options.method = self.method;
	const callback = function(error, response, body) {
		clearInterval(health);
		if (!error) {
			self.emit('success', body);
			return;
		}
		self.emit('error', error instanceof Error ? error : new Error(JSON.stringify(error)));
		if ((self.retry > 0) || (self.retry <= -1)) {
			self.retry--;
			self.getProgress(function(err, res, b) {
				if (typeof res.headers.range !== 'undefined') {
					self.byteCount = res.headers.range.substring(8); //parse response
				} else {
					self.byteCount = 0;
				}
				self.send();
			});
		}
	};
	try {
		const result = await nodeFetch(options.url, options)
		if (result.status !== 200) {
			callback({status: result.status, statusText: result.statusText}, null);
		} else {
			callback(null, result, await result.text());
		}
	} catch (error) {
		callback(error);
	}
}

resumableUpload.prototype.getProgress = async function(handler) {
	var self = this;
	var options = {
		url: self.location,
		headers: {
		  'Authorization':	'Bearer ' + self.tokens.access_token,
		  'Content-Length':	0,
		  'Content-Range':	'bytes */' + self.content.length
		},
		method: 'PUT'
	};
	const response = await nodeFetch(options.url, options);
	handler(response.json())
}

module.exports = resumableUpload;