gremlin-javascript/src/main/javascript/gremlin-javascript/lib/driver/connection.js - tinkerpop - Git at Google

 /*
  *  Licensed to the Apache Software Foundation (ASF) under one
  *  or more contributor license agreements.  See the NOTICE file
  *  distributed with this work for additional information
  *  regarding copyright ownership.  The ASF licenses this file
  *  to you under the Apache License, Version 2.0 (the
  *  "License"); you may not use this file except in compliance
  *  with the License.  You may obtain a copy of the License at
  *
  *  http://www.apache.org/licenses/LICENSE-2.0
  *
  *  Unless required by applicable law or agreed to in writing,
  *  software distributed under the License is distributed on an
  *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  *  KIND, either express or implied.  See the License for the
  *  specific language governing permissions and limitations
  *  under the License.
  */

 /**
  * @author Jorge Bay Gondra
  */
 'use strict';

 const EventEmitter = require('events');
 const Stream = require('stream');
 const WebSocket = require('ws');
 const util = require('util');
 const utils = require('../utils');
 const serializer = require('../structure/io/graph-serializer');
 const { graphBinaryReader, graphBinaryWriter } = require('../structure/io/binary/GraphBinary');
 const ResultSet = require('./result-set');
 const ResponseError = require('./response-error');

 const responseStatusCode = {
   success: 200,
   noContent: 204,
   partialContent: 206,
   authenticationChallenge: 407,
 };

 const defaultMimeType = 'application/vnd.gremlin-v3.0+json';
 const graphSON2MimeType = 'application/vnd.gremlin-v2.0+json';
 const graphBinaryMimeType = 'application/vnd.graphbinary-v1.0';

 const pingIntervalDelay = 60 * 1000;
 const pongTimeoutDelay = 30 * 1000;

 /**
  * Represents a single connection to a Gremlin Server.
  */
 class Connection extends EventEmitter {
   /**
    * Creates a new instance of {@link Connection}.
    * @param {String} url The resource uri.
    * @param {Object} [options] The connection options.
    * @param {Array} [options.ca] Trusted certificates.
    * @param {String|Array|Buffer} [options.cert] The certificate key.
    * @param {String} [options.mimeType] The mime type to use.
    * @param {String|Buffer} [options.pfx] The private key, certificate, and CA certs.
    * @param {GraphSONReader} [options.reader] The reader to use.
    * @param {Boolean} [options.rejectUnauthorized] Determines whether to verify or not the server certificate.
    * @param {String} [options.traversalSource] The traversal source. Defaults to: 'g'.
    * @param {GraphSONWriter} [options.writer] The writer to use.
    * @param {Authenticator} [options.authenticator] The authentication handler to use.
    * @param {Object} [options.headers] An associative array containing the additional header key/values for the initial request.
    * @param {Boolean} [options.enableUserAgentOnConnect] Determines if a user agent will be sent during connection handshake. Defaults to: true
    * @param {Boolean} [options.pingEnabled] Setup ping interval. Defaults to: true.
    * @param {Number} [options.pingInterval] Ping request interval in ms if ping enabled. Defaults to: 60000.
    * @param {Number} [options.pongTimeout] Timeout of pong response in ms after sending a ping. Defaults to: 30000.
    * @param {Boolean} [options.connectOnStartup] Deprecated and non-functional. Open websocket on startup.
    * @constructor
    */
   constructor(url, options) {
     super();

     this.url = url;
     this.options = options = options || {};

     /**
      * Gets the MIME type.
      * @type {String}
      */
     this.mimeType = options.mimeType || defaultMimeType;

     // A map containing the request id and the handler. The id should be in lower case to prevent string comparison issues.
     this._responseHandlers = {};
     this._reader = options.reader || this._getDefaultReader(this.mimeType);
     this._writer = options.writer || this._getDefaultWriter(this.mimeType);
     this._openPromise = null;
     this._openCallback = null;
     this._closePromise = null;
     this._closeCallback = null;
     this._pingInterval = null;
     this._pongTimeout = null;

     this._header = String.fromCharCode(this.mimeType.length) + this.mimeType; // TODO: what if mimeType.length > 255
     this._header_buf = Buffer.from(this._header);
     this.isOpen = false;
     this.traversalSource = options.traversalSource || 'g';
     this._authenticator = options.authenticator;
     this._enableUserAgentOnConnect = options.enableUserAgentOnConnect !== false;

     this._pingEnabled = this.options.pingEnabled === false ? false : true;
     this._pingIntervalDelay = this.options.pingInterval || pingIntervalDelay;
     this._pongTimeoutDelay = this.options.pongTimeout || pongTimeoutDelay;

     if (this.options.connectOnStartup) {
       console.warn(
         'connectOnStartup is now deprecated and non-functional. To open a connection, please call open() after instantiating connection object.',
       );
     }
   }

   /**
    * Opens the connection, if its not already opened.
    * @returns {Promise}
    */
   open() {
     if (this.isOpen) {
       return Promise.resolve();
     }
     if (this._openPromise) {
       return this._openPromise;
     }

     this.emit('log', 'ws open');
     let headers = this.options.headers;
     if (this._enableUserAgentOnConnect) {
       if (!headers) {
         headers = [];
       }
       headers[utils.getUserAgentHeader()] = utils.getUserAgent();
     }

     this._ws = new WebSocket(this.url, {
       headers: headers,
       ca: this.options.ca,
       cert: this.options.cert,
       pfx: this.options.pfx,
       rejectUnauthorized: this.options.rejectUnauthorized,
     });

     this._ws.on('message', (data) => this._handleMessage(data));
     this._ws.on('close', (code, message) => this._handleClose(code, message));

     this._ws.on('pong', () => {
       this.emit('log', 'ws pong received');
       if (this._pongTimeout) {
         clearTimeout(this._pongTimeout);
         this._pongTimeout = null;
       }
     });
     this._ws.on('ping', () => {
       this.emit('log', 'ws ping received');
       this._ws.pong();
     });

     return (this._openPromise = new Promise((resolve, reject) => {
       this._ws.on('open', () => {
         this.isOpen = true;
         if (this._pingEnabled) {
           this._pingHeartbeat();
         }
         resolve();
       });
       this._ws.on('error', (err) => {
         this._handleError(err);
         reject(err);
       });
     }));
   }

   /** @override */
   submit(processor, op, args, requestId) {
     // TINKERPOP-2847: Use lower case to prevent string comparison issues.
     const rid = (requestId || utils.getUuid()).toLowerCase();
     return this.open().then(
       () =>
         new Promise((resolve, reject) => {
           if (op !== 'authentication') {
             this._responseHandlers[rid] = {
               callback: (err, result) => (err ? reject(err) : resolve(result)),
               result: null,
             };
           }

           const request = {
             requestId: rid,
             op: op || 'bytecode',
             // if using op eval need to ensure processor stays unset if caller didn't set it.
             processor: !processor && op !== 'eval' ? 'traversal' : processor,
             args: args || {},
           };

           const request_buf = this._writer.writeRequest(request);
           const message = Buffer.concat([this._header_buf, request_buf]);
           this._ws.send(message);
         }),
     );
   }

   /** @override */
   stream(processor, op, args, requestId) {
     // TINKERPOP-2847: Use lower case to prevent string comparison issues.
     const rid = (requestId || utils.getUuid()).toLowerCase();

     const readableStream = new Stream.Readable({
       objectMode: true,
       read() {},
     });

     this._responseHandlers[rid] = {
       callback: (err) => (err ? readableStream.destroy(err) : readableStream.push(null)),
       result: readableStream,
     };

     this.open()
       .then(() => {
         const request = {
           requestId: rid,
           op: op || 'bytecode',
           // if using op eval need to ensure processor stays unset if caller didn't set it.
           processor: !processor && op !== 'eval' ? 'traversal' : processor,
           args: args || {},
         };

         const request_buf = this._writer.writeRequest(request);
         const message = Buffer.concat([this._header_buf, request_buf]);
         this._ws.send(message);
       })
       .catch((err) => readableStream.destroy(err));

     return readableStream;
   }

   _getDefaultReader(mimeType) {
     if (mimeType === graphBinaryMimeType) {
       return graphBinaryReader;
     }

     return mimeType === graphSON2MimeType ? new serializer.GraphSON2Reader() : new serializer.GraphSONReader();
   }

   _getDefaultWriter(mimeType) {
     if (mimeType === graphBinaryMimeType) {
       return graphBinaryWriter;
     }

     return mimeType === graphSON2MimeType ? new serializer.GraphSON2Writer() : new serializer.GraphSONWriter();
   }

   _pingHeartbeat() {
     if (this._pingInterval) {
       clearInterval(this._pingInterval);
       this._pingInterval = null;
     }

     this._pingInterval = setInterval(() => {
       if (this.isOpen === false) {
         // in case of if not open..
         if (this._pingInterval) {
           clearInterval(this._pingInterval);
           this._pingInterval = null;
         }
       }

       this._pongTimeout = setTimeout(() => {
         this._ws.terminate();
       }, this._pongTimeoutDelay);

       this._ws.ping();
     }, this._pingIntervalDelay);
   }

   _handleError(err) {
     this.emit('log', `ws error ${err}`);
     this._cleanupWebsocket(err);
     this.emit('socketError', err);
   }

   _handleClose(code, message) {
     this.emit('log', `ws close code=${code} message=${message}`);
     this._cleanupWebsocket();
     if (this._closeCallback) {
       this._closeCallback();
     }
     this.emit('close', code, message);
   }

   _handleMessage(data) {
     const response = this._reader.readResponse(data);
     if (response.requestId === null || response.requestId === undefined) {
       // There was a serialization issue on the server that prevented the parsing of the request id
       // We invoke any of the pending handlers with an error
       Object.keys(this._responseHandlers).forEach((requestId) => {
         const handler = this._responseHandlers[requestId];
         this._clearHandler(requestId);
         if (response.status !== undefined && response.status.message) {
           return handler.callback(
             // TINKERPOP-2285: keep the old server error message in case folks are parsing that - fix in a future breaking version
             new ResponseError(
               util.format(
                 'Server error (no request information): %s (%d)',
                 response.status.message,
                 response.status.code,
               ),
               response.status,
             ),
           );
         }
         // TINKERPOP-2285: keep the old server error message in case folks are parsing that - fix in a future breaking version
         return handler.callback(
           new ResponseError(util.format('Server error (no request information): %j', response), response.status),
         );
       });
       return;
     }

     // TINKERPOP-2847: Use lower case to prevent string comparison issues.
     response.requestId = response.requestId.toLowerCase();
     const handler = this._responseHandlers[response.requestId];

     if (!handler) {
       // The handler for a given request id was not found
       // It was probably invoked earlier due to a serialization issue.
       return;
     }

     if (response.status.code === responseStatusCode.authenticationChallenge && this._authenticator) {
       this._authenticator
         .evaluateChallenge(response.result.data)
         .then((res) => this.submit(undefined, 'authentication', res, response.requestId))
         .catch(handler.callback);

       return;
     } else if (response.status.code >= 400) {
       // callback in error
       return handler.callback(
         // TINKERPOP-2285: keep the old server error message in case folks are parsing that - fix in a future breaking version
         new ResponseError(
           util.format('Server error: %s (%d)', response.status.message, response.status.code),
           response.status,
         ),
       );
     }

     const isStreamingResponse = handler.result instanceof Stream.Readable;

     switch (response.status.code) {
       case responseStatusCode.noContent:
         this._clearHandler(response.requestId);
         if (isStreamingResponse) {
           handler.result.push(new ResultSet(utils.emptyArray, response.status.attributes));
           return handler.callback(null);
         }
         return handler.callback(null, new ResultSet(utils.emptyArray, response.status.attributes));
       case responseStatusCode.partialContent:
         if (isStreamingResponse) {
           handler.result.push(new ResultSet(response.result.data, response.status.attributes));
           break;
         }
         handler.result = handler.result || [];
         handler.result.push.apply(handler.result, response.result.data);
         break;
       default:
         if (isStreamingResponse) {
           handler.result.push(new ResultSet(response.result.data, response.status.attributes));
           return handler.callback(null);
         }
         if (handler.result) {
           handler.result.push.apply(handler.result, response.result.data);
         } else {
           handler.result = response.result.data;
         }
         this._clearHandler(response.requestId);
         return handler.callback(null, new ResultSet(handler.result, response.status.attributes));
     }
   }

   /**
    * clean websocket context
    */
   _cleanupWebsocket(err) {
     if (this._pingInterval) {
       clearInterval(this._pingInterval);
     }
     this._pingInterval = null;
     if (this._pongTimeout) {
       clearTimeout(this._pongTimeout);
     }
     this._pongTimeout = null;

     // Invoke waiting callbacks to complete Promises when closing the websocket
     Object.keys(this._responseHandlers).forEach((requestId) => {
       const handler = this._responseHandlers[requestId];
       const isStreamingResponse = handler.result instanceof Stream.Readable;
       if (isStreamingResponse) {
         handler.callback(null);
       } else {
         const cause = err ? err : new Error('Connection has been closed.');
         handler.callback(cause);
       }
     });
     this._ws.removeAllListeners();
     this._openPromise = null;
     this._closePromise = null;
     this.isOpen = false;
   }

   /**
    * Clears the internal state containing the callback and result buffer of a given request.
    * @param requestId
    * @private
    */
   _clearHandler(requestId) {
     delete this._responseHandlers[requestId];
   }

   /**
    * Closes the Connection.
    * @return {Promise}
    */
   close() {
     if (this.isOpen === false) {
       return Promise.resolve();
     }
     if (!this._closePromise) {
       this._closePromise = new Promise((resolve) => {
         this._closeCallback = resolve;
         this._ws.close();
       });
     }
     return this._closePromise;
   }
 }

 module.exports = Connection;
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*/

	/**
	* @author Jorge Bay Gondra
	*/
	'use strict';

	const EventEmitter = require('events');
	const Stream = require('stream');
	const WebSocket = require('ws');
	const util = require('util');
	const utils = require('../utils');
	const serializer = require('../structure/io/graph-serializer');
	const { graphBinaryReader, graphBinaryWriter } = require('../structure/io/binary/GraphBinary');
	const ResultSet = require('./result-set');
	const ResponseError = require('./response-error');

	const responseStatusCode = {
	success: 200,
	noContent: 204,
	partialContent: 206,
	authenticationChallenge: 407,
	};

	const defaultMimeType = 'application/vnd.gremlin-v3.0+json';
	const graphSON2MimeType = 'application/vnd.gremlin-v2.0+json';
	const graphBinaryMimeType = 'application/vnd.graphbinary-v1.0';

	const pingIntervalDelay = 60 * 1000;
	const pongTimeoutDelay = 30 * 1000;

	/**
	* Represents a single connection to a Gremlin Server.
	*/
	class Connection extends EventEmitter {
	/**
	* Creates a new instance of {@link Connection}.
	* @param {String} url The resource uri.
	* @param {Object} [options] The connection options.
	* @param {Array} [options.ca] Trusted certificates.
	* @param {String\|Array\|Buffer} [options.cert] The certificate key.
	* @param {String} [options.mimeType] The mime type to use.
	* @param {String\|Buffer} [options.pfx] The private key, certificate, and CA certs.
	* @param {GraphSONReader} [options.reader] The reader to use.
	* @param {Boolean} [options.rejectUnauthorized] Determines whether to verify or not the server certificate.
	* @param {String} [options.traversalSource] The traversal source. Defaults to: 'g'.
	* @param {GraphSONWriter} [options.writer] The writer to use.
	* @param {Authenticator} [options.authenticator] The authentication handler to use.
	* @param {Object} [options.headers] An associative array containing the additional header key/values for the initial request.
	* @param {Boolean} [options.enableUserAgentOnConnect] Determines if a user agent will be sent during connection handshake. Defaults to: true
	* @param {Boolean} [options.pingEnabled] Setup ping interval. Defaults to: true.
	* @param {Number} [options.pingInterval] Ping request interval in ms if ping enabled. Defaults to: 60000.
	* @param {Number} [options.pongTimeout] Timeout of pong response in ms after sending a ping. Defaults to: 30000.
	* @param {Boolean} [options.connectOnStartup] Deprecated and non-functional. Open websocket on startup.
	* @constructor
	*/
	constructor(url, options) {
	super();

	this.url = url;
	this.options = options = options \|\| {};

	/**
	* Gets the MIME type.
	* @type {String}
	*/
	this.mimeType = options.mimeType \|\| defaultMimeType;

	// A map containing the request id and the handler. The id should be in lower case to prevent string comparison issues.
	this._responseHandlers = {};
	this._reader = options.reader \|\| this._getDefaultReader(this.mimeType);
	this._writer = options.writer \|\| this._getDefaultWriter(this.mimeType);
	this._openPromise = null;
	this._openCallback = null;
	this._closePromise = null;
	this._closeCallback = null;
	this._pingInterval = null;
	this._pongTimeout = null;

	this._header = String.fromCharCode(this.mimeType.length) + this.mimeType; // TODO: what if mimeType.length > 255
	this._header_buf = Buffer.from(this._header);
	this.isOpen = false;
	this.traversalSource = options.traversalSource \|\| 'g';
	this._authenticator = options.authenticator;
	this._enableUserAgentOnConnect = options.enableUserAgentOnConnect !== false;

	this._pingEnabled = this.options.pingEnabled === false ? false : true;
	this._pingIntervalDelay = this.options.pingInterval \|\| pingIntervalDelay;
	this._pongTimeoutDelay = this.options.pongTimeout \|\| pongTimeoutDelay;

	if (this.options.connectOnStartup) {
	console.warn(
	'connectOnStartup is now deprecated and non-functional. To open a connection, please call open() after instantiating connection object.',
	);
	}
	}

	/**
	* Opens the connection, if its not already opened.
	* @returns {Promise}
	*/
	open() {
	if (this.isOpen) {
	return Promise.resolve();
	}
	if (this._openPromise) {
	return this._openPromise;
	}

	this.emit('log', 'ws open');
	let headers = this.options.headers;
	if (this._enableUserAgentOnConnect) {
	if (!headers) {
	headers = [];
	}
	headers[utils.getUserAgentHeader()] = utils.getUserAgent();
	}

	this._ws = new WebSocket(this.url, {
	headers: headers,
	ca: this.options.ca,
	cert: this.options.cert,
	pfx: this.options.pfx,
	rejectUnauthorized: this.options.rejectUnauthorized,
	});

	this._ws.on('message', (data) => this._handleMessage(data));
	this._ws.on('close', (code, message) => this._handleClose(code, message));

	this._ws.on('pong', () => {
	this.emit('log', 'ws pong received');
	if (this._pongTimeout) {
	clearTimeout(this._pongTimeout);
	this._pongTimeout = null;
	}
	});
	this._ws.on('ping', () => {
	this.emit('log', 'ws ping received');
	this._ws.pong();
	});

	return (this._openPromise = new Promise((resolve, reject) => {
	this._ws.on('open', () => {
	this.isOpen = true;
	if (this._pingEnabled) {
	this._pingHeartbeat();
	}
	resolve();
	});
	this._ws.on('error', (err) => {
	this._handleError(err);
	reject(err);
	});
	}));
	}

	/** @override */
	submit(processor, op, args, requestId) {
	// TINKERPOP-2847: Use lower case to prevent string comparison issues.
	const rid = (requestId \|\| utils.getUuid()).toLowerCase();
	return this.open().then(
	() =>
	new Promise((resolve, reject) => {
	if (op !== 'authentication') {
	this._responseHandlers[rid] = {
	callback: (err, result) => (err ? reject(err) : resolve(result)),
	result: null,
	};
	}

	const request = {
	requestId: rid,
	op: op \|\| 'bytecode',
	// if using op eval need to ensure processor stays unset if caller didn't set it.
	processor: !processor && op !== 'eval' ? 'traversal' : processor,
	args: args \|\| {},
	};

	const request_buf = this._writer.writeRequest(request);
	const message = Buffer.concat([this._header_buf, request_buf]);
	this._ws.send(message);
	}),
	);
	}

	/** @override */
	stream(processor, op, args, requestId) {
	// TINKERPOP-2847: Use lower case to prevent string comparison issues.
	const rid = (requestId \|\| utils.getUuid()).toLowerCase();

	const readableStream = new Stream.Readable({
	objectMode: true,
	read() {},
	});

	this._responseHandlers[rid] = {
	callback: (err) => (err ? readableStream.destroy(err) : readableStream.push(null)),
	result: readableStream,
	};

	this.open()
	.then(() => {
	const request = {
	requestId: rid,
	op: op \|\| 'bytecode',
	// if using op eval need to ensure processor stays unset if caller didn't set it.
	processor: !processor && op !== 'eval' ? 'traversal' : processor,
	args: args \|\| {},
	};

	const request_buf = this._writer.writeRequest(request);
	const message = Buffer.concat([this._header_buf, request_buf]);
	this._ws.send(message);
	})
	.catch((err) => readableStream.destroy(err));

	return readableStream;
	}

	_getDefaultReader(mimeType) {
	if (mimeType === graphBinaryMimeType) {
	return graphBinaryReader;
	}

	return mimeType === graphSON2MimeType ? new serializer.GraphSON2Reader() : new serializer.GraphSONReader();
	}

	_getDefaultWriter(mimeType) {
	if (mimeType === graphBinaryMimeType) {
	return graphBinaryWriter;
	}

	return mimeType === graphSON2MimeType ? new serializer.GraphSON2Writer() : new serializer.GraphSONWriter();
	}

	_pingHeartbeat() {
	if (this._pingInterval) {
	clearInterval(this._pingInterval);
	this._pingInterval = null;
	}

	this._pingInterval = setInterval(() => {
	if (this.isOpen === false) {
	// in case of if not open..
	if (this._pingInterval) {
	clearInterval(this._pingInterval);
	this._pingInterval = null;
	}
	}

	this._pongTimeout = setTimeout(() => {
	this._ws.terminate();
	}, this._pongTimeoutDelay);

	this._ws.ping();
	}, this._pingIntervalDelay);
	}

	_handleError(err) {
	this.emit('log', `ws error ${err}`);
	this._cleanupWebsocket(err);
	this.emit('socketError', err);
	}

	_handleClose(code, message) {
	this.emit('log', `ws close code=${code} message=${message}`);
	this._cleanupWebsocket();
	if (this._closeCallback) {
	this._closeCallback();
	}
	this.emit('close', code, message);
	}

	_handleMessage(data) {
	const response = this._reader.readResponse(data);
	if (response.requestId === null \|\| response.requestId === undefined) {
	// There was a serialization issue on the server that prevented the parsing of the request id
	// We invoke any of the pending handlers with an error
	Object.keys(this._responseHandlers).forEach((requestId) => {
	const handler = this._responseHandlers[requestId];
	this._clearHandler(requestId);
	if (response.status !== undefined && response.status.message) {
	return handler.callback(
	// TINKERPOP-2285: keep the old server error message in case folks are parsing that - fix in a future breaking version
	new ResponseError(
	util.format(
	'Server error (no request information): %s (%d)',
	response.status.message,
	response.status.code,
	),
	response.status,
	),
	);
	}
	// TINKERPOP-2285: keep the old server error message in case folks are parsing that - fix in a future breaking version
	return handler.callback(
	new ResponseError(util.format('Server error (no request information): %j', response), response.status),
	);
	});
	return;
	}

	// TINKERPOP-2847: Use lower case to prevent string comparison issues.
	response.requestId = response.requestId.toLowerCase();
	const handler = this._responseHandlers[response.requestId];

	if (!handler) {
	// The handler for a given request id was not found
	// It was probably invoked earlier due to a serialization issue.
	return;
	}

	if (response.status.code === responseStatusCode.authenticationChallenge && this._authenticator) {
	this._authenticator
	.evaluateChallenge(response.result.data)
	.then((res) => this.submit(undefined, 'authentication', res, response.requestId))
	.catch(handler.callback);

	return;
	} else if (response.status.code >= 400) {
	// callback in error
	return handler.callback(
	// TINKERPOP-2285: keep the old server error message in case folks are parsing that - fix in a future breaking version
	new ResponseError(
	util.format('Server error: %s (%d)', response.status.message, response.status.code),
	response.status,
	),
	);
	}

	const isStreamingResponse = handler.result instanceof Stream.Readable;

	switch (response.status.code) {
	case responseStatusCode.noContent:
	this._clearHandler(response.requestId);
	if (isStreamingResponse) {
	handler.result.push(new ResultSet(utils.emptyArray, response.status.attributes));
	return handler.callback(null);
	}
	return handler.callback(null, new ResultSet(utils.emptyArray, response.status.attributes));
	case responseStatusCode.partialContent:
	if (isStreamingResponse) {
	handler.result.push(new ResultSet(response.result.data, response.status.attributes));
	break;
	}
	handler.result = handler.result \|\| [];
	handler.result.push.apply(handler.result, response.result.data);
	break;
	default:
	if (isStreamingResponse) {
	handler.result.push(new ResultSet(response.result.data, response.status.attributes));
	return handler.callback(null);
	}
	if (handler.result) {
	handler.result.push.apply(handler.result, response.result.data);
	} else {
	handler.result = response.result.data;
	}
	this._clearHandler(response.requestId);
	return handler.callback(null, new ResultSet(handler.result, response.status.attributes));
	}
	}

	/**
	* clean websocket context
	*/
	_cleanupWebsocket(err) {
	if (this._pingInterval) {
	clearInterval(this._pingInterval);
	}
	this._pingInterval = null;
	if (this._pongTimeout) {
	clearTimeout(this._pongTimeout);
	}
	this._pongTimeout = null;

	// Invoke waiting callbacks to complete Promises when closing the websocket
	Object.keys(this._responseHandlers).forEach((requestId) => {
	const handler = this._responseHandlers[requestId];
	const isStreamingResponse = handler.result instanceof Stream.Readable;
	if (isStreamingResponse) {
	handler.callback(null);
	} else {
	const cause = err ? err : new Error('Connection has been closed.');
	handler.callback(cause);
	}
	});
	this._ws.removeAllListeners();
	this._openPromise = null;
	this._closePromise = null;
	this.isOpen = false;
	}

	/**
	* Clears the internal state containing the callback and result buffer of a given request.
	* @param requestId
	* @private
	*/
	_clearHandler(requestId) {
	delete this._responseHandlers[requestId];
	}

	/**
	* Closes the Connection.
	* @return {Promise}
	*/
	close() {
	if (this.isOpen === false) {
	return Promise.resolve();
	}
	if (!this._closePromise) {
	this._closePromise = new Promise((resolve) => {
	this._closeCallback = resolve;
	this._ws.close();
	});
	}
	return this._closePromise;
	}
	}

	module.exports = Connection;