remove unused module

lib/storage/metadata/mongoclient/LogConsumer.js

@@ -1,13 +1,14 @@
 'use strict'; // eslint-disable-line
 
-const async = require('async');
 const stream = require('stream');
 const MongoClient = require('mongodb').MongoClient;
-const Timestamp = require('bson').Timestamp;
+const { Timestamp, Long } = require('bson');
+
+let lastEndID = undefined;
 
 const ops = {
-    i: 'insert',
-    u: 'update',
+    i: 'put',
+    u: 'put',
     d: 'delete',
 };
 
@@ -18,6 +19,10 @@ class ListRecordStream extends stream.Transform {
         this.hasStarted = false;
         this.start = undefined;
         this.end = undefined;
+        this.lastUniqID = undefined;
+        // this.unpublishedListing is true once we pass the oplog that has the
+        // start seq timestamp and uniqID 'h'
+        this.unpublishedListing = undefined;
     }
 
     _transform(itemObj, encoding, callback) {
@@ -26,21 +31,41 @@ class ListRecordStream extends stream.Transform {
             this.emit('info', {
                 start: this.start,
                 end: this.end,
+                uniqID: this.lastUniqID,
             });
             return callback();
         }
 
+        // always update to most recent uniqID
+        this.lastUniqID = itemObj.h.toString();
+
+        if (this.end === undefined || itemObj.ts.toNumber() > this.end) {
+            this.end = itemObj.ts.toNumber();
+        }
+
+        // only push to stream unpublished objects
+        if (!this.unpublishedListing) {
+            if (lastEndID === itemObj.h.toString()) {
+                this.unpublishedListing = true;
+            }
+            return callback();
+        }
+
         if (!this.hasStarted) {
             this.hasStarted = true;
             this.start = itemObj.ts.toNumber();
+            this.emit('info', {
+                start: this.start,
+                end: this.end,
+                uniqId: this.lastUniqID,
+            });
         }
 
-        if (this.end === undefined || itemObj.ts.toNumber() > this.end) {
-            this.end = itemObj.ts.toNumber();
+        // don't push oplogs that have already been sent
+        if (!this.unpublishedListing) {
+            return callback();
         }
 
-        console.log('turn back into timestamp,', Timestamp.fromNumber(timestampNum));
-
         const streamObject = {
             timestamp: new Date(itemObj.ts.high_ * 1000),
             db: itemObj.ns,
@@ -48,12 +73,22 @@ class ListRecordStream extends stream.Transform {
                 {
                     type: ops[itemObj.op],
                     key: itemObj.o._id,
-                    value: itemObj.o.value,
+                    value: JSON.stringify(itemObj.o.value),
                 },
             ],
         };
         return callback(null, streamObject);
     }
+
+    _flush(callback) {
+        this.emit('info', {
+            start: this.start,
+            end: this.end,
+            uniqID: this.lastUniqID,
+        });
+        this.push(null);
+        callback();
+    }
 }
 
 /**
@@ -137,7 +172,9 @@ class LogConsumer {
     readRecords(params, cb) {
         const recordStream = new ListRecordStream(this.logger);
         const limit = params.limit || 10000;
-        const startSeq = params.startSeq || 0;
+        const startIDandSeq = params.startSeq.toString().split('|');
+        const startSeq = parseInt(startIDandSeq[0], 10) || 0;
+        lastEndID = startIDandSeq[1];
         // need to somehow limit entries to limit and then stop the stream
 
         this.coll = this.db.collection('oplog.rs');
@@ -158,7 +195,7 @@ class LogConsumer {
                 recordStream.write(data);
             });
             stream.on('end', () => {
-                recordStream.write();
+                recordStream.write(undefined);
             });
             recordStream.once('info', info => {
                 recordStream.removeAllListeners('error');
@@ -176,7 +213,7 @@ class LogConsumer {
 // tester.connectMongo(() => {
 //     return async.waterfall([
 //         next => {
-//             return tester.readRecords({ startSeq: 6519603494532415000, limit: 5 }, (err, res) => {
+//             return tester.readRecords({ startSeq: 0, limit: 5 }, (err, res) => {
 //                 console.log('cb called for read records');
 //                 res.log.on('data', data => {
 //                    console.log('Streamed formatted data', data);

lib/storage/metadata/mongoclient/readStream.js

@@ -59,13 +59,10 @@ class MongoReadStream extends Readable {
             delete query._id;
         }
 
-<<<<<<< 4193394340acb6f242c036548a2860cf3766b84e
         if (searchOptions) {
             Object.assign(query, searchOptions);
         }
 
-=======
->>>>>>> Move mongoclient from s3 to arsenal
         this._cursor = c.find(query).sort({
             _id: options.reverse ? -1 : 1,
         });

package.json

@@ -3,11 +3,7 @@
   "engines": {
     "node": "6.9.5"
   },
-<<<<<<< 703aafeb723c77937798b9f0f6dcc89014a8fd39
   "version": "7.4.0",
-=======
-  "version": "7.2.2",
->>>>>>> DROP ME: bump arsenal version
   "description": "Common utilities for the S3 project components",
   "main": "index.js",
   "repository": {
@@ -32,7 +28,6 @@
     "level": "~1.6.0",
     "level-sublevel": "~6.6.1",
     "mongodb": "^3.0.1",
-    "bson": "^1.0.4",
     "node-forge": "^0.7.1",
     "simple-glob": "^0.1",
     "socket.io": "~1.7.3",