don't crash when queued write fails
This commit is contained in:
parent
d227084339
commit
0dcc43316f
62
lib/net.js
62
lib/net.js
@ -84,6 +84,7 @@ function initSocketHandle(self) {
|
|||||||
self._flags = 0;
|
self._flags = 0;
|
||||||
self._connectQueueSize = 0;
|
self._connectQueueSize = 0;
|
||||||
self.destroyed = false;
|
self.destroyed = false;
|
||||||
|
self.errorEmitted = false;
|
||||||
self.bytesRead = 0;
|
self.bytesRead = 0;
|
||||||
self.bytesWritten = 0;
|
self.bytesWritten = 0;
|
||||||
|
|
||||||
@ -244,7 +245,7 @@ Socket.prototype.end = function(data, encoding) {
|
|||||||
var shutdownReq = this._handle.shutdown();
|
var shutdownReq = this._handle.shutdown();
|
||||||
|
|
||||||
if (!shutdownReq) {
|
if (!shutdownReq) {
|
||||||
this.destroy(errnoException(errno, 'shutdown'));
|
this._destroy(errnoException(errno, 'shutdown'));
|
||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -267,7 +268,7 @@ function afterShutdown(status, handle, req) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if (self._flags & FLAG_GOT_EOF || !self.readable) {
|
if (self._flags & FLAG_GOT_EOF || !self.readable) {
|
||||||
self.destroy();
|
self._destroy();
|
||||||
} else {
|
} else {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -278,7 +279,7 @@ Socket.prototype.destroySoon = function() {
|
|||||||
this._flags |= FLAG_DESTROY_SOON;
|
this._flags |= FLAG_DESTROY_SOON;
|
||||||
|
|
||||||
if (this._pendingWriteReqs == 0) {
|
if (this._pendingWriteReqs == 0) {
|
||||||
this.destroy();
|
this._destroy();
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
@ -290,11 +291,24 @@ Socket.prototype._connectQueueCleanUp = function(exception) {
|
|||||||
};
|
};
|
||||||
|
|
||||||
|
|
||||||
Socket.prototype.destroy = function(exception) {
|
Socket.prototype._destroy = function(exception, cb) {
|
||||||
if (this.destroyed) return;
|
|
||||||
|
|
||||||
var self = this;
|
var self = this;
|
||||||
|
|
||||||
|
function fireErrorCallbacks() {
|
||||||
|
if (cb) cb(exception);
|
||||||
|
if (exception && !self.errorEmitted) {
|
||||||
|
process.nextTick(function() {
|
||||||
|
self.emit('error', exception);
|
||||||
|
});
|
||||||
|
self.errorEmitted = true;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
if (this.destroyed) {
|
||||||
|
fireErrorCallbacks();
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
self._connectQueueCleanUp();
|
self._connectQueueCleanUp();
|
||||||
|
|
||||||
debug('destroy');
|
debug('destroy');
|
||||||
@ -315,8 +329,9 @@ Socket.prototype.destroy = function(exception) {
|
|||||||
this._handle = null;
|
this._handle = null;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fireErrorCallbacks();
|
||||||
|
|
||||||
process.nextTick(function() {
|
process.nextTick(function() {
|
||||||
if (exception) self.emit('error', exception);
|
|
||||||
self.emit('close', exception ? true : false);
|
self.emit('close', exception ? true : false);
|
||||||
});
|
});
|
||||||
|
|
||||||
@ -324,6 +339,11 @@ Socket.prototype.destroy = function(exception) {
|
|||||||
};
|
};
|
||||||
|
|
||||||
|
|
||||||
|
Socket.prototype.destroy = function(exception) {
|
||||||
|
this._destroy(exception);
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
function onread(buffer, offset, length) {
|
function onread(buffer, offset, length) {
|
||||||
var handle = this;
|
var handle = this;
|
||||||
var self = handle.socket;
|
var self = handle.socket;
|
||||||
@ -362,7 +382,7 @@ function onread(buffer, offset, length) {
|
|||||||
|
|
||||||
// We call destroy() before end(). 'close' not emitted until nextTick so
|
// We call destroy() before end(). 'close' not emitted until nextTick so
|
||||||
// the 'end' event will come first as required.
|
// the 'end' event will come first as required.
|
||||||
if (!self.writable) self.destroy();
|
if (!self.writable) self._destroy();
|
||||||
|
|
||||||
if (!self.allowHalfOpen) self.end();
|
if (!self.allowHalfOpen) self.end();
|
||||||
if (self._events && self._events['end']) self.emit('end');
|
if (self._events && self._events['end']) self.emit('end');
|
||||||
@ -370,9 +390,9 @@ function onread(buffer, offset, length) {
|
|||||||
} else {
|
} else {
|
||||||
// Error
|
// Error
|
||||||
if (errno == 'ECONNRESET') {
|
if (errno == 'ECONNRESET') {
|
||||||
self.destroy();
|
self._destroy();
|
||||||
} else {
|
} else {
|
||||||
self.destroy(errnoException(errno, 'read'));
|
self._destroy(errnoException(errno, 'read'));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -450,13 +470,16 @@ Socket.prototype.write = function(data, arg1, arg2) {
|
|||||||
Socket.prototype._write = function(data, encoding, cb) {
|
Socket.prototype._write = function(data, encoding, cb) {
|
||||||
timers.active(this);
|
timers.active(this);
|
||||||
|
|
||||||
if (!this._handle) throw new Error('This socket is closed.');
|
if (!this._handle) {
|
||||||
|
this._destroy(new Error('This socket is closed.'), cb);
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
// `encoding` is unused right now, `data` is always a buffer.
|
// `encoding` is unused right now, `data` is always a buffer.
|
||||||
var writeReq = this._handle.write(data);
|
var writeReq = this._handle.write(data);
|
||||||
|
|
||||||
if (!writeReq) {
|
if (!writeReq) {
|
||||||
this.destroy(errnoException(errno, 'write'));
|
this._destroy(errnoException(errno, 'write'), cb);
|
||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -477,7 +500,7 @@ function afterWrite(status, handle, req, buffer) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if (status) {
|
if (status) {
|
||||||
self.destroy(errnoException(errno, 'write'));
|
self._destroy(errnoException(errno, 'write'), req.cb);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -494,7 +517,7 @@ function afterWrite(status, handle, req, buffer) {
|
|||||||
if (req.cb) req.cb();
|
if (req.cb) req.cb();
|
||||||
|
|
||||||
if (self._pendingWriteReqs == 0 && self._flags & FLAG_DESTROY_SOON) {
|
if (self._pendingWriteReqs == 0 && self._flags & FLAG_DESTROY_SOON) {
|
||||||
self.destroy();
|
self._destroy();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -522,7 +545,7 @@ function connect(self, address, port, addressType) {
|
|||||||
if (connectReq !== null) {
|
if (connectReq !== null) {
|
||||||
connectReq.oncomplete = afterConnect;
|
connectReq.oncomplete = afterConnect;
|
||||||
} else {
|
} else {
|
||||||
self.destroy(errnoException(errno, 'connect'));
|
self._destroy(errnoException(errno, 'connect'));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -570,7 +593,7 @@ Socket.prototype.connect = function(port /* [host], [cb] */) {
|
|||||||
// error event to the next tick.
|
// error event to the next tick.
|
||||||
process.nextTick(function() {
|
process.nextTick(function() {
|
||||||
self.emit('error', err);
|
self.emit('error', err);
|
||||||
self.destroy();
|
self._destroy();
|
||||||
});
|
});
|
||||||
} else {
|
} else {
|
||||||
timers.active(self);
|
timers.active(self);
|
||||||
@ -619,8 +642,9 @@ function afterConnect(status, handle, req, readable, writable) {
|
|||||||
|
|
||||||
if (self._connectQueue) {
|
if (self._connectQueue) {
|
||||||
debug('Drain the connect queue');
|
debug('Drain the connect queue');
|
||||||
for (var i = 0; i < self._connectQueue.length; i++) {
|
var connectQueue = self._connectQueue;
|
||||||
self._write.apply(self, self._connectQueue[i]);
|
for (var i = 0; i < connectQueue.length; i++) {
|
||||||
|
self._write.apply(self, connectQueue[i]);
|
||||||
}
|
}
|
||||||
self._connectQueueCleanUp();
|
self._connectQueueCleanUp();
|
||||||
}
|
}
|
||||||
@ -634,7 +658,7 @@ function afterConnect(status, handle, req, readable, writable) {
|
|||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
self._connectQueueCleanUp();
|
self._connectQueueCleanUp();
|
||||||
self.destroy(errnoException(errno, 'connect'));
|
self._destroy(errnoException(errno, 'connect'));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -24,21 +24,23 @@ var assert = require('assert');
|
|||||||
var net = require('net');
|
var net = require('net');
|
||||||
|
|
||||||
var gotError = false;
|
var gotError = false;
|
||||||
|
var gotWriteCB = false;
|
||||||
|
|
||||||
process.on('exit', function() {
|
process.on('exit', function() {
|
||||||
assert(gotError);
|
assert(gotError);
|
||||||
|
assert(gotWriteCB);
|
||||||
});
|
});
|
||||||
|
|
||||||
var server = net.createServer(function(socket) {
|
var server = net.createServer(function(socket) {
|
||||||
setTimeout(function() {
|
socket.on('error', function(error) {
|
||||||
assert.throws(
|
|
||||||
function() {
|
|
||||||
socket.write('test');
|
|
||||||
},
|
|
||||||
/This socket is closed/
|
|
||||||
);
|
|
||||||
server.close();
|
server.close();
|
||||||
gotError = true;
|
gotError = true;
|
||||||
|
});
|
||||||
|
|
||||||
|
setTimeout(function() {
|
||||||
|
socket.write('test', function(e) {
|
||||||
|
gotWriteCB = true;
|
||||||
|
});
|
||||||
}, 250);
|
}, 250);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
Loading…
x
Reference in New Issue
Block a user