Chiebot-Mirror
/
grpc
8
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Removed some duplicate stream code

Browse Source
pull/189/head
murgatroid99 10 years ago
parent 25290d2990
commit c92499d1d9
5 changed files with 56 additions and 83 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 6
      src/node/client.js
  2. 2
      src/node/server.js
  3. 56
      src/node/surface_client.js
  4. 73
      src/node/surface_server.js
  5. 2
      src/node/test/interop_sanity_test.js

6
src/node/client.js
Unescape View File

@ -105,7 +105,7 @@ function GrpcClientStream(call, serialize, deserialize) {
return;
}
var data = event.data;
if (self.push(data) && data != null) {
if (self.push(self.deserialize(data)) && data != null) {
self._call.startRead(readCallback);
} else {
reading = false;
@ -155,7 +155,7 @@ GrpcClientStream.prototype._read = function(size) {
*/
GrpcClientStream.prototype._write = function(chunk, encoding, callback) {
var self = this;
self._call.startWrite(chunk, function(event) {
self._call.startWrite(self.serialize(chunk), function(event) {
callback();
}, 0);
};
@ -185,7 +185,7 @@ function makeRequest(channel,
if (metadata) {
call.addMetadata(metadata);
}
return new GrpcClientStream(call);
return new GrpcClientStream(call, serialize, deserialize);
}
/**

2
src/node/server.js
Unescape View File

@ -151,7 +151,7 @@ function GrpcServerStream(call, serialize, deserialize) {
return;
}
var data = event.data;
if (self.push(deserialize(data)) && data != null) {
if (self.push(self.deserialize(data)) && data != null) {
self._call.startRead(readCallback);
} else {
reading = false;

56
src/node/surface_client.js
Unescape View File

@ -63,18 +63,16 @@ util.inherits(ClientReadableObjectStream, Readable);
* client side. Extends from stream.Readable.
* @constructor
* @param {stream} stream Underlying binary Duplex stream for the call
* @param {function(Buffer)} deserialize Function for deserializing binary data
* @param {object} options Stream options
*/
function ClientReadableObjectStream(stream, deserialize, options) {
options = _.extend(options, {objectMode: true});
function ClientReadableObjectStream(stream) {
var options = {objectMode: true};
Readable.call(this, options);
this._stream = stream;
var self = this;
forwardEvent(stream, this, 'status');
forwardEvent(stream, this, 'metadata');
this._stream.on('data', function forwardData(chunk) {
if (!self.push(deserialize(chunk))) {
if (!self.push(chunk)) {
self._stream.pause();
}
});
@ -88,14 +86,11 @@ util.inherits(ClientWritableObjectStream, Writable);
* client side. Extends from stream.Writable.
* @constructor
* @param {stream} stream Underlying binary Duplex stream for the call
* @param {function(*):Buffer} serialize Function for serializing objects
* @param {object} options Stream options
*/
function ClientWritableObjectStream(stream, serialize, options) {
options = _.extend(options, {objectMode: true});
function ClientWritableObjectStream(stream) {
var options = {objectMode: true};
Writable.call(this, options);
this._stream = stream;
this._serialize = serialize;
forwardEvent(stream, this, 'status');
forwardEvent(stream, this, 'metadata');
this.on('finish', function() {
@ -111,20 +106,16 @@ util.inherits(ClientBidiObjectStream, Duplex);
* client side. Extends from stream.Duplex.
* @constructor
* @param {stream} stream Underlying binary Duplex stream for the call
* @param {function(*):Buffer} serialize Function for serializing objects
* @param {function(Buffer)} deserialize Function for deserializing binary data
* @param {object} options Stream options
*/
function ClientBidiObjectStream(stream, serialize, deserialize, options) {
options = _.extend(options, {objectMode: true});
function ClientBidiObjectStream(stream) {
var options = {objectMode: true};
Duplex.call(this, options);
this._stream = stream;
this._serialize = serialize;
var self = this;
forwardEvent(stream, this, 'status');
forwardEvent(stream, this, 'metadata');
this._stream.on('data', function forwardData(chunk) {
if (!self.push(deserialize(chunk))) {
if (!self.push(chunk)) {
self._stream.pause();
}
});
@ -160,7 +151,7 @@ ClientBidiObjectStream.prototype._read = _read;
* @param {function(Error)} callback Callback to call when finished writing
*/
function _write(chunk, encoding, callback) {
this._stream.write(this._serialize(chunk), encoding, callback);
this._stream.write(chunk, encoding, callback);
}
/**
@ -196,15 +187,16 @@ function makeUnaryRequestFunction(method, serialize, deserialize) {
* @return {EventEmitter} An event emitter for stream related events
*/
function makeUnaryRequest(argument, callback, metadata, deadline) {
var stream = client.makeRequest(this.channel, method, metadata, deadline);
var stream = client.makeRequest(this.channel, method, serialize,
deserialize, metadata, deadline);
var emitter = new EventEmitter();
forwardEvent(stream, emitter, 'status');
forwardEvent(stream, emitter, 'metadata');
stream.write(serialize(argument));
stream.write(argument);
stream.end();
stream.on('data', function forwardData(chunk) {
try {
callback(null, deserialize(chunk));
callback(null, chunk);
} catch (e) {
callback(e);
}
@ -236,11 +228,12 @@ function makeClientStreamRequestFunction(method, serialize, deserialize) {
* @return {EventEmitter} An event emitter for stream related events
*/
function makeClientStreamRequest(callback, metadata, deadline) {
var stream = client.makeRequest(this.channel, method, metadata, deadline);
var obj_stream = new ClientWritableObjectStream(stream, serialize, {});
var stream = client.makeRequest(this.channel, method, serialize,
deserialize, metadata, deadline);
var obj_stream = new ClientWritableObjectStream(stream);
stream.on('data', function forwardData(chunk) {
try {
callback(null, deserialize(chunk));
callback(null, chunk);
} catch (e) {
callback(e);
}
@ -272,9 +265,10 @@ function makeServerStreamRequestFunction(method, serialize, deserialize) {
* @return {EventEmitter} An event emitter for stream related events
*/
function makeServerStreamRequest(argument, metadata, deadline) {
var stream = client.makeRequest(this.channel, method, metadata, deadline);
var obj_stream = new ClientReadableObjectStream(stream, deserialize, {});
stream.write(serialize(argument));
var stream = client.makeRequest(this.channel, method, serialize,
deserialize, metadata, deadline);
var obj_stream = new ClientReadableObjectStream(stream);
stream.write(argument);
stream.end();
return obj_stream;
}
@ -301,11 +295,9 @@ function makeBidiStreamRequestFunction(method, serialize, deserialize) {
* @return {EventEmitter} An event emitter for stream related events
*/
function makeBidiStreamRequest(metadata, deadline) {
var stream = client.makeRequest(this.channel, method, metadata, deadline);
var obj_stream = new ClientBidiObjectStream(stream,
serialize,
deserialize,
{});
var stream = client.makeRequest(this.channel, method, serialize,
deserialize, metadata, deadline);
var obj_stream = new ClientBidiObjectStream(stream);
return obj_stream;
}
return makeBidiStreamRequest;

73
src/node/surface_server.js
Unescape View File

@ -54,11 +54,9 @@ util.inherits(ServerReadableObjectStream, Readable);
* server side. Extends from stream.Readable.
* @constructor
* @param {stream} stream Underlying binary Duplex stream for the call
* @param {function(Buffer)} deserialize Function for deserializing binary data
* @param {object} options Stream options
*/
function ServerReadableObjectStream(stream, deserialize, options) {
options = _.extend(options, {objectMode: true});
function ServerReadableObjectStream(stream) {
var options = {objectMode: true};
Readable.call(this, options);
this._stream = stream;
Object.defineProperty(this, 'cancelled', {
@ -66,7 +64,7 @@ function ServerReadableObjectStream(stream, deserialize, options) {
});
var self = this;
this._stream.on('data', function forwardData(chunk) {
if (!self.push(deserialize(chunk))) {
if (!self.push(chunk)) {
self._stream.pause();
}
});
@ -83,14 +81,11 @@ util.inherits(ServerWritableObjectStream, Writable);
* server side. Extends from stream.Writable.
* @constructor
* @param {stream} stream Underlying binary Duplex stream for the call
* @param {function(*):Buffer} serialize Function for serializing objects
* @param {object} options Stream options
*/
function ServerWritableObjectStream(stream, serialize, options) {
options = _.extend(options, {objectMode: true});
function ServerWritableObjectStream(stream) {
var options = {objectMode: true};
Writable.call(this, options);
this._stream = stream;
this._serialize = serialize;
this.on('finish', function() {
this._stream.end();
});
@ -103,18 +98,14 @@ util.inherits(ServerBidiObjectStream, Duplex);
* server side. Extends from stream.Duplex.
* @constructor
* @param {stream} stream Underlying binary Duplex stream for the call
* @param {function(*):Buffer} serialize Function for serializing objects
* @param {function(Buffer)} deserialize Function for deserializing binary data
* @param {object} options Stream options
*/
function ServerBidiObjectStream(stream, serialize, deserialize, options) {
options = _.extend(options, {objectMode: true});
function ServerBidiObjectStream(stream) {
var options = {objectMode: true};
Duplex.call(this, options);
this._stream = stream;
this._serialize = serialize;
var self = this;
this._stream.on('data', function forwardData(chunk) {
if (!self.push(deserialize(chunk))) {
if (!self.push(chunk)) {
self._stream.pause();
}
});
@ -153,7 +144,7 @@ ServerBidiObjectStream.prototype._read = _read;
* @param {function(Error)} callback Callback to call when finished writing
*/
function _write(chunk, encoding, callback) {
this._stream.write(this._serialize(chunk), encoding, callback);
this._stream.write(chunk, encoding, callback);
}
/**
@ -168,11 +159,9 @@ ServerBidiObjectStream.prototype._write = _write;
/**
* Creates a binary stream handler function from a unary handler function
* @param {function(Object, function(Error, *))} handler Unary call handler
* @param {function(*):Buffer} serialize Serialization function
* @param {function(Buffer):*} deserialize Deserialization function
* @return {function(stream)} Binary stream handler
*/
function makeUnaryHandler(handler, serialize, deserialize) {
function makeUnaryHandler(handler) {
/**
* Handles a stream by reading a single data value, passing it to the handler,
* and writing the response back to the stream.
@ -180,7 +169,7 @@ function makeUnaryHandler(handler, serialize, deserialize) {
*/
return function handleUnaryCall(stream) {
stream.on('data', function handleUnaryData(value) {
var call = {request: deserialize(value)};
var call = {request: value};
Object.defineProperty(call, 'cancelled', {
get: function() { return stream.cancelled;}
});
@ -188,7 +177,7 @@ function makeUnaryHandler(handler, serialize, deserialize) {
if (err) {
stream.emit('error', err);
} else {
stream.write(serialize(value));
stream.write(value);
stream.end();
}
});
@ -201,23 +190,21 @@ function makeUnaryHandler(handler, serialize, deserialize) {
* function
* @param {function(Readable, function(Error, *))} handler Client stream call
* handler
* @param {function(*):Buffer} serialize Serialization function
* @param {function(Buffer):*} deserialize Deserialization function
* @return {function(stream)} Binary stream handler
*/
function makeClientStreamHandler(handler, serialize, deserialize) {
function makeClientStreamHandler(handler) {
/**
* Handles a stream by passing a deserializing stream to the handler and
* writing the response back to the stream.
* @param {stream} stream Binary data stream
*/
return function handleClientStreamCall(stream) {
var object_stream = new ServerReadableObjectStream(stream, deserialize, {});
var object_stream = new ServerReadableObjectStream(stream);
handler(object_stream, function sendClientStreamData(err, value) {
if (err) {
stream.emit('error', err);
} else {
stream.write(serialize(value));
stream.write(value);
stream.end();
}
});
@ -228,11 +215,9 @@ function makeClientStreamHandler(handler, serialize, deserialize) {
* Creates a binary stream handler function from a server stream handler
* function
* @param {function(Writable)} handler Server stream call handler
* @param {function(*):Buffer} serialize Serialization function
* @param {function(Buffer):*} deserialize Deserialization function
* @return {function(stream)} Binary stream handler
*/
function makeServerStreamHandler(handler, serialize, deserialize) {
function makeServerStreamHandler(handler) {
/**
* Handles a stream by attaching it to a serializing stream, and passing it to
* the handler.
@ -240,10 +225,8 @@ function makeServerStreamHandler(handler, serialize, deserialize) {
*/
return function handleServerStreamCall(stream) {
stream.on('data', function handleClientData(value) {
var object_stream = new ServerWritableObjectStream(stream,
serialize,
{});
object_stream.request = deserialize(value);
var object_stream = new ServerWritableObjectStream(stream);
object_stream.request = value;
handler(object_stream);
});
};
@ -252,21 +235,16 @@ function makeServerStreamHandler(handler, serialize, deserialize) {
/**
* Creates a binary stream handler function from a bidi stream handler function
* @param {function(Duplex)} handler Unary call handler
* @param {function(*):Buffer} serialize Serialization function
* @param {function(Buffer):*} deserialize Deserialization function
* @return {function(stream)} Binary stream handler
*/
function makeBidiStreamHandler(handler, serialize, deserialize) {
function makeBidiStreamHandler(handler) {
/**
* Handles a stream by wrapping it in a serializing and deserializing object
* stream, and passing it to the handler.
* @param {stream} stream Binary data stream
*/
return function handleBidiStreamCall(stream) {
var object_stream = new ServerBidiObjectStream(stream,
serialize,
deserialize,
{});
var object_stream = new ServerBidiObjectStream(stream);
handler(object_stream);
};
}
@ -341,10 +319,13 @@ function makeServerConstructor(services) {
common.fullyQualifiedName(method) + ' not provided.');
}
var binary_handler = handler_makers[method_type](
service_handlers[service_name][decapitalize(method.name)],
common.serializeCls(method.resolvedResponseType.build()),
common.deserializeCls(method.resolvedRequestType.build()));
server.register(prefix + capitalize(method.name), binary_handler);
service_handlers[service_name][decapitalize(method.name)]);
var serialize = common.serializeCls(
method.resolvedResponseType.build());
var deserialize = common.deserializeCls(
method.resolvedRequestType.build());
server.register(prefix + capitalize(method.name), binary_handler,
serialize, deserialize);
});
}, this);
}

2
src/node/test/interop_sanity_test.js
Unescape View File

@ -65,7 +65,7 @@ describe('Interop tests', function() {
it('should pass ping_pong', function(done) {
interop_client.runTest(port, name_override, 'ping_pong', true, done);
});
it.skip('should pass empty_stream', function(done) {
it('should pass empty_stream', function(done) {
interop_client.runTest(port, name_override, 'empty_stream', true, done);
});
});

Write Preview
Loading…
Cancel
Save