Huge News!Announcing our $40M Series B led by Abstract Ventures.Learn More →
Socket
Sign inDemoInstall
Socket
Sign inDemoInstall

mongodb-core

Package Overview
Dependencies
Maintainers
1
Versions
177
Alerts
File Explorer

Advanced tools

License
Socket logo

Install Socket

Detect and block malicious and high-risk dependencies

Install

mongodb-core - npm Package Compare versions

Comparing version 1.0.0-alpha7 to 1.0.0-alpha8

lib/auth/scram.js
lib/wireprotocol/2_4_support.js
lib/wireprotocol/2_6_support.js
lib/wireprotocol/commands.js

319

lib/cursor.js

@@ -44,6 +44,3 @@ var Response = require('./connection/commands').Response

// Unpack options
var batchSize = options.batchSize || cmd.batchSize || 0;
var limit = options.limit || cmd.limit || 0;
var skip = options.skip || cmd.skip || 0;
// Current limit processed, used to cut the number of returned docs correctly
var currentLimit = 0;

@@ -53,2 +50,4 @@

var connection = null;
// Cursor server
var server = null;

@@ -58,10 +57,17 @@ // Do we have a not connected handler

// All internal state
// Cursor reference
var self = this;
var cursorId = null;
var documents = options.documents || [];
var dead = false;
var killed = false;
var init = false;
// All internal state
var cursorState = {
cursorId: null
, documents: options.documents || []
, dead: false
, killed: false
, init: false
, limit: options.limit || cmd.limit || 0
, skip: options.skip || cmd.skip || 0
, batchSize: options.batchSize || cmd.batchSize || 0
}
// Callback controller

@@ -76,5 +82,5 @@ var callbacks = null;

if(typeof cmd == 'number') {
cursorId = Long.fromNumber(cmd);
cursorState.cursorId = Long.fromNumber(cmd);
} else if(cmd instanceof Long) {
cursorId = cmd;
cursorState.cursorId = cmd;
}

@@ -86,4 +92,4 @@

enumerable:true,
set: function(value) { batchSize = value; }
, get: function() { return batchSize; }
set: function(value) { cursorState.batchSize = value; }
, get: function() { return cursorState.batchSize; }
});

@@ -94,4 +100,4 @@

enumerable:true,
set: function(value) { limit = value; }
, get: function() { return limit; }
set: function(value) { cursorState.limit = value; }
, get: function() { return cursorState.limit; }
});

@@ -102,4 +108,4 @@

enumerable:true,
set: function(value) { skip = value; }
, get: function() { return skip; }
set: function(value) { cursorState.skip = value; }
, get: function() { return cursorState.skip; }
});

@@ -134,6 +140,6 @@

