diff --git a/README.md b/README.md index f358b0d..5136423 100644 --- a/README.md +++ b/README.md @@ -283,9 +283,9 @@ mongocollection.find({}) #### Mongo -# etl.mongo.insert(collection [,options]) +# etl.mongo.insert(collection [,options]) [DEPRECATED] -Inserts incoming data into the provided mongodb collection. The supplied collection can be a promise on a collection. The options are passed on to both streamz and the mongodb insert comand. By default this object doesn't push anything downstream, but it `pushResults` is set as `true` in options, the results from mongo will be pushed downstream. +Inserts incoming data into the provided mongodb collection. The supplied collection can be a promise on a collection. The options are passed on to both streamz and the mongodb insert command. By default this object doesn't push anything downstream, but it `pushResults` is set as `true` in options, the results from mongo will be pushed downstream. Example @@ -305,6 +305,47 @@ etl.file('test.csv') ``` +# etl.mongo.insertOne(collection [,options]) + +Inserts one incoming data into the provided mongodb collection. The supplied collection can be a promise on a collection. The options are passed on to both streamz and the mongodb insertOne command. By default this object doesn't push anything downstream, but it `pushResults` is set as `true` in options, the results from mongo will be pushed downstream. + +Example + +```js +// The following inserts data from a csv, one record at a time into a mongo collection + +var db = mongo.ConnectAsync('mongodb://localhost:27017/testdb'); +var collection = db.then(function(db) { + return db.collection('testcollection'); +}); + +etl.file('test.csv') + .pipe(etl.csv()) + .pipe(etl.mongo.insertOne(collection)); + +``` + +# etl.mongo.insertOne(collection [,options]) + +Inserts array of incoming data into the provided mongodb collection. The supplied collection can be a promise on a collection. The options are passed on to both streamz and the mongodb insertMany command. By default this object doesn't push anything downstream, but it `pushResults` is set as `true` in options, the results from mongo will be pushed downstream. + +Example + +```js +// The following inserts data from a csv, 10 records at a time into a mongo collection + +var db = mongo.ConnectAsync('mongodb://localhost:27017/testdb'); +var collection = db.then(function(db) { + return db.collection('testcollection'); +}); + +etl.file('test.csv') + .pipe(etl.csv()) + .pipe(etl.collect(10)) + .pipe(etl.mongo.insertMany(collection)); + +``` + # etl.mongo.update(collection [,keys] [,options]) Updates incoming data by building a `criteria` from an array of `keys` and the incoming data. Supplied collection can be a promise and results can be pushed downstream by declaring `pushResults : true`. The options are passed to mongo so defining `upsert : true` in options will ensure an upsert of the data. diff --git a/lib/mongo/index.js b/lib/mongo/index.js index f6beb52..c6ca6c8 100644 --- a/lib/mongo/index.js +++ b/lib/mongo/index.js @@ -1,5 +1,7 @@ module.exports = { insert : require('./insert'), + insertOne : require('./insertOne'), + insertMany : require('./insertMany'), update : require('./update'), bulk: require('./bulk'), upsert : function() { diff --git a/lib/mongo/insert.js b/lib/mongo/insert.js index d967b09..2abd4d7 100644 --- a/lib/mongo/insert.js +++ b/lib/mongo/insert.js @@ -1,3 +1,4 @@ + const Streamz = require('streamz'); const Promise = require('bluebird'); const util = require('util'); diff --git a/lib/mongo/insertMany.js b/lib/mongo/insertMany.js new file mode 100644 index 0000000..634c8f1 --- /dev/null +++ b/lib/mongo/insertMany.js @@ -0,0 +1,31 @@ +const Streamz = require('streamz'); +const Promise = require('bluebird'); +const util = require('util'); + +function insertMany(_c,collection,options) { + if (!(this instanceof Streamz)) + return new insertMany(_c,collection); + + if (isNaN(_c)) { + options = collection; + collection = _c; + _c = undefined; + } + + Streamz.call(this, _c, null, options); + this.collection = Promise.resolve(collection); + this.options = options || {}; +} + +util.inherits(insertMany,Streamz); + +insertMany.prototype._fn = function(d) { + return this.collection + .then(collection =>collection.insertMany(d,this.options)) + .then(d => { + if (this.options.pushResults) + return d.result; + }); +}; + +module.exports = insertMany; \ No newline at end of file diff --git a/lib/mongo/insertOne.js b/lib/mongo/insertOne.js new file mode 100644 index 0000000..afdc647 --- /dev/null +++ b/lib/mongo/insertOne.js @@ -0,0 +1,31 @@ +const Streamz = require('streamz'); +const Promise = require('bluebird'); +const util = require('util'); + +function insertOne(_c,collection,options) { + if (!(this instanceof Streamz)) + return new insertOne(_c,collection); + + if (isNaN(_c)) { + options = collection; + collection = _c; + _c = undefined; + } + + Streamz.call(this, _c, null, options); + this.collection = Promise.resolve(collection); + this.options = options || {}; +} + +util.inherits(insertOne,Streamz); + +insertOne.prototype._fn = function(d) { + return this.collection + .then(collection =>collection.insertOne(d,this.options)) + .then(d => { + if (this.options.pushResults) + return d.result; + }); +}; + +module.exports = insertOne; \ No newline at end of file diff --git a/package.json b/package.json index f3bdbfc..9090c9a 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "etl", - "version": "0.6.11", + "version": "0.6.12", "description": "Collection of stream-based components that form an ETL pipeline", "main": "index.js", "author": "Ziggy Jonsson (http://github.com/zjonsson/)", diff --git a/test/lib/mongo.js b/test/lib/mongo.js index 7dd5119..7254741 100644 --- a/test/lib/mongo.js +++ b/test/lib/mongo.js @@ -22,6 +22,8 @@ async function clear() { await Promise.all( [ db.collection("insert").deleteMany({}), + db.collection("insert-one").deleteMany({}), + db.collection("insert-many").deleteMany({}), db.collection("update-empty").deleteMany({}), db.collection("update-populated").deleteMany({}), db.collection("upsert").deleteMany({}), diff --git a/test/mongo-insert-many-test.js b/test/mongo-insert-many-test.js new file mode 100644 index 0000000..a9051b8 --- /dev/null +++ b/test/mongo-insert-many-test.js @@ -0,0 +1,58 @@ +const etl = require('../index'); +const data = require('./data'); +const {getCollection, clear} = require('./lib/mongo'); +const t = require('tap'); +const Promise = require('bluebird'); + +t.test('mongo.insertMany', async t => { + + t.teardown(() => t.end()); + + t.test('piping data into mongo.insertMany',async t => { + const collection = await getCollection('insert-many'); + const d = await data.stream() + .pipe(etl.collect(1)) + .pipe(etl.mongo.insertMany(collection,{pushResult:true})) + .promise(); + d.forEach(d => t.same(d,{ok:1,n:1},'inserts each record')); + }); + + t.test('mongo collection',async t => { + const collection = await getCollection('insert-many'); + const d = await collection.find({},{ projection: {_id:0}}).toArray(); + + t.same(d,data.data,'reveals data'); + }); + + t.test('pushResults == false and collection as promise',async t => { + const collection = await getCollection('insert-many'); + const d = await data.stream(etl.mongo.insertMany(collection)) + .pipe(etl.collect(4)) + .pipe(etl.mongo.insertMany(collection)) + .promise(); + + t.same(d,[],'returns nothing'); + }); + + t.test('error in collection', async t => { + const collection = Promise.reject({message: 'CONNECTION_ERROR'}); + collection.suppressUnhandledRejections(); + const e = await etl.toStream({test:true}) + .pipe(etl.collect(1)) + .pipe(etl.mongo.insertMany(collection)) + .promise() + .then(() => {throw 'SHOULD_ERROR';}, Object); + + t.same(e.message,'CONNECTION_ERROR','should bubble down'); + }); +}) +.then(() => clear()) +.then(() => t.end()) +.catch(e => { + if (e.message.includes('ECONNREFUSED')) + console.warn('Warning: MongoDB server not available'); + else + console.warn(e.message); +}); + + \ No newline at end of file diff --git a/test/mongo-insert-one-test.js b/test/mongo-insert-one-test.js new file mode 100644 index 0000000..8c3d78c --- /dev/null +++ b/test/mongo-insert-one-test.js @@ -0,0 +1,55 @@ +const etl = require('../index'); +const data = require('./data'); +const {getCollection, clear} = require('./lib/mongo'); +const t = require('tap'); +const Promise = require('bluebird'); + +t.test('mongo.insertOne', async t => { + + t.teardown(() => t.end()); + + t.test('piping data into mongo.insertOne',async t => { + const collection = await getCollection('insert-one'); + const d = await data.stream() + .pipe(etl.mongo.insertOne(collection,{pushResult:true})) + .promise(); + d.forEach(d => t.same(d,{ok:1,n:1},'inserts each record')); + }); + + t.test('mongo collection',async t => { + const collection = await getCollection('insert-one'); + const d = await collection.find({},{ projection: {_id:0}}).toArray(); + + t.same(d,data.data,'reveals data'); + }); + + t.test('pushResults == false and collection as promise',async t => { + const collection = await getCollection('insert-one'); + const d = await data.stream(etl.mongo.insertOne(collection)) + .pipe(etl.mongo.insertOne(collection)) + .promise(); + + t.same(d,[],'returns nothing'); + }); + + t.test('error in collection', async t => { + const collection = Promise.reject({message: 'CONNECTION_ERROR'}); + collection.suppressUnhandledRejections(); + const e = await etl.toStream({test:true}) + .pipe(etl.mongo.insertOne(collection,'_id')) + .promise() + .then(() => {throw 'SHOULD_ERROR';}, Object); + + t.same(e.message,'CONNECTION_ERROR','should bubble down'); + }); +}) +.then(() => clear()) +.then(() => t.end()) +.catch(e => { + if (e.message.includes('ECONNREFUSED')) + console.warn('Warning: MongoDB server not available'); + else + console.warn(e.message); +}); + + \ No newline at end of file diff --git a/test/mongo-insert-test.js b/test/mongo-insert-test.js index cc6aef8..269ff5f 100644 --- a/test/mongo-insert-test.js +++ b/test/mongo-insert-test.js @@ -36,7 +36,7 @@ t.test('mongo.insert', async t => { const collection = Promise.reject({message: 'CONNECTION_ERROR'}); collection.suppressUnhandledRejections(); const e = await etl.toStream({test:true}) - .pipe(etl.mongo.update(collection,'_id')) + .pipe(etl.mongo.insert(collection,'_id')) .promise() .then(() => {throw 'SHOULD_ERROR';}, Object); @@ -51,5 +51,3 @@ t.test('mongo.insert', async t => { else console.warn(e.message); }); - - \ No newline at end of file