feat: mongodb replica set support (#43)

BREAKING CHANGE: Now defaults to new mongo db client options `useNewUrlParser` and `useUnifiedTopology`. Travis changed to test last two Node versions, 12 and 14.

* feat: mongodb replica set support

* chore: add 10 to tests

* fix: lock to sourced 2.0.5

* fix: sourced bug - upgrade to fixed 2.0.7

* fix: sourced@2.0.8

* chore: update travis file

* chore: rm debug log statements

* chore: rm debug log statements

* fix: make options configurable

* fix: cherry pick changes from @debianw - PR #35

Co-authored-by: Waly <debianw@gmail.com>

Author: patrickleet

.travis.yml

@@ -1,27 +1,23 @@
 language: node_js
 node_js:
-  - 8
-  - 10
-  - 11
-
+  - '12'
+  - '14'
+cache: npm
 notifications:
   email: false
-
-before_install:
-  - npm install -g codecov
 script:
   - make test
+after_success:
+  - npm install -g codecov
   - codecov
-
 branches:
   except:
     - '/^v\d+\.\d+\.\d+$/'
-
 jobs:
   include:
     - stage: deploy
       if: branch == master && !fork
       node_js: node # pre-installed version
       script:
         - npm install -g semantic-release
-        - semantic-release
+        - semantic-release

index.js

@@ -29,12 +29,12 @@ function Repository (entityType, options) {
   };
 
   self.indices.forEach(function (index) {
-    snapshots.ensureIndex(index, error);
-    events.ensureIndex(index, error);
+    snapshots.createIndex(index, error);
+    events.createIndex(index, error);
   });
-  events.ensureIndex({ id: 1, version: 1 }, error);
-  snapshots.ensureIndex({ id: 1, version: 1 }, error);
-  snapshots.ensureIndex('snapshotVersion', error);
+  events.createIndex({ id: 1, version: 1 }, error);
+  snapshots.createIndex({ id: 1, version: 1 }, error);
+  snapshots.createIndex('snapshotVersion', error);
 
   log('initialized %s entity store', self.entityType.name);
 
@@ -165,7 +165,7 @@ Repository.prototype._commitEvents = function _commitEvents (entity, cb) {
       event[index] = entity[index];
     });
   });
-  self.events.insert(events, function (err) {
+  self.events.insertMany(events, function (err) {
     if (err) return cb(err);
     log('committed %s.events for id %s', self.entityType.name, entity.id);
     entity.newEvents = [];
@@ -195,7 +195,7 @@ Repository.prototype._commitAllEvents = function _commitEvents (entities, cb) {
 
   if (events.length === 0) return cb();
 
-  self.events.insert(events, function (err) {
+  self.events.insertMany(events, function (err) {
     if (err) return cb(err);
     log('committed %s.events for ids %j', self.entityType.name, _.map(entities, 'id'));
     entities.forEach(function (entity) {
@@ -212,7 +212,7 @@ Repository.prototype._commitSnapshots = function _commitSnapshots (entity, optio
   if (options.forceSnapshot || entity.version >= entity.snapshotVersion + self.snapshotFrequency) {
     var snapshot = entity.snapshot();
     if (snapshot && snapshot._id) delete snapshot._id; // mongo will blow up if we try to insert multiple _id keys
-    self.snapshots.insert(snapshot, function (err) {
+    self.snapshots.insertOne(snapshot, function (err) {
       if (err) return cb(err);
       log('committed %s.snapshot for id %s %j', self.entityType.name, entity.id, snapshot);
       return cb(null, entity);
@@ -239,7 +239,7 @@ Repository.prototype._commitAllSnapshots = function _commitAllSnapshots (entitie
 
   if (snapshots.length === 0) return cb();
 
-  self.snapshots.insert(snapshots, function (err) {
+  self.snapshots.insertMany(snapshots, function (err) {
     if (err) return cb(err);
     log('committed %s.snapshot for ids %s %j', self.entityType.name, _.map(entities, 'id'), snapshots);
     return cb(null, entities);

mongo.js

@@ -13,7 +13,10 @@ function Mongo () {
 
 util.inherits(Mongo, EventEmitter);
 
-Mongo.prototype.connect = function connect (mongoUrl) {
+Mongo.prototype.connect = function connect (mongoUrl, database, options = {
+  useNewUrlParser: true,
+  useUnifiedTopology: true
+}) {
   var self = this;
   return new Promise((resolve, reject) => {
     self.on('connected', (db) => {
@@ -22,13 +25,14 @@ Mongo.prototype.connect = function connect (mongoUrl) {
     self.on('error', (err) => {
       reject(err)
     })
-    MongoClient.connect(mongoUrl, function (err, client) {
+    MongoClient.connect(mongoUrl, options, function (err, client) {
       if (err) {
         log('✗ MongoDB Connection Error. Please make sure MongoDB is running: ', err);
         self.emit('error', err);
       }
       var expanded = url.parse(mongoUrl);
-      var dbName = expanded.pathname.replace('/', '');
+      // replica set url does not include db, it is passed in separately
+      var dbName = database || expanded.pathname.replace('/', '');
       self.client = client;
       var db = client.db(dbName);
       self.db = db;