Utility (Nodejs): Streaming Collections - Snippet Editor

Save Snippet View Snippet

"use strict"
var mongoose = require('mongoose');
//helper function for dynamically creating models
const capitalize = (s) => typeof s !== 'string' ? '' : s.charAt(0).toUpperCase() + s.slice(1);
var log = {errors:0, recordsAdded:0}
var config = {
    db:"reviews", //mongodb database name
    io:{
      input:"printer", //mongodb source collection
      output:"printertest" //mongodb destination collection
    }
}
const processDoc = (doc) => {
  return doc;
}
const saveDoc = async ({db,doc}) => {
  try{
    doc = processDoc(doc);
    //do stuff to do here before saving it...
    log.recordsAdded++
    var newDoc = new db.output(doc);
    await newDoc.save();
  }catch(e){
    log.errors++;
    log.recordsAdded > 0 ? log.recordsAdded-- : 0;
    console.log({log,error:"saveDoc"})
    log.recordsAdded == 0 ? console.log({log,msg:"Process Completed With Errors"}) : null;
  }
}
const pipeRecords = (db) => {
  console.log("DB Connected. Piping Records...")
  const pipeline = db.input.find().limit(100).lean()
    .cursor()
    .on('data', (doc) => saveDoc({db,doc}))
    .on('end', () => console.log({log}));
}
const dbInit = () => {
  var db = {};
  var buildModel = (entry) => { // collection name ** lowercase singular **
    try{
      var schema =  entry[0] ? entry[0]+"Schema" : false;
      var modelKey = entry[1] ? capitalize(entry[1]) : false;
      db[schema] = new mongoose.Schema({},{ strict: false });
      db[entry[0]] =  mongoose.model(modelKey, db[schema]);
    }catch(e){console.log(e)}
  }
  Object.entries(config.io).forEach(buildModel)
  return db;
}
//establish connection to mongoDB server
const connectDB = () => {
  var db = dbInit();
  mongoose.connect('mongodb://localhost/'+config.db, {useNewUrlParser: true});
  var dbc = mongoose.connection;
  dbc.on('error', console.error.bind(console, 'connection error:'));
  dbc.once('open', () => pipeRecords(db))
}
connectDB() //calls pipeRecords() once the database is connected