nicdex/node-eventstore-client
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Fix a bug in tcpConnection that was blocking sending of packet > 64k. Fixes #27

Browse Source
This commit is contained in:
Nicolas Dextraze 2017-03-24 19:04:01 -07:00
parent eaec15fe74
commit 217c1c6bb8
1 changed files with 13 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
src/transport/tcp/tcpConnection.js
View File

@ -1,7 +1,7 @@
var net = require('net'); var net = require('net');
var createBufferSegment = require('../../common/bufferSegment'); var createBufferSegment = require('../../common/bufferSegment');
const MaxSendPacketSize = 64 * 1000; const MaxSendPacketSize = 64 * 1024;
function TcpConnection(log, connectionId, remoteEndPoint, onConnectionClosed) { function TcpConnection(log, connectionId, remoteEndPoint, onConnectionClosed) {
this._socket = null; this._socket = null;
@ -34,8 +34,11 @@ TcpConnection.prototype._initSocket = function(socket) {
this._localEndPoint = {host: socket.localAddress, port: socket.localPort}; this._localEndPoint = {host: socket.localAddress, port: socket.localPort};
this._remoteEndPoint.host = socket.remoteAddress; this._remoteEndPoint.host = socket.remoteAddress;
this._socket.on('drain', this._trySend.bind(this));
this._socket.on('error', this._processError.bind(this)); this._socket.on('error', this._processError.bind(this));
this._socket.on('data', this._processReceive.bind(this)); this._socket.on('data', this._processReceive.bind(this));
this._trySend();
}; };
TcpConnection.prototype.enqueueSend = function(bufSegmentArray) { TcpConnection.prototype.enqueueSend = function(bufSegmentArray) {
@ -54,19 +57,20 @@ TcpConnection.prototype._trySend = function() {
var buffers = []; var buffers = [];
var bytes = 0; var bytes = 0;
var sendPiece = this._sendQueue.shift(); var sendPiece;
while(sendPiece) { while(sendPiece = this._sendQueue.shift()) {
if (bytes + sendPiece.length > MaxSendPacketSize)
break;
buffers.push(sendPiece); buffers.push(sendPiece);
bytes += sendPiece.length; bytes += sendPiece.length;
if (bytes > MaxSendPacketSize)
sendPiece = this._sendQueue.shift(); break;
} }
var joinedBuffers = Buffer.concat(buffers, bytes); var joinedBuffers = Buffer.concat(buffers, bytes);
this._socket.write(joinedBuffers); if (!this._socket.write(joinedBuffers)) {
return;
}
setImmediate(this._trySend.bind(this));
}; };
TcpConnection.prototype._processError = function(err) { TcpConnection.prototype._processError = function(err) {