Emit errors from upstream on file upload

lib/storage/file.js

@@ -988,6 +988,10 @@ File.prototype.startResumableUpload_ = function(stream, metadata) {
     var writeStream = request(reqOpts);
     writeStream.callback = util.noop;
 
+    writeStream.on('error', function(err) {
+      handleError(err);
+    });
+
     writeStream.on('complete', function(res) {
       util.handleResp(null, res, res.body, function(err, data) {
         if (err) {
@@ -1056,7 +1060,6 @@ File.prototype.startResumableUpload_ = function(stream, metadata) {
     }
 
     stream.emit('error', err);
-    stream.end();
   }
 };
 

test/storage/file.js

@@ -528,6 +528,44 @@ describe('File', function() {
       writable.write('data');
     });
 
+    it('should re-emit errors', function(done) {
+      var error = new Error('Error.');
+      var requestCount = 0;
+      file.bucket.storage.makeAuthorizedRequest_ = function(reqOpts, cb) {
+        requestCount++;
+
+        // respond to creation POST.
+        if (requestCount === 1) {
+          cb(null, null, { headers: { location: 'http://resume' }});
+          return;
+        }
+
+        // create an authorized request for the first PUT.
+        if (requestCount === 2) {
+          cb.onAuthorized(null, { headers: {} });
+        }
+      };
+
+      // respond to first upload PUT request.
+      request_Override = function() {
+        var stream = through();
+        setImmediate(function() {
+          stream.emit('error', error);
+        });
+        return stream;
+      };
+
+      var stream = duplexify();
+
+      stream
+        .on('error', function(err) {
+          assert.equal(err, error);
+          done();
+        });
+
+      file.startResumableUpload_(stream);
+    });
+
     it('should start a simple upload if specified', function(done) {
       var writable = file.createWriteStream({
         metadata: METADATA,