Stream Uploads to MongoDB GridStore. Node server side async.parallel image processing using: busboy, gridfs-stream, async, sharp

files.coffee

# express file upload controller

#Mongoose configuration.
mongoose              = require "mongoose"
Busboy                = require "busboy"
secrets               = require "../config/secrets"
GridStream            = require "gridfs-stream"
Files                 = require "../models/files"
sharp                 = require "sharp"
async                 = require "async"


mongoose.connect secrets.db

# best practices for mongoose start/stop events
mongoose.connection.on "error", (err)->
  console.log "Mongoose Driver: Connection Error #{err}."

mongoose.connection.on "connected", ->
  console.log "Mongoose Driver: connected!."

mongoose.connection.on "disconnected", ->
  console.log "Mongoose Driver: disconnected!"

process.on 'SIGINT', ->
  mongoose.connection.close ->
    console.log "Mongoose Driver: connection closed on SIGINT app shutdown."
  process.exit 0




# this handles an http POST; the body contains a JS object
exports.postDeleteFileMap = (req, res)->
  return res.status(401).json() unless req.user

  clientFile = req.body # sent from client
  # remove files from mongodb
  Files.deleteFileMap clientFile, (err, deletedFile) ->  # this deletes from MongoDB store
    return res.status(503).json() if err
    req.user.removeFile deletedFile, (err, file)->    # this removes from the user files []
      return res.status(503).json() if err
      res.status(200).json(deleted: true)


# GET a file by mongo gridFS id; fileapi/:id
exports.getFileBlob = (req, res)->
  return res.status(401).json() unless req.user

  Files.streamFileById req.params.id, (err, stream)->
    res.status(404).json() if err

    # double check for stream errors
    stream.on 'error', (err)->
      res.status(404).json()

    stream.pipe(res)
    # note: we could transform the files 'on the fly'
    # like this!!!  very handy
    #transformer = sharp()
    #  .resize 320,180
    #  .crop sharp.gravity.north
    #stream.pipe(transformer).pipe(res)


# get files by user (NOTE: NOT USED, deprecated)
exports.getFilesByUser = (req, res) ->
  return res.status 401
    .json status: 'Unauthorized' unless req.user

  Files.findByUser req.user._id, (err, files) ->
    return res.status 500
      .json status: err if err
    res.status 200
      .json files


# POST files
# process the uploaded file with nodeJS stream module == Busboy
# pass this file stream from busboy to MongoDB GridFS using mongoose 
# gridfs-stream 'writestream',
# Then, getting fancy! Use Async.parallel for image processing.
# save multiple resized copies: 'raw', 'desktop', 'mobile', thumb'

exports.mongodbUpload = (req, res, next) ->
  FIELDS          = ['title','tag']
  fieldValues     = {}

  gridStream = new GridStream(mongoose.connection.db, mongoose.mongo)

  busboy = new Busboy
    headers: req.headers
    limits:
      fileSize:   1024 * 1024 * 15
      files:      1

  #busboy field event
  busboy.on 'field', (fieldname, val, fieldNameTrunc, valTrunc) ->
    if fieldname in FIELDS
      fieldValues[fieldname] = val

  #busboy file event
  busboy.on "file", (fieldname, stream, filename, encoding, mimetype) ->
    async.parallel
      raw: (done) ->
        writestream = gridStream.createWriteStream
          mode:           "w"
          filename:       filename
          content_type:   mimetype
          metadata:
            userid:       req.user._id
            fields:       fieldValues
            encoding:     encoding
            size:         'raw'

        writestream.on 'error', (err) ->
          done err, msg: 'error'

        writestream.on "close", (file) ->
          done null, file

        stream.pipe writestream

      desktop: (done) ->
        transformer = sharp()
          .resize 1920,1080
          .crop sharp.gravity.north

        writestream = gridStream.createWriteStream
          mode:           "w"
          filename:       filename
          content_type:   mimetype
          metadata:
            userid:       req.user._id
            fields:       fieldValues
            encoding:     encoding
            size:         'desktop'

        writestream.on 'error', (err) ->
          done err, msg: 'error'

        writestream.on "close", (file) ->
          done null, file

        stream.pipe(transformer).pipe(writestream) 

      tablet: (done) ->
        transformer = sharp()
          .resize 1024,768
          .crop sharp.gravity.north

        writestream = gridStream.createWriteStream
          mode:           "w"
          filename:       filename
          content_type:   mimetype
          metadata:
            userid:       req.user._id
            fields:       fieldValues
            encoding:     encoding
            size:         'tablet'

        writestream.on 'error', (err) ->
          done err, msg: 'error'

        writestream.on "close", (file) ->
          done null, file

        stream.pipe(transformer).pipe(writestream) 

      mobile: (done) ->
        transformer = sharp()
          .resize 480,320
          .crop sharp.gravity.north

        writestream = gridStream.createWriteStream
          mode:           "w"
          filename:       filename
          content_type:   mimetype
          metadata:
            userid:       req.user._id
            fields:       fieldValues
            encoding:     encoding
            size:         'mobile'

        writestream.on 'error', (err) ->
          done err, msg: 'error'

        writestream.on "close", (file) ->
          done null, file

        stream.pipe(transformer).pipe(writestream) 

      thumb: (done) ->
        transformer = sharp()
          .resize 64,64
          .crop sharp.gravity.north

        writestream = gridStream.createWriteStream
          mode:           "w"
          filename:       filename
          content_type:   mimetype
          metadata:
            userid:       req.user._id
            fields:       fieldValues
            encoding:     encoding
            size:         'thumb'

        writestream.on 'error', (err) ->
          done err, msg: 'error'

        writestream.on "close", (file) ->
          done null, file

        stream.pipe(transformer).pipe(writestream) 

    , (err, result) ->
      return res.status(500).json() if err
      # transMutateReconfig into our fileMap
      fileMap =
        tag:        result.raw.metadata.fields.tag
        title:      result.raw.metadata.fields.title
        raw:        result.raw._id
        desktop:    result.desktop._id
        tablet:     result.tablet._id
        mobile:     result.mobile._id
        thumb:      result.thumb._id

      # addFile will add the mongo _id via result  
      req.user.addFile fileMap, (err, result) ->
        return res.status(503).json() if err
        #created 201
        res.status(201).json(newFile: result) # client will newFile prop with


  busboy.on "finish", -> # returns async
    req.user.fileCacheTime = Date.now()

  busboy.on "error", (err)-> # returns async

  req.pipe busboy

Thanks! Finally I understood the streaming! From busboy to sharp