'use strict';
var net = require('net');
var tls = require('tls');
var events = require('events');
var util = require('util');
var fs = require('fs');
var logger = require('winston');
var ByteBuffer = require('bytebuffer');
var StartTls = require('./starttls');
var AuthReq = require('./authreq');
var utils = require('./utils');
var rpb = require('../protobuf/riakprotobuf');
var DEFAULT_MAX_BUFFER = 2048 * 1024;
var DEFAULT_INIT_BUFFER = 2 * 1024;
// TODO FUTURE these are shared with RiakNode
var DEFAULT_CONNECTION_TIMEOUT = 3000;
var DEFAULT_REQUEST_TIMEOUT = 5000;
// NB: fixes GH 104
// https://github.com/basho/riak-nodejs-client/issues/104
// TODO FUTURE: remove this when Riak uses Erlang R17 or higher.
var RIAK_R16_CIPHERS = 'DHE-RSA-AES128-SHA256:DHE-RSA-AES128-SHA:DHE-RSA-AES256-SHA256:DHE-RSA-AES256-SHA:AES128-SHA256:AES128-SHA:AES256-SHA256:AES256-SHA:RC4-SHA';
var cid = {};
function debugOutputConnectionListeners(name, conn) {
if (logger.debug) {
logger.debug('%s listeners for "close" event: %d', name, conn.listenerCount('close'));
logger.debug('%s listeners for "connect" event: %d', name, conn.listenerCount('connect'));
logger.debug('%s listeners for "data" event: %d', name, conn.listenerCount('data'));
logger.debug('%s listeners for "drain" event: %d', name, conn.listenerCount('drain'));
logger.debug('%s listeners for "end" event: %d', name, conn.listenerCount('end'));
logger.debug('%s listeners for "error" event: %d', name, conn.listenerCount('error'));
logger.debug('%s listeners for "lookup" event: %d', name, conn.listenerCount('lookup'));
logger.debug('%s listeners for "timeout" event: %d', name, conn.listenerCount('timeout'));
}
}
/**
* @module Core
*/
/**
* Provides the RiakConnection class.
* @class RiakConnection
* @constructor
* @param {Object} options - the options to use.
*/
function RiakConnection(options) {
events.EventEmitter.call(this);
this.remoteAddress = options.remoteAddress;
this.remotePort = options.remotePort;
// This is to facilitate debugging
if (!cid[this.remotePort]) {
cid[this.remotePort] = 1;
}
this.name = util.format('[RiakConnection] (%s:%d-%d)',
this.remoteAddress, this.remotePort, cid[this.remotePort]);
cid[this.remotePort]++;
if (options.cork) {
this.cork = true;
}
if (options.auth) {
this.auth = options.auth;
this.auth.ciphers = RIAK_R16_CIPHERS;
}
if (options.healthCheck) {
this.healthCheck = options.healthCheck;
}
this.connectionTimeout = DEFAULT_CONNECTION_TIMEOUT;
if (options.hasOwnProperty('connectionTimeout')) {
this.connectionTimeout = options.connectionTimeout;
}
this.requestTimeout = DEFAULT_REQUEST_TIMEOUT;
if (options.hasOwnProperty('requestTimeout')) {
this.requestTimeout = options.requestTimeout;
}
this.maxBufferSize = DEFAULT_MAX_BUFFER;
if (options.hasOwnProperty('maxBufferSize')) {
this.maxBufferSize = options.maxBufferSize;
}
var initBufferSize = DEFAULT_INIT_BUFFER;
if (options.hasOwnProperty('initBufferSize')) {
initBufferSize = options.initBufferSize;
}
this._emitAndClose = function(evt, evt_args) {
if (!this.closed) {
// NB: this can be useful
// logger.debug("%s emitting '%s' args '%s'", this.name, evt, evt_args);
// NB: RiakNode checks inFlight to re-try command if necessary
// so don't set inFlight to false here, it will be set to false in close()
this.closed = true;
this._connection.end();
this.emit(evt, this, evt_args);
this.close();
}
};
this._connHandleEnd = function () {
logger.debug('%s handling "end" event', this.name);
this._emitAndClose('connectionClosed');
};
this._connHandleTimeout = function (command) {
var err = util.format("%s command '%s' timed out (in-flight: %s)",
this.name, command.name, this.inFlight);
if (logger.debug) {
logger.debug(err);
}
this._emitAndClose('connectionClosed');
};
this._clearSocketTimeout = function() {
if (this._connection) {
if (this._boundConnectionTimeout) {
this._connection.removeListener('timeout', this._boundConnectionTimeout);
this._boundConnectionTimeout = null;
}
this._connection.setTimeout(0);
}
};
// buffer is private
var buffer = null;
// private buffer functions
function initBuffer(data) {
// Create a new buffer to receive data if needed
if (buffer === null) {
buffer = new ByteBuffer(initBufferSize);
}
buffer.append(data);
buffer.flip();
}
function getProtobufsFromBuffer(protobufArray) {
if (arguments.length === 0) {
protobufArray = [];
}
if (buffer.remaining() >= 4) {
buffer.mark();
var messageLength = buffer.readUint32();
// See if we have the complete message
if (buffer.remaining() >= messageLength) {
// We have a complete message from riak
var slice = buffer.slice(undefined, buffer.offset + messageLength);
var code = slice.readUint8();
// Our fun API does some creative things like ... returning only
// a code, with 0 bytes following. In those cases we want to set
// decoded to null.
var decoded = null;
if (messageLength - 1 > 0) {
var ResponseProto = rpb.getProtoFor(code);
// GH issue #45
// Must use 'true' as argument to force copy of data
// otherwise, subsequent fetches will clobber data
decoded = ResponseProto.decode(slice.toBuffer(true));
}
protobufArray[protobufArray.length] = { msgCode : code, protobuf : decoded };
// skip past message in buffer
buffer.skip(messageLength);
// recursively call this until we are out of messages
return getProtobufsFromBuffer(protobufArray);
} else {
// rewind the offset
buffer.reset();
}
}
// ByteBuffer's 'flip()' effectively clears the buffer which we don't
// want. We want to flip while preserving anything in the buffer and
// compact if necessary.
var newOffset = buffer.remaining();
// Compact if necessary
if (newOffset > 0 && buffer.offset !== 0) {
buffer.copyTo(buffer, 0);
}
buffer.offset = newOffset;
buffer.limit = buffer.capacity();
return protobufArray;
}
function closeBuffer() {
if (buffer) {
buffer.clear();
buffer = null;
}
}
// protected buffer functions
this._closeBuffer = function () {
closeBuffer();
};
this._resetBuffer = function () {
if (buffer && buffer.capacity() > this.maxBufferSize) {
closeBuffer();
}
};
this._buildProtobufArray = function (data) {
initBuffer(data);
return getProtobufsFromBuffer();
};
// protected execute functions
this._executeInit = function() {
this.lastUsed = Date.now();
this.executeDone();
};
this._executeStart = function(command) {
this.command = command;
logger.debug('%s execute command:', this.name, command.name);
this.inFlight = true;
this.lastUsed = Date.now();
};
this._executeInit();
this.closed = false;
this._connectedEmitted = false;
this._connection = new net.Socket();
if (this._connection.setKeepAlive) {
this._connection.setKeepAlive(true, 0);
}
if (this._connection.setNoDelay) {
this._connection.setNoDelay(true);
}
// Note: useful for debugging event issues
/*
debugOutputConnectionListeners(this.name, this._connection);
this.setMaxListeners(1);
this._connection.setMaxListeners(1);
*/
if (this.cork && !this._connection.cork) {
logger.warn('%s wanted to use cork/uncork but not supported!', this.name);
this.cork = false;
} else {
logger.debug('%s using cork() / uncork()', this.name);
}
}
util.inherits(RiakConnection, events.EventEmitter);
RiakConnection.prototype.connect = function() {
this._boundConnectionError = this._connectionError.bind(this);
this._connection.on('error', this._boundConnectionError);
// This *is* the read/write timeout as well as idle timeout
// https://nodejs.org/api/net.html#net_socket_settimeout_timeout_callback
this._boundConnectionTimeout = this._connectionTimeout.bind(this);
this._connection.setTimeout(this.connectionTimeout, this._boundConnectionTimeout);
this._boundConnected = this._connected.bind(this);
this._connection.connect(this.remotePort, this.remoteAddress, this._boundConnected);
};
RiakConnection.prototype._connected = function() {
logger.debug('%s connected', this.name);
this._connection.removeListener('connect', this._boundConnected);
this._boundConnected = null;
this._connection.removeListener('error', this._boundConnectionError);
this._boundConnectionError = this._socketError.bind(this);
this._connection.on('error', this._boundConnectionError);
this._connection.on('end', this._connHandleEnd.bind(this));
this._connection.on('data', this._receiveData.bind(this));
if (this.auth) {
/*
* NB: at this point, we have not yet emitted the 'connected' event,
* so listeners will not have yet registered for 'connectionClosed'.
* This is why the 'close' event must raise 'connectFailed' via
* _boundConnectionError
*/
this._connection.on('close', this._boundConnectionError);
this._boundResponseReceived = this._receiveStartTls.bind(this);
this.on('responseReceived', this._boundResponseReceived);
var command = new StartTls();
this.execute(command);
} else if (this.healthCheck) {
// NB: see above comment re: 'close' event
this._connection.on('close', this._boundConnectionError);
this._boundResponseReceived = this._receiveHealthCheck.bind(this);
this.on('responseReceived', this._boundResponseReceived);
this.execute(this.healthCheck);
} else {
this._clearSocketTimeout();
this._connection.on('close', this._connClosed.bind(this));
logger.debug('%s emit connected, no-auth', this.name);
this._connectedEmitted = true;
this.emit('connected', this);
}
};
RiakConnection.prototype._connectionError = function(err) {
this._emitAndClose('connectFailed', err);
};
RiakConnection.prototype._connectionTimeout = function(err) {
if (!err) {
err = 'timed out or other error trying to connect';
}
this._connectionError(err);
};
RiakConnection.prototype._socketError = function(err) {
// This is only called if we have an error after a successful connection
// log only because close will be called right after
// https://nodejs.org/api/net.html#net_event_error
if (err) {
logger.error('%s _socketError:', this.name);
}
};
RiakConnection.prototype._receiveHealthCheck = function(conn, command, code, decoded) {
// NB: this function is similar to _responseReceived in RiakNode
logger.debug('%s receive health check response', this.name);
this.executeDone();
this.removeListener('responseReceived', this._boundResponseReceived);
this._boundResponseReceived = null;
this._connection.removeListener('close', this._boundConnectionError);
this._boundConnectionError = null;
this._connection.on('close', this._connClosed.bind(this));
var self = this;
function onError(err) {
logger.error(err.msg);
self._connectionError(err.msg);
}
function onSuccess() {
logger.debug('%s health check, emit connected', self.name);
self._connectedEmitted = true;
self.emit('connected', self);
}
var data = {
conn: conn,
command: command,
code: code,
decoded: decoded,
shouldCallback: true
};
utils.handleRiakResponse(data, onError, onSuccess);
};
RiakConnection.prototype._receiveStartTls = function(conn, command, code, decoded) {
// NB: this function is similar to _responseReceived in RiakNode
logger.debug('%s receive StartTls response', this.name);
this.executeDone();
this.removeListener('responseReceived', this._boundResponseReceived);
this._boundResponseReceived = null;
var self = this;
function onError(err) {
logger.error(err.msg);
self._connectionError(err.msg);
}
function onSuccess() {
var tls_secure_context = tls.createSecureContext(self.auth);
var tls_socket_options = {
isServer: false, // NB: required
secureContext: tls_secure_context
};
self._connection.removeListener('error', self._boundConnectionError);
self._connection = new tls.TLSSocket(self._connection, tls_socket_options);
// NB: *must* re-register for data event!
self._connection.on('data', self._receiveData.bind(self));
self._connection.on('error', self._boundConnectionError);
// NB: this is necessary since we have not yet emitted the 'connected' event
// in which case the execute method would set up the timeout
// _receiveData clears the timeout as well as the bound method
self._boundConnectionTimeout = self._connectionTimeout.bind(self);
self._connection.setTimeout(self.connectionTimeout, self._boundConnectionTimeout);
var auth_options = {
user: self.auth.user,
password: self.auth.password
};
// On responseReceived event, move to next sequence in TLS negotiation
self._boundResponseReceived = self._receiveAuthResp.bind(self);
self.on('responseReceived', self._boundResponseReceived);
// Execute AuthReq command
/*
if (logger.debug) {
debugOutputConnectionListeners(self.name, self._connection);
}
*/
var command = new AuthReq(auth_options);
self.execute(command);
}
var data = {
conn: conn,
command: command,
code: code,
decoded: decoded,
shouldCallback: true
};
utils.handleRiakResponse(data, onError, onSuccess);
};
RiakConnection.prototype._receiveAuthResp = function(conn, command, code, decoded) {
logger.debug('%s receive RpbAuthResp', this.name);
this.executeDone();
this.removeListener('responseReceived', this._boundResponseReceived);
this._boundResponseReceived = null;
this._connection.removeListener('close', this._boundConnectionError);
this._boundConnectionError = null;
this._connection.on('close', this._connClosed.bind(this));
var self = this;
function onError(err) {
logger.error(err.msg);
self._connectionError(err.msg);
}
function onSuccess() {
logger.debug('%s emit connected, with-auth', self.name);
self._connectedEmitted = true;
self.emit('connected', self);
}
var data = {
conn: conn,
command: command,
code: code,
decoded: decoded,
shouldCallback: true
};
utils.handleRiakResponse(data, onError, onSuccess);
};
RiakConnection.prototype._receiveData = function(data) {
var protobufArray = this._buildProtobufArray(data);
for (var i = 0; i < protobufArray.length; i++) {
this._clearSocketTimeout();
this.emit('responseReceived', this,
this.command, protobufArray[i].msgCode, protobufArray[i].protobuf);
}
};
// TODO FUTURE: what does "had_error" really mean?
RiakConnection.prototype._connClosed = function(had_error) {
this._emitAndClose('connectionClosed');
};
RiakConnection.prototype.close = function() {
this.closed = true;
this.executeDone();
this.removeAllListeners();
this._closeBuffer();
if (this._connection) {
this._connection.end();
this._connection.removeAllListeners();
this._connection.on('error', function (err) {
if (err) {
logger.error('%s error AFTER close:', this.name, err);
}
});
this._connection = null;
}
logger.debug('%s closed', this.name);
};
RiakConnection.prototype.executeDone = function() {
this.inFlight = false;
this.command = {
name: 'no-command'
};
this._resetBuffer();
};
// command includes user callback
RiakConnection.prototype.execute = function(command) {
if (this.inFlight === true) {
logger.error('%s attempted to run command "%s" on in-use connection',
this.name, command.name);
return false;
}
this._executeStart(command);
// write PB to socket
var message = command.getRiakMessage();
/*
* NB: only bind to 'timeout' if 'connected' event has been emitted.
* Initial connection, health check and starting TLS bind 'timeout'
* to a handler that will raise 'connectFailed' on timeout
*/
if (this._connectedEmitted) {
if (this._boundConnectionTimeout) {
this._connection.removeListener('timeout', this._boundConnectionTimeout);
}
this._boundConnectionTimeout = this._connHandleTimeout.bind(this, command);
this._connection.setTimeout(this.requestTimeout, this._boundConnectionTimeout);
}
/*
* Use of cork()/uncork() suggested by Doug Luce
* https://github.com/dougluce
* https://github.com/basho/riak-nodejs-client/pull/56
* https://github.com/basho/riak-nodejs-client/pull/57
*/
if (this.cork) {
this._connection.cork();
}
this._connection.write(message.header);
if (message.protobuf) {
this._connection.write(message.protobuf);
}
if (this.cork) {
this._connection.uncork();
}
return true;
};
module.exports = RiakConnection;
