through -> streams2 migrate - final

lib/index.js

@@ -164,17 +164,18 @@ module.exports = function(config_hash) {
 		var complete = false;
 		req.on('end', function() {
 			complete = true;
+			stream.done();
 		});
 		req.on('close', function() {
 			if (!complete) {
-				stream.emit('abort');
+				stream.abort();
 			}
 		});
 
 		stream.on('error', function(err) {
 			return next(err);
 		});
-		stream.on('end', function() {
+		stream.on('success', function() {
 			res.status(201);
 			return res.send({
 				ok: 'tarball uploaded successfully'

lib/local-fs.js

@@ -47,13 +47,15 @@ function write_stream(name) {
 		var file = fs.createWriteStream(tmpname);
 		stream.pipe(file);
 		stream.done = function() {
-			file.on('close', function() {
+			stream.on('end', function() {
-				fs.rename(tmpname, name, function(err) {
+				file.on('close', function() {
-					if (err) stream.emit('error', err);
+					fs.rename(tmpname, name, function(err) {
-					stream.emit('close');
+						if (err) stream.emit('error', err);
+						stream.emit('success');
+					});
 				});
+				file.destroySoon();
 			});
-			file.destroySoon();
 		};
 		stream.abort = function() {
 			file.on('close', function() {

lib/local-storage.js

@@ -1,7 +1,6 @@
 var fs = require('fs');
 var semver = require('semver');
 var Path = require('path');
-var through = require('through');
 var fs_storage = require('./local-fs');
 var UError = require('./error').UserError;
 var utils = require('./utils');
@@ -136,14 +135,10 @@ Storage.prototype.add_version = function(name, version, metadata, tag, callback)
 }
 
 Storage.prototype.add_tarball = function(name, filename) {
-	var stream = through(function(data) {
+	var stream = new mystreams.UploadTarballStream();
-		wstream.write(data);
-	}, function() {
-		wstream.end();
-	});
 	
 	var self = this;
-	if (name === info_file) {
+	if (name === info_file || name === '__proto__') {
 		stream.emit('error', new UError({
 			status: 403,
 			msg: 'can\'t use this filename'
@@ -167,12 +162,17 @@ Storage.prototype.add_tarball = function(name, filename) {
 		// re-emitting open because it's handled in storage.js
 		stream.emit('open');
 	});
-	wstream.on('close', function() {
+	wstream.on('success', function() {
-		stream.emit('end');
+		// re-emitting open because it's handled in index.js
-	});
+		stream.emit('success');
-	stream.on('abort', function() {
-		wstream.emit('abort');
 	});
+	stream.abort = function() {
+		wstream.abort();
+	};
+	stream.done = function() {
+		wstream.done();
+	};
+	stream.pipe(wstream);
 
 	return stream;
 }

lib/storage.js

@@ -1,6 +1,5 @@
 var async = require('async');
 var semver = require('semver');
-var through = require('through');
 var UError = require('./error').UserError;
 var Local = require('./local-storage');
 var Proxy = require('./up-storage');
@@ -144,13 +143,10 @@ Storage.prototype.get_tarball = function(name, filename) {
 
 			var savestream = self.local.add_tarball(name, filename);
 			savestream.on('error', function(err) {
-console.log('xx!', name, filename);
 				savestream.abort();
 				stream.emit('error', err);
 			});
-console.log('!', name, filename);
 			savestream.on('open', function() {
-console.log('x!', name, filename);
 				var rstream2 = uplink.get_url(file.url);
 				rstream2.on('error', function(err) {
 					stream.emit('error', err);

lib/streams.js

@@ -18,7 +18,7 @@ module.exports.ReadTarballStream = ReadTarball;
 // This stream is used to upload tarballs to a repository
 //
 function UploadTarball(options) {
-	stream.Writable.call(this, options);
+	stream.PassThrough.call(this, options);
 	
 	// called when user closes connection before upload finishes
 	add_abstract_method(this, 'abort');
@@ -27,7 +27,7 @@ function UploadTarball(options) {
 	add_abstract_method(this, 'done');
 }
 
-util.inherits(UploadTarball, stream.Writable);
+util.inherits(UploadTarball, stream.PassThrough);
 module.exports.UploadTarballStream = UploadTarball;
 
 //
@@ -51,7 +51,7 @@ function add_abstract_method(self, name) {
 	});
 }
 
-module.exports.__test = function() {
+function __test() {
 	var test = new ReadTarball();
 	test.abort();
 	setTimeout(function() {

lib/up-storage.js

@@ -1,6 +1,5 @@
 var URL = require('url');
 var request = require('request');
-var through = require('through');
 var UError = require('./error').UserError;
 var mystreams = require('./streams');
 

package.yaml

@@ -25,7 +25,6 @@ dependencies:
   async: '*'
   semver: '*'
   minimatch: '*'
-  through: '*'
 
 devDependencies:
   rimraf: '*'

test/lib/server.js

@@ -15,16 +15,14 @@ function prep(cb) {
 }
 
 Server.prototype.request = function(options, cb) {
-	options.headers = options.headers || {};
+	var headers = options.headers || {};
+	headers.accept = headers.accept || 'application/json';
+	headers['user-agent'] = headers['user-agent'] || this.userAgent;
+	headers.authorization = headers.authorization || this.auth;
 	return request({
 		url: this.url + options.uri,
 		method: options.method || 'GET',
-		headers: {
+		headers: headers,
-			accept: options.headers.accept || 'application/json',
-			'user-agent': options.headers['user-agent'] || this.userAgent,
-			'content-type': options.headers['content-type'],
-			authorization: this.auth,
-		},
 		json: options.json || true,
 	}, cb);
 }
@@ -86,5 +84,24 @@ Server.prototype.put_tarball = function(name, filename, data, cb) {
 	}, prep(cb)).end(data);
 }
 
+Server.prototype.put_tarball_incomplete = function(name, filename, data, size, cb) {
+	var req = this.request({
+		uri: '/'+escape(name)+'/-/'+escape(filename)+'/whatever',
+		method: 'PUT',
+		headers: {
+			'content-type': 'application/octet-stream',
+			'content-length': size,
+		},
+		timeout: 1000,
+	}, function(err) {
+		assert(err);
+		cb();
+	});
+	req.write(data);
+	setTimeout(function() {
+		req.req.abort();
+	}, 20);
+}
+
 module.exports = Server;
 

test/tests.js

@@ -67,6 +67,12 @@ ex['downloading non-existent tarball'] = function(cb) {
 	});
 };
 
+ex['uploading incomplete tarball'] = function(cb) {
+	server.put_tarball_incomplete('testpkg', 'blahblah1', readfile('fixtures/binary'), 3000, function(res, body) {
+		cb();
+	});
+};
+
 ex['uploading new tarball'] = function(cb) {
 	server.put_tarball('testpkg', 'blahblah', readfile('fixtures/binary'), function(res, body) {
 		assert(res.statusCode === 201);