this.next = function(callback) {
if(killed) return handleCallback(callback, null, null);
if(dead) return handleCallback(callback, new MongoError("cursor is dead"));
if(cursorState.killed) return handleCallback(callback, null, null);
if(cursorState.dead) return handleCallback(callback, new MongoError("cursor is dead"));
// We have just started the cursor
if(!init) {
if(!cursorState.init) {
// Topology is not connected, save the call in the provided store to be

@@ -147,3 +153,3 @@ // Executed at some point when the handler deems it's reconnected

// Get a server
var server = topology.getServer(options);
server = topology.getServer(options);
// Get a connection

@@ -158,12 +164,6 @@ connection = server.getConnection();

// Set as init
init = true;
// Establish type of command
if(cmd.find) {
query = setupClassicFind(ns, cmd, topology, options)
} else if(cursorId != null) {
} else if(cmd) {
query = setupCommand(ns, cmd, topology, options);
} else {
throw new MongoError(f("command %s does not return a cursor", JSON.stringify(cmd)));
}
cursorState.init = true;
// Get the right wire protocol command
query = server.wireProtocolHandler.command(bson, ns, cmd, cursorState, topology, options);
}

@@ -174,3 +174,3 @@

if(err) {
dead = true;
cursorState.dead = true;
callbacks.unregister(query.requestId);

@@ -181,7 +181,7 @@ return callback(err);

// Concatenate all the documents
documents = documents.concat(result.documents);
cursorState.documents = cursorState.documents.concat(result.documents);
// If we have no documents left
if(Long.ZERO.equals(result.cursorId)) {
cursorId = Long.ZERO;
cursorState.cursorId = Long.ZERO;
callbacks.unregister(query.requestId);

@@ -195,4 +195,4 @@ return self.next(callback);

// Initial result
if(cursorId == null) {
cursorId = result.cursorId;
if(cursorState.cursorId == null) {
cursorState.cursorId = result.cursorId;
self.next(callback);

@@ -203,3 +203,3 @@ }

// If we have exhaust
if(options.exhaust && cursorId == null) {
if(options.exhaust && cursorState.cursorId == null) {
// Handle all the exhaust responses

@@ -209,5 +209,5 @@ callbacks.register(query.requestId, processExhaustMessages);

return connection.write(query);
} else if(options.exhaust && documents.length > 0) {
return handleCallback(callback, null, documents.shift());
} else if(options.exhaust && Long.ZERO.equals(cursorId)) {
} else if(options.exhaust && cursorState.documents.length > 0) {
return handleCallback(callback, null, cursorState.documents.shift());
} else if(options.exhaust && Long.ZERO.equals(cursorState.cursorId)) {
callbacks.unregister(query.requestId);

@@ -217,3 +217,3 @@ return handleCallback(callback, null, null);

return setTimeout(function() {
if(Long.ZERO.equals(cursorId)) return;
if(Long.ZERO.equals(cursorState.cursorId)) return;
self.next(callback);

@@ -224,16 +224,16 @@ }, 1);

// If we don't have a cursorId execute the first query
if(cursorId == null) {
if(cursorState.cursorId == null) {
execInitialQuery(query, function(err, r) {
if(err) return handleCallback(callback, err, null);
if(documents.length == 0) return handleCallback(callback, null, null);
if(cursorState.documents.length == 0) return handleCallback(callback, null, null);
self.next(callback);
});
} else if(documents.length == 0 && !Long.ZERO.equals(cursorId)) {
} else if(cursorState.documents.length == 0 && !Long.ZERO.equals(cursorState.cursorId)) {
execGetMore(function(err, doc) {
if(err) return handleCallback(callback, err);
if(documents.length == 0 && Long.ZERO.equals(cursorId)) dead = true;
if(cursorState.documents.length == 0 && Long.ZERO.equals(cursorState.cursorId)) cursorState.dead = true;
// Tailable cursor getMore result, notify owner about it
// No attempt is made here to retry, this is left to the user of the
// core module to handle to keep core simple
if(documents.length == 0 && options.tailable) {
if(cursorState.documents.length == 0 && options.tailable) {
return handleCallback(callback, MongoError.create({

@@ -246,5 +246,5 @@ message: "No more documents in tailed cursor"

if(limit > 0 && currentLimit >= limit) {
dead = true;
documents = [];
if(cursorState.limit > 0 && currentLimit >= cursorState.limit) {
cursorState.dead = true;
cursorState.documents = [];
return handleCallback(callback, null, null);

@@ -255,3 +255,3 @@ }

});
} else if(documents.length == 0 && options.tailable) {
} else if(cursorState.documents.length == 0 && options.tailable) {
return handleCallback(callback, MongoError.create({

@@ -262,8 +262,8 @@ message: "No more documents in tailed cursor"

}));
} else if(documents.length == 0 && Long.ZERO.equals(cursorId)) {
dead = true;
} else if(cursorState.documents.length == 0 && Long.ZERO.equals(cursorState.cursorId)) {
cursorState.dead = true;
handleCallback(callback, null, null);
} else {
if(limit > 0 && currentLimit >= limit) {
dead = true;
if(cursorState.limit > 0 && currentLimit >= cursorState.limit) {
cursorState.dead = true;
return handleCallback(callback, null, null);

@@ -273,3 +273,3 @@ }

currentLimit += 1;
handleCallback(callback, null, documents.shift());
handleCallback(callback, null, cursorState.documents.shift());
}

@@ -284,3 +284,3 @@ }

this.isDead = function() {
return dead == true;
return cursorState.dead == true;
}

@@ -294,3 +294,3 @@

this.bufferedCount = function() {
return documents.length;
return cursorState.documents.length;
}

@@ -304,4 +304,4 @@

this.readBufferedDocuments = function(number) {
var length = number < documents.length ? number : documents.length;
var elements = documents.splice(0, length);
var length = number < cursorState.documents.length ? number : cursorState.documents.length;
var elements = cursorState.documents.splice(0, length);
currentLimit = currentLimit + length;

@@ -317,4 +317,4 @@ return elements;

this.rewind = function() {
if(init) {
if(!dead) {
if(cursorState.init) {
if(!cursorState.dead) {
this.kill();

@@ -324,7 +324,7 @@ }

currentLimit = 0;
init = false;
dead = false;
killed = false;
documents = [];
cursorId = null;
cursorState.init = false;
cursorState.dead = false;
cursorState.killed = false;
cursorState.documents = [];
cursorState.cursorId = null;
}

@@ -340,8 +340,9 @@ }

// Set cursor to dead
dead = true;
killed = true;
cursorState.dead = true;
cursorState.killed = true;
// Remove documents
documents = [];
cursorState.documents = [];
// If no cursor id just return
if(cursorId == null || cursorId.isZero()) {
if(cursorState.cursorId == null || cursorState.cursorId.isZero() || cursorState.init == false) {
if(callback) callback(null, null);

@@ -351,13 +352,6 @@ return;

// Create a kill cursor command
var killCursor = new KillCursor(bson, [cursorId]);
// Execute the kill cursor command
if(connection && connection.isConnected()) connection.write(killCursor);
// Set cursor to 0
cursorId = Long.ZERO;
// Return to caller
if(callback) callback(null, null);
// Execute command
server.wireProtocolHandler.killCursor(bson, cursorState.cursorId, connection, callback);
}
//

@@ -367,23 +361,6 @@ // Execute getMore command

if(logger.isDebug()) logger.debug(f("schedule getMore call for query [%s]", JSON.stringify(query)))
// Create getMore command
var getMore = new GetMore(bson, ns, cursorId, {numberToReturn: batchSize});
// Query callback
var queryCallback = function(err, r) {
if(err) return callback(err);
documents = r.documents;
cursorId = r.cursorId;
// Return
callback(null);
}
// If we have a raw query decorate the function
if(options.raw || cmd.raw) {
queryCallback.raw = options.raw || cmd.raw;
}
// Register a callback
callbacks.register(getMore.requestId, queryCallback);
// Write out the getMore command
connection.write(getMore);
// Determine if it's a raw query
var raw = options.raw || cmd.raw;
// We have a wire protocol handler
server.wireProtocolHandler.getMore(bson, ns, cursorState, cursorState.batchSize, raw, connection, callbacks, options, callback);
}

@@ -414,6 +391,6 @@

// Promote id to long if needed
cursorId = typeof id == 'number' ? Long.fromNumber(id) : id;
cursorState.cursorId = typeof id == 'number' ? Long.fromNumber(id) : id;
// If we have a firstBatch set it
if(Array.isArray(result.documents[0].cursor.firstBatch)) {
documents = result.documents[0].cursor.firstBatch;
cursorState.documents = result.documents[0].cursor.firstBatch;
}

@@ -426,4 +403,4 @@

if(Array.isArray(result.documents[0].result)) {
documents = result.documents[0].result;
cursorId = Long.ZERO;
cursorState.documents = result.documents[0].result;
cursorState.cursorId = Long.ZERO;
return callback(null, null);

@@ -434,4 +411,4 @@ }

// Otherwise fall back to regular find path
cursorId = result.cursorId;
documents = result.documents;
cursorState.cursorId = result.cursorId;
cursorState.documents = result.documents;
callback(null, null);

@@ -451,132 +428,4 @@ }

}
//
// Execute a find command
var setupClassicFind = function(ns, cmd, topology, options) {
var readPreference = options.readPreference || new ReadPreference('primary');
if(typeof readPreference == 'string') readPreference = new ReadPreference(readPreference);
if(!(readPreference instanceof ReadPreference)) throw new MongoError('readPreference must be a ReadPreference instance');
// Ensure we have at least some options
options = options || {};
// Set the optional batchSize
batchSize = cmd.batchSize || batchSize;
var numberToReturn = 0;
// Unpack the limit and batchSize values
if(limit == 0) {
numberToReturn = batchSize;
} else if(limit < 0 || limit < batchSize || (limit > 0 && batchSize == 0)) {
numberToReturn = limit;
} else {
numberToReturn = batchSize;
}
var numberToSkip = skip || 0;
// Build actual find command
var findCmd = {};
// Using special modifier
var usesSpecialModifier = false;
// We have a Mongos topology, check if we need to add a readPreference
if(topology.type == 'mongos' && readPreference) {
findCmd['$readPreference'] = readPreference.toJSON();
usesSpecialModifier = true;
}
// Add special modifiers to the query
if(cmd.sort) findCmd['orderby'] = cmd.sort, usesSpecialModifier = true;
if(cmd.hint) findCmd['$hint'] = cmd.hint, usesSpecialModifier = true;
if(cmd.snapshot) findCmd['$snapshot'] = cmd.snapshot, usesSpecialModifier = true;
if(cmd.returnKey) findCmd['$returnKey'] = cmd.returnKey, usesSpecialModifier = true;
if(cmd.maxScan) findCmd['$maxScan'] = cmd.maxScan, usesSpecialModifier = true;
if(cmd.min) findCmd['$min'] = cmd.min, usesSpecialModifier = true;
if(cmd.max) findCmd['$max'] = cmd.max, usesSpecialModifier = true;
if(cmd.showDiskLoc) findCmd['$showDiskLoc'] = cmd.showDiskLoc, usesSpecialModifier = true;
if(cmd.comment) findCmd['$comment'] = cmd.comment, usesSpecialModifier = true;
if(cmd.maxTimeMS) findCmd['$maxTimeMS'] = cmd.maxTimeMS, usesSpecialModifier = true;
// If we have explain, return a single document and close cursor
if(cmd.explain) {
numberToReturn = -1;
usesSpecialModifier = true;
findCmd['$explain'] = true;
}
// If we have a special modifier
if(usesSpecialModifier) {
findCmd['$query'] = cmd.query;
} else {
findCmd = cmd.query;
}
// Build Query object
var query = new Query(bson, ns, findCmd, {
numberToSkip: numberToSkip, numberToReturn: numberToReturn
, checkKeys: false, returnFieldSelector: cmd.fields
});
// Set query flags
query.slaveOk = readPreference.slaveOk();
// Set up the option bits for wire protocol
if(options.tailable) { query.tailable = options.tailable; }
if(options.oplogReply)query.oplogReply = options.oplogReply;
if(options.noCursorTimeout) query.noCursorTimeout = options.noCursorTimeout;
if(options.awaitData) query.awaitData = options.awaitData;
if(options.exhaust) query.exhaust = options.exhaust;
if(options.partial) query.partial = options.partial;
// Return the query
return query;
}
//
// Set up a command cursor
var setupCommand = function(ns, cmd, topology, options) {
var readPreference = options.readPreference || new ReadPreference('primary');
if(typeof readPreference == 'string') readPreference = new ReadPreference(readPreference);
if(!(readPreference instanceof ReadPreference)) throw new MongoError('readPreference must be a ReadPreference instance');
// Set empty options object
options = options || {}
// Final query
var finalCmd = {};
for(var name in cmd) {
finalCmd[name] = cmd[name];
}
// Build command namespace
var parts = ns.split(/\./);
// Remove namespace db
parts.pop()
// Add command for initial execution
parts.push("$cmd");
// We have a Mongos topology, check if we need to add a readPreference
if(topology.type == 'mongos' && readPreference) {
finalCmd['$readPreference'] = readPreference.toJSON();
}
// Build Query object
var query = new Query(bson, parts.join("."), finalCmd, {
numberToSkip: 0, numberToReturn: -1
, checkKeys: false
});
// Set query flags
query.slaveOk = readPreference.slaveOk();
// Options
if(options.tailable) query.tailable = options.tailable;
if(options.oplogReply)query.oplogReply = options.oplogReply;
if(options.noCursorTimeout) query.noCursorTimeout = options.noCursorTimeout;
if(options.awaitdata) query.awaitdata = options.awaitdata;
if(options.exhaust) query.exhaust = options.exhaust;
if(options.partial) query.partial = options.partial;
// Return the query
return query;
}
}
module.exports = Cursor;

83

lib/topologies/server.js

@@ -15,3 +15,4 @@ var inherits = require('util').inherits

, BSON = require('bson').native().BSON
, LegacySupport = require('../legacy/legacy_support')
, PreTwoSixWireProtocolSupport = require('../wireprotocol/2_4_support')
, TwoSixWireProtocolSupport = require('../wireprotocol/2_6_support')
, Session = require('./session')

@@ -23,3 +24,4 @@ , Logger = require('../connection/logger')

, GSSAPI = require('../auth/gssapi')
, SSPI = require('../auth/sspi');
, SSPI = require('../auth/sspi')
, ScramSHA1 = require('../auth/scram');

@@ -172,5 +174,5 @@ // All bson types

//
// Fallback methods
// wireProtocolHandler methods
//
var fallback = options.fallback || new LegacySupport();
var wireProtocolHandler = options.wireProtocolHandler || new PreTwoSixWireProtocolSupport();

@@ -205,2 +207,3 @@ //

getProperty(this, 'bson', 'bson', options, {});
getProperty(this, 'wireProtocolHandler', 'wireProtocolHandler', options, {});
getSingleProperty(this, 'id', id);

@@ -285,2 +288,14 @@

//
// createWireProtocolHandler
var createWireProtocolHandler = function(result) {
// 2.6 wire protocol handler
if(result && result.maxWireVersion >= 2) {
return new TwoSixWireProtocolSupport();
}
// 2.4 or earlier wire protocol handler
return new PreTwoSixWireProtocolSupport();
}
//
// Handlers

@@ -377,2 +392,3 @@ var messageHandler = function(response, connection) {

// Set the current ismaster
if(!err) {

@@ -382,6 +398,13 @@ ismaster = r.result;

// Determine the wire protocol handler
wireProtocolHandler = createWireProtocolHandler(ismaster);
// Set the wireProtocolHandler
options.wireProtocolHandler = wireProtocolHandler;
// Log the ismaster if available
if(logger.isInfo()) logger.info(f('server %s connected with ismaster [%s]', self.name, JSON.stringify(r.result)));
// Validate if we it's a server we can connect to
if(!supportsServer() && fallback == null) {
if(!supportsServer() && wireProtocolHandler == null) {
state = DISCONNECTED

@@ -496,34 +519,2 @@ return self.emit('error', new MongoError("non supported server version"), self);

//
// Execute a write operation
var executeWrite = function(self, type, opsField, ns, ops, options, callback) {
if(ops.length == 0) throw new MongoError("insert must contain at least one document");
if(typeof options == 'function') {
callback = options;
options = {};
}
// Split the ns up to get db and collection
var p = ns.split(".");
var d = p.shift();
// Options
var ordered = typeof options.ordered == 'boolean' ? options.ordered : true;
var writeConcern = options.writeConcern || {};
// return skeleton
var writeCommand = {};
writeCommand[type] = p.join('.');
writeCommand[opsField] = ops;
writeCommand.ordered = ordered;
writeCommand.writeConcern = writeConcern;
// Options object
var opts = {};
if(type == 'insert') opts.checkKeys = true;
// Ensure we support serialization of functions
if(options.serializeFunctions) opts.serializeFunctions = options.serializeFunctions;
// Execute command
self.command(f("%s.$cmd", d), writeCommand, opts, callback);
}
//
// Execute readPreference Strategies

@@ -662,3 +653,5 @@ var notifyStrategies = function(op, params, callback) {

if(error) return callback(MongoError.create(error));
callback(null, new CommandResult(result.documents[0], connections));
// Execute callback, catch and rethrow if needed
try { callback(null, new CommandResult(result.documents[0], connections)); }
catch(err) { process.nextTick(function() { throw err}); }
}

@@ -687,3 +680,5 @@ });

|| result.documents[0]['code']) return callback(MongoError.create(result.documents[0]));
callback(null, new CommandResult(result.documents[0], connection));
// Execute callback, catch and rethrow if needed
try { callback(null, new CommandResult(result.documents[0], connection)); }
catch(err) { process.nextTick(function() { throw err}); }
});

@@ -712,4 +707,3 @@ }

// Execute write
if(fallback && (ismaster.maxWireVersion == null || ismaster.maxWireVersion == 0)) return fallback.insert(ismaster, ns, bson, pool, callbacks, ops, options, callback);
executeWrite(this, 'insert', 'documents', ns, ops, options, callback);
return wireProtocolHandler.insert(self, ismaster, ns, bson, pool, callbacks, ops, options, callback);
}

@@ -737,4 +731,3 @@

// Execute write
if(fallback && (ismaster.maxWireVersion == null || ismaster.maxWireVersion == 0)) return fallback.update(ismaster, ns, bson, pool, callbacks, ops, options, callback);
executeWrite(this, 'update', 'updates', ns, ops, options, callback);
return wireProtocolHandler.update(self, ismaster, ns, bson, pool, callbacks, ops, options, callback);
}

@@ -762,4 +755,3 @@

// Execute write
if(fallback && (ismaster.maxWireVersion == null || ismaster.maxWireVersion == 0)) return fallback.remove(ismaster, ns, bson, pool, callbacks, ops, options, callback);
executeWrite(this, 'delete', 'deletes', ns, ops, options, callback);
return wireProtocolHandler.remove(self, ismaster, ns, bson, pool, callbacks, ops, options, callback);
}

@@ -960,2 +952,3 @@

this.addAuthProvider('sspi', new SSPI());
this.addAuthProvider('scram-sha-1', new ScramSHA1());
}

@@ -962,0 +955,0 @@

2

package.json
{
"name": "mongodb-core",
"version": "1.0.0-alpha7",
"version": "1.0.0-alpha8",
"description": "Core MongoDB driver functionality, no bells and whistles and meant for integration not end applications",

@@ -5,0 +5,0 @@ "main": "index.js",

lib/legacy/commands.js
lib/legacy/legacy_support.js
SocketSocket SOC 2 Logo

Product

  • Package Alerts
  • Integrations
  • Docs
  • Pricing
  • FAQ
  • Roadmap
  • Changelog

About

Packages

npm

Go

Maven

PyPI

Rubygems

Stay in touch

Get open source security insights delivered straight into your inbox.

  • Terms
  • Privacy
  • Security

Made with ⚡️ by Socket Inc