lib/storage/metadata/mongoclient/LogConsumer.js

@@ -1,14 +1,13 @@
 'use strict'; // eslint-disable-line
 
+const async = require('async');
 const stream = require('stream');
 const MongoClient = require('mongodb').MongoClient;
-const { Timestamp, Long } = require('bson');
-
-let lastEndID = undefined;
+const Timestamp = require('bson').Timestamp;
 
 const ops = {
-    i: 'put',
-    u: 'put',
+    i: 'insert',
+    u: 'update',
     d: 'delete',
 };
 
@@ -19,10 +18,6 @@ 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) {
@@ -31,41 +26,21 @@ 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,
-            });
         }
 
-        // don't push oplogs that have already been sent
-        if (!this.unpublishedListing) {
-            return callback();
+        if (this.end === undefined || itemObj.ts.toNumber() > this.end) {
+            this.end = itemObj.ts.toNumber();
         }
 
+        console.log('turn back into timestamp,', Timestamp.fromNumber(timestampNum));
+
         const streamObject = {
             timestamp: new Date(itemObj.ts.high_ * 1000),
             db: itemObj.ns,
@@ -73,22 +48,12 @@ class ListRecordStream extends stream.Transform {
                 {
                     type: ops[itemObj.op],
                     key: itemObj.o._id,
-                    value: JSON.stringify(itemObj.o.value),
+                    value: 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();
-    }
 }
 
 /**
@@ -172,9 +137,7 @@ class LogConsumer {
     readRecords(params, cb) {
         const recordStream = new ListRecordStream(this.logger);
         const limit = params.limit || 10000;
-        const startIDandSeq = params.startSeq.toString().split('|');
-        const startSeq = parseInt(startIDandSeq[0], 10) || 0;
-        lastEndID = startIDandSeq[1];
+        const startSeq = params.startSeq || 0;
         // need to somehow limit entries to limit and then stop the stream
 
         this.coll = this.db.collection('oplog.rs');
@@ -195,7 +158,7 @@ class LogConsumer {
                 recordStream.write(data);
             });
             stream.on('end', () => {
-                recordStream.write(undefined);
+                recordStream.write();
             });
             recordStream.once('info', info => {
                 recordStream.removeAllListeners('error');
@@ -213,7 +176,7 @@ class LogConsumer {
 // tester.connectMongo(() => {
 //     return async.waterfall([
 //         next => {
-//             return tester.readRecords({ startSeq: 0, limit: 5 }, (err, res) => {
+//             return tester.readRecords({ startSeq: 6519603494532415000, 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,10 +59,13 @@ 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,7 +3,11 @@
   "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": {
@@ -28,6 +32,7 @@
     "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",