Mongo Live Server

Description

Mongo Live Server serves live data from MongoDB over a websocket with just one simple configuration of less then 15 lines. Operations in MongoDB are served directly from a replica set oplog over an websocket API in a read-only fashion using the new MongoDB Change Streams. So, if you have a service that writes to a MongoDB database and you're in need of a service that reads those updates, Mongo Live Server is a simple and elegant solution! Instead of writing complete microservices using boilerplate code over and over again, use one module instead. Why reinvent the wheel? Also, using Mongo Live Server across multiple services provides API consistency; very convenient!

Mongo Live Server combines the following well-known modules:

• MongoDB: The official MongoDB driver for Node.js.
• WS: Simple, blazing fast, and thoroughly tested WebSocket server implementation.

At the same time, Mongo Live Server aims to provide full access to all functionalities of these dependencies.

Optionally, Mongo Live Server can also be used in combination with Mongoose, a MongoDB object modeling tool, mainly for writing database models. With one single setting Mongo Live Server can start up with either MongoDB connections or Mongoose connections, respectively relying on MongoDB collections or Mongoose models. To do so, set useMongoose setting to true and run npm i mongoose in your project.

Inspired by

We highly recommend using Express Restify Mongoose for creating database services with RESTful API's in a similar fashion. Easy to understand. Quick and painless.

Moreover, Mongo Live Server is the Express Restify Mongoose for live data!

How to install

npm i mongo-live-server

How to use

Make sure to provide the replicaSet option. MongoDB Change Streams rely on the database to be a replicaset. Note that replicasets are allowed to have one single member.

Configuration

config =
    # WS server port (optional). Default: random 
    port:                 7777
 
    mongo:
        # Database name (required). 
        database:        "people-database"
 
        # Whether to throw errors on mongo connection problems (optional). Default: false 
        throwHappy:       true
 
        # Use this to let the module init a replicaset automatically on startup 
        initReplset: true
 
        # MongoDB (or Mongoose) connection options 
        options:
            poolSize:     50   # (optional). Default: 5. Minimum: 5 
            replicaSet:  "rs0" # (required). 
 
        # Hosts array (required). 
        # Needs at least one item. 
        # Can supply multiple hosts for multiple membered replicasets. 
        hosts: [
            host: "localhost" # MongoDB host name (required). 
            port: 27017       # MongoDB port (required). 
        ]
 
        # Whether or not to use Mongoose (optional). Default: false 
        useMongoose: false
 
    # Watch configuration. (required) 
    # Which MongoDB collections / Mongoose models should updates be served from? 
    # Needs at least one item. 
    watches: [
        # path (required). 
        # Serve data for this collection/model over a socket connection 
        # Address is ws://[host][:port]/[path]/live 
        path:             "pets"
 
        # collection/model (required). 
        # Serve data for this MongoDB collection or Mongoose model. 
        # either 
        collection:       "pets"
        # or 
        model:            "Pet"
 
        # identityKey (optional). Default: "identity" 
        # The field that is used for ACL in the aggregation pipeline of the change stream. 
        # see `getAllowed` 
        identityKey:      "identity"
 
        # Non-readable fields. (optional) 
        blacklistFields: ["bankAcct"]
    ]
 
    # Header name or query parameter for sending along user id (optional). 
    # Default: "user-id" 
    userIdentityKey: "user-id"
 
    # Asynchronous function for filtering requested ids (optional) 
    # `ids` array can be a unique field like "identity" or shared field like "role". 
    # This depends on `identityKey` 
    # Example: 
    getAllowed: ({ ids, userIdentity }, cb) ->
 
        # Perform a database call -> allowedIds 
 
        # expensive filter: 
        filteredIds = ids.filter (doc) ->
            allowedIds.includes userIdentity
 
        cb null, filteredIds

Start and stop

For convenience sake and testing purposes, Mongo Live Server has start and stop functions. Both can be called synchronously ánd asynchronously, with or without a callback that is.

Moreover, if the initReplset setting is enabled, starting up Mongo Live Server will automatically initialise a replica set for the database on the provided host(s) and port(s). Obviously, this is not recommended in production.

mongoLiveServer.start() # synchronous 
 
mongoLiveServer.start (error) =>
    return cb error if error
 
    console.log "Mongo Live Server has successfully started!"
 

mongoLiveServer.stop() # synchronous 
 
mongoLiveServer.stop (error) =>
    return cb error if error
 
    console.log "Mongo Live Server has successfully stopped!"
 

Convenience variables:

Use the Mongo Live Server to access the database. The variable mongoLiveServer.db is a reference to the currert connected database, specified in config.mongo.database. The variable mongoLiveServer.db can be used for example to open collections and write documents with native MongoDB functions, like:

mongoLiveServer.db.collection("mycollection").insert

The connected native MongoDB Client can be found here" mongoLiveServer.mongoClient.

When using Mongoose and thus useMongoose is true, mongoLiveServer.mongooseConnection is the connection created with mongoose.createConnection. Use it to instantiate Mongoose models in the ordinary way, for example like this:

schema = new Schema({ name: String })
 
connection = mongoLiveServer.mongooseConnection
 
connection.model("MyCollection", schema)
 

Once that is done, use mongoLiveServer.models to access the instantiated models, for example like:

mongoLiveServer.models.MyCollection

Connecting

To recieve all data for a collection/model, connect to the configured path postfixed with "/live". The live postfix is relevant to discriminate between routes used for live-data endpoints and routes used for e.g. RESTful endpoints. This way an application can have uniform paths.

new WebSocket "ws://localhost:7777/pets/live", "user-id": "tes-user-123"

JSON messages received over the socket will, after parsing, be namespaced based on operation type. This means that if the operation type was update, then the object recieved will have a update key for which the value is the relevant document data:

{
    "update": {
        "_id": "5751636e5539480100df6c43",
        "identity": "Kofi Annan",
        "job": "diplomat"
    }
}

Messages for the insert operation result in full documents. Messages for the update operation only contain the changed fields. Messags for the delete operation only contain the _id field.

Querying

To request specific data, use a query string. To stringify it use the qs module. The query-string module is currently not supported. If demanded, please create an issue. All query parameters are optional.

qs = require "qs"
 
qs.stringify
    # Subscribe to specific operation types 
    # Let the server know what MongoDB operation types, e.g. "insert" and/or "update" and/or "delete" 
    subscribe:     ["update"]
 
    # Let the server know what fields the send documents should have. 
    fields:        ["last name", "first name", "city", "email", "role", "bankAcct"]
 
    # Only in case of an subscription to "update" operations. 
    # Messages for update operations only contain the updated fields. 
    # Therefore, let the server know what fields should in all cases be added to the message. 
    # This is needed for example for finding the relevant document in the client. 
    extension:     ["identity"]
 
    # Let the server know what ids (relates to `identityKey`) should be watched 
    ids:           []

Server Example

_                              = require "underscore"
async                          = require "async"
MongoLiveServer                = require "mongo-live-server"
manyPeople                     = require "random-people"
 
 
getAllowed = ({ ids, userIdentity }, cb) ->
    if userIdentity is "Kofi Annan"
        return cb null, [ "admin" ]
 
    cb null, [ "user", "designer" ]
 
 
mongoLiveServer = new MongoLiveServer
    port:       7777
    mongo:
        database:        "mongo-live-server-example"
        throwHappy:       true
        options:
            poolSize:     50
            replicaSet:  "rs0"
        hosts: [
            host: "localhost"
            port: 27017
        ]
    watches: [
        path:             "persons"
        collection:       "persons"
        identityKey:      "role"
        blacklistFields: ["bankAcct"]
    ]
    getAllowed:           getAllowed
 
Server = class Server
    constructor: ->
 
    start: (cb) ->
        mongoLiveServer.start (error) =>
            return cb error if error
 
            cb()
 
            @collection = mongoLiveServer.mongoConnector.db.collection "persons"
 
            @createPeople()
 
    createPeople: =>
        @interval = setInterval @createPerson, 3000
 
    createPerson: =>
        people = manyPeople 3
 
        people[0].role = "admin"
        people[1].role = "user"
        people[2].role = "designer"
 
        @collection.insert people, (error) ->
            console.error error if error
 
            console.log "Added #{people[0]["first name"]}
             & #{people[1]["first name"]}
             & #{people[2]["first name"]}"
 
    stop: (cb) =>
        clearInterval @interval
        mongoLiveServer.stop cb
 
 
server = new Server
 
server.start (error) ->
    throw error if error
 
    _.each [
        "SIGUSR2"
        "SIGTERM"
        "SIGINT"
    ], (signal) ->
        process.on signal, _.once ->
            console.info "Got signal `#{signal}`. Closing down..."
 
            server.stop (error) ->
                throw error if error
                process.exit 0
 
process.on "unhandledRejection", (error) ->
    console.error "unhandledRejection", error

Client Example

qs                   = require "qs"
WebSocket            = require "ws"
 
handleMessage = (message) ->
    try
        message = JSON.parse message
    catch error
        throw new Error "Parse error: #{error}. Message:", message
 
    console.log message
 
options =
    headers:
        "user-id": "joeri" # Change this name to test `getAllowed` ACl functionality 
 
query = qs.stringify
    subscribe: ["insert"]
    fields:    ["last name", "first name", "city", "email", "role", "bankAcct"]
 
new WebSocket "ws://localhost:7777/persons/live?#{query}", options
    .on "message", handleMessage
 

Contribute

Create an issue or a pull request.

Release

• Login with the right NPM account
• Run npm run release-test to make sure that everything is ready and correct for publication on NPM.
• If the previous step succeeded, release a new master version with a raised version tag following the Git flow method.
• Run npm run release to finally publish the module on NPM.

How to test

Run all the tests once with coverage:

npm test

Run all the tests without coverage and with a watch:

npm tests

Contact the author

Contact Pim per email at heijden.pvd[at]gmail.com