Installation =>npm install oplog.rx

Main features

How the Oplog works

{"ts":"6533791416483577857","t":4,"h":"8859258976700926266","v":2,"op":"i","ns":"test.foo","o":{"_id":"5ab94bb","username":"fox"}}

Here is a handy table talking about the fields in an oplog doc:

Basic Usage

import{ObservableOplog}from'oplog.rx';constoplog=newObservableOplog();oplog.tail().then(function(){console.log('successfully started tailing the oplog.');});oplog.getEmitter().on('update',function(){}).on('insert',function(){}).on('delete',function(){});

Useful Query/Filter options

import{ObservableOplog}from'oplog.rx';import{Timestamp}from'bson';constoplog=newObservableOplog({ts:Timestamp.fromInt(Date.now()-45000),// search for documents that are younger than 45 seconds agons:{$in:['mydb.coll1','mydb.coll2',/mydb2\.*/],}});// or if you need something very custom, use query:constoplog=newObservableOplog({query:{ts:{$gt:Timestamp.fromInt(Date.now()-45000)},$and:[{ns:{$nin:[/foo/,/rolo/]}},{ns:{$in:[/bar/]}},]}});// if the query parameter is provided, it will be used directly to search the oplog.rs collection:// like so:constcoll=db.collection('oplog.rs');constcursor=coll.find(query);

Usage with RxJS Observables

constoplog=newObservableOplog();oplog.tail().then(function(){console.log('successfully started tailing the oplog.');});constops=oplog.getOps();ops.insert.subscribe(v=>{});ops.delete.subscribe(v=>{});ops.update.subscribe(v=>{});// or use:const{insert, update, del}=oplog.getOps();

Usage with Node.js Streams

constoplog=newObservableOplog();oplog.tail().then(function(){console.log('successfully started tailing the oplog.');});// create a transform stream which only forwards the desired dataconstt=oplog.getFilteredStream({namespace:'foobar'});// the above stream is a transform stream which you can pipe elsewhere// to send the data to another process, convert it to JSON firstconstJSONStdio=require('json-stdio');consttransform=JSONStdio.transformObject2JSON();constsocket=getClientConnection();// get a tcp connection from wherevert.pipe(transform).pipe(socket);// with the above code, you can listen for certain events// and pipe the data to wherever it needs to go// streams are especially useful for performant networking between processes.

Advanced usage - Client receives stream data

importnet= require('net');importJSONStdio= require('json-stdio');import{getOplogStreamInterpreter}from'oplog.rx';constc=net.createConnection(6969,'localhost');constjsonParser=JSONStdio.createParser();conststrm=c.pipe(jsonParser);// parse the JSON stream into JS objectsconst{ops, emitter}=getOplogStreamInterpreter(strm);// listen for data events// we can use observablesops.delete.subscribe(v=>{console.log('delete happened.');});ops.insert.subscribe(v=>{console.log('insert happened.');});ops.update.subscribe(v=>{console.log('update happened.');});// or just use an event emitteremitter.on('update',function(){console.log('update happened.');});emitter.on('delete',function(){console.log('delete happened.');});emitter.on('insert',function(){console.log('insert happened.');});