Skip to content

Commit 153b755

Browse files
ryry
committed
Change IOWatcher constructor to have no arguments
1 parent 6e5abf4 commit 153b755

3 files changed

Lines changed: 41 additions & 44 deletions

File tree

‎lib/net.js‎

Lines changed: 35 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ function debug (x) {
77
}
88

99

10+
var IOWatcher = process.IOWatcher;
1011
var assert = process.assert;
1112
var socket = process.socket;
1213
var bind = process.bind;
@@ -25,18 +26,19 @@ var needsLookup = process.needsLookup;
2526
var EINPROGRESS = process.EINPROGRESS;
2627

2728

28-
function Stream (peerInfo) {
29+
function Socket (peerInfo) {
2930
process.EventEmitter.call();
3031

3132
var self = this;
3233

3334
// Allocated on demand.
3435
self.recvBuffer = null;
3536

36-
self.readWatcher = new process.IOWatcher(function () {
37+
self.readWatcher = new IOWatcher()
38+
self.readWatcher.callback = function () {
3739
// If this is the first recv (recvBuffer doesn't exist) or we've used up
3840
// most of the recvBuffer, allocate a new one.
39-
if (!self.recvBuffer ||
41+
if (!self.recvBuffer ||
4042
self.recvBuffer.length - self.recvBuffer.used < 128) {
4143
self._allocateNewRecvBuf();
4244
}
@@ -52,18 +54,18 @@ function Stream (peerInfo) {
5254
self.readable = false;
5355
self.readWatcher.stop();
5456
self.emit('eof');
55-
if (!self.writable) self.forceClose();
57+
if (!self.writable) self.forceClose();
5658
} else {
5759
var slice = self.recvBuffer.slice(self.recvBuffer.used,
5860
self.recvBuffer.used + bytesRead);
5961
self.recvBuffer.used += bytesRead;
6062
self.emit('receive', slice);
6163
}
62-
});
64+
};
6365
self.readable = false;
6466

6567
self.sendQueue = []; // queue of buffers that need to be written to socket
66-
// XXX use link list?
68+
// XXX use link list?
6769
self.sendQueueSize = 0; // in bytes, not to be confused with sendQueue.length!
6870
self._doFlush = function () {
6971
assert(self.sendQueueSize > 0);
@@ -72,7 +74,8 @@ function Stream (peerInfo) {
7274
self.emit("drain");
7375
}
7476
};
75-
self.writeWatcher = new process.IOWatcher(self._doFlush);
77+
self.writeWatcher = new IOWatcher();
78+
self.writeWatcher.callback = self._doFlush;
7679
self.writable = false;
7780

7881
if (peerInfo) {
@@ -88,18 +91,18 @@ function Stream (peerInfo) {
8891
self.writable = true;
8992
}
9093
};
91-
process.inherits(Stream, process.EventEmitter);
92-
exports.Stream = Stream;
94+
process.inherits(Socket, process.EventEmitter);
95+
exports.Socket = Socket;
9396

9497

9598
exports.createConnection = function (port, host) {
96-
var s = new Stream();
99+
var s = new Socket();
97100
s.connect(port, host);
98101
return s;
99102
};
100103

101104

102-
Stream.prototype._allocateNewRecvBuf = function () {
105+
Socket.prototype._allocateNewRecvBuf = function () {
103106
var self = this;
104107

105108
var newBufferSize = 1024; // TODO make this adjustable from user API
@@ -125,7 +128,7 @@ Stream.prototype._allocateNewRecvBuf = function () {
125128
};
126129

127130

128-
Stream.prototype._allocateSendBuffer = function () {
131+
Socket.prototype._allocateSendBuffer = function () {
129132
var b = new process.Buffer(1024);
130133
b.used = 0;
131134
b.sent = 0;
@@ -134,9 +137,9 @@ Stream.prototype._allocateSendBuffer = function () {
134137
};
135138

136139

137-
Stream.prototype._sendString = function (data, encoding) {
140+
Socket.prototype._sendString = function (data, encoding) {
138141
var self = this;
139-
if (!self.writable) throw new Error('Stream is not writable');
142+
if (!self.writable) throw new Error('Socket is not writable');
140143
var buffer;
141144
if (self.sendQueue.length == 0) {
142145
buffer = self._allocateSendBuffer();
@@ -191,9 +194,9 @@ Stream.prototype._sendString = function (data, encoding) {
191194
// Returns true if all the data was flushed to socket. Returns false if
192195
// something was queued. If data was queued, then the "drain" event will
193196
// signal when it has been finally flushed to socket.
194-
Stream.prototype.send = function (data, encoding) {
197+
Socket.prototype.send = function (data, encoding) {
195198
var self = this;
196-
if (!self.writable) throw new Error('Stream is not writable');
199+
if (!self.writable) throw new Error('Socket is not writable');
197200
if (typeof(data) == 'string') {
198201
self._sendString(data, encoding);
199202
} else {
@@ -220,9 +223,9 @@ Stream.prototype.send = function (data, encoding) {
220223

221224

222225
// Flushes the write buffer out. Emits "drain" if the buffer is empty.
223-
Stream.prototype.flush = function () {
226+
Socket.prototype.flush = function () {
224227
var self = this;
225-
if (!self.writable) throw new Error('Stream is not writable');
228+
if (!self.writable) throw new Error('Socket is not writable');
226229

227230
var bytesWritten;
228231
while (self.sendQueue.length > 0) {
@@ -253,13 +256,13 @@ Stream.prototype.flush = function () {
253256
};
254257

255258

256-
// var stream = new Stream();
259+
// var stream = new Socket();
257260
// stream.connect(80) - TCP connect to port 80 on the localhost
258261
// stream.connect(80, 'nodejs.org') - TCP connect to port 80 on nodejs.org
259262
// stream.connect('/tmp/socket') - UNIX connect to socket specified by path
260-
Stream.prototype.connect = function () {
263+
Socket.prototype.connect = function () {
261264
var self = this;
262-
if (self.fd) throw new Error('Stream already opened');
265+
if (self.fd) throw new Error('Socket already opened');
263266

264267
if (typeof(arguments[0]) == 'string' && arguments.length == 1) {
265268
self.fd = process.socket('UNIX');
@@ -304,7 +307,7 @@ Stream.prototype.connect = function () {
304307
};
305308

306309

307-
Stream.prototype.forceClose = function (exception) {
310+
Socket.prototype.forceClose = function (exception) {
308311
if (this.fd) {
309312
this.readable = false;
310313
this.writable = false;
@@ -314,20 +317,20 @@ Stream.prototype.forceClose = function (exception) {
314317
close(this.fd);
315318
debug('close peer ' + this.fd);
316319
this.fd = null;
317-
this.emit('close', exception);
320+
this.emit('close', exception);
318321
}
319322
};
320323

321324

322-
Stream.prototype._shutdown = function () {
325+
Socket.prototype._shutdown = function () {
323326
if (this.writable) {
324327
this.writable = false;
325328
shutdown(this.fd, "write");
326329
}
327330
};
328331

329332

330-
Stream.prototype.close = function () {
333+
Socket.prototype.close = function () {
331334
var self = this;
332335
var closeMethod;
333336
if (self.readable && self.writable) {
@@ -336,7 +339,7 @@ Stream.prototype.close = function () {
336339
// already got EOF
337340
closeMethod = self.forceClose;
338341
}
339-
// In the case we've already shutdown write side,
342+
// In the case we've already shutdown write side,
340343
// but haven't got EOF: ignore. In the case we're
341344
// fully closed already: ignore.
342345

@@ -358,16 +361,17 @@ function Server (listener) {
358361
self.addListener('connection', listener);
359362
}
360363

361-
self.watcher = new process.IOWatcher(function (readable, writeable) {
364+
self.watcher = new IOWatcher();
365+
self.watcher.callback = function (readable, writeable) {
362366
while (self.fd) {
363367
var peerInfo = accept(self.fd);
364368
debug('accept: ' + JSON.stringify(peerInfo));
365369
if (!peerInfo) return;
366-
var peer = new Stream(peerInfo);
370+
var peer = new Socket(peerInfo);
367371
self.emit('connection', peer);
368372
}
369-
});
370-
};
373+
};
374+
}
371375
process.inherits(Server, process.EventEmitter);
372376
exports.Server = Server;
373377

@@ -411,7 +415,7 @@ Server.prototype.listen = function () {
411415
listen(self.fd, 128);
412416
self.emit("listening");
413417

414-
self.watcher.set(self.fd, true, false);
418+
self.watcher.set(self.fd, true, false);
415419
self.watcher.start();
416420
};
417421

‎src/node_io_watcher.cc‎

Lines changed: 5 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,11 @@ void IOWatcher::Callback(EV_P_ ev_io *w, int revents) {
3737
HandleScope scope;
3838

3939
Local<Value> callback_v = io->handle_->Get(callback_symbol);
40-
assert(callback_v->IsFunction());
40+
if (!callback_v->IsFunction()) {
41+
io->Stop();
42+
return;
43+
}
44+
4145
Local<Function> callback = Local<Function>::Cast(callback_v);
4246

4347
TryCatch try_catch;
@@ -64,19 +68,9 @@ void IOWatcher::Callback(EV_P_ ev_io *w, int revents) {
6468
Handle<Value> IOWatcher::New(const Arguments& args) {
6569
HandleScope scope;
6670

67-
if (!args[0]->IsFunction()) {
68-
return ThrowException(Exception::TypeError(
69-
String::New("First arg should a callback.")));
70-
}
71-
72-
Local<Function> callback = Local<Function>::Cast(args[0]);
73-
7471
IOWatcher *s = new IOWatcher();
75-
7672
s->Wrap(args.This());
7773

78-
s->handle_->Set(callback_symbol, callback);
79-
8074
return args.This();
8175
}
8276

@@ -136,7 +130,6 @@ Handle<Value> IOWatcher::Stop(const Arguments& args) {
136130

137131
void IOWatcher::Stop () {
138132
if (watcher_.active) {
139-
HandleScope scope;
140133
ev_io_stop(EV_DEFAULT_UC_ &watcher_);
141134
Unref();
142135
}

‎src/node_io_watcher.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@ class IOWatcher : ObjectWrap {
2020
}
2121

2222
~IOWatcher() {
23-
Stop();
23+
ev_io_stop(EV_DEFAULT_UC_ &watcher_);
2424
}
2525

2626
static v8::Handle<v8::Value> New(const v8::Arguments& args);

0 commit comments

Comments
 (0)