Pub/Sub Module

Abstract

This module provides an easy-to use abstraction to send and receive events with attached payloads, both locally and remotely:

• Locally (namely from a single container), it can be used as a clean replacement for traditional callback or promises-based mechanisms for long-running operations.
• Remotely (across containers), it can be used to send messages, such as triggering a long-running operation as an independent job.

Features

When an event is triggered, several behaviors could be activated without any additional code based on its specification:

• Payload validation: Validate events payloads using a Joi schema
• Logging: Log the event using our logging module
• Telemetry: Track the event in telemetry
• Audit trail: Make the event audit trail accessible to clients
• Organization webhooks: Make the event accessible to webhook subscriptions by clients
• System webhook: Relay the event as a webhook for internal use
• Queuing: Push the event to a queue for remote consumption by a queue consumer
• Scheduling: Schedule the triggering of an event using a cron tab

How is it used today?

• Everywhere in the app to log, track and trigger webhooks based on events.
• The jobs service uses the events specifications to instanciate the required queues consumers that are processing messages asynchronously. See src/services/jobs/jobs.js.
• The scheduler service uses the events specifications to trigger some events using their defined cron tab. See src/services/scheduler/scheduler.js.

Major dependencies

This module relies on the following modules:

• events: Standard node.js events emitter, used for "local" communication.
• bull: Queue package handling distributed jobs and messages in NodeJS, used for "remote" communication. See: https://github.com/OptimalBits/bull

🚀 Get Started

Register events

Before using an event (emitting or listening to), it has to be properly registered and its characteristics and behaviors specified.

Conventions:

• Events specifications should be exported as default from a module located in the specifications directory.
• Each module represents a different logical category of events.
• Events specifications are loaded dynamically and their schema validated at runtime. See events-specifications.js.
• Events codes MUST be snail_cased.

For more details, the events specification schema is defined as the eventSpecificationSchema constant in events-specifications.js.

Send and receive messages locally

// Import the module
const pubSub = require('@graphshieldhq/pubsub')
// Create a local event listener
pubSub.on('event_code', (payload) => { 
  console.debug(payload)
})
// Emit an event
pubSub.emit('event_code', {status: 'success'})

Send and receive messages remotely

Define the event with the queue property as true:

module.exports = {
  event_code: {
    ...
    queue: true
    ...
  }
}

Create a consumer module in the /src/services/jobs/consumers directory, using your event code as the file name:

// File: 'event_code.js'
// Import the module
const pubSub = require('@graphshieldhq/pubsub')
// Open the queue associated with your event
const queue = pubSub.queues.event_code
// Process messages
queue.process((job, done) => {
  // N.B: `job` is a job object from the `bull` module
  job.progress(1) // Update job progress
  job.log('Start') // Write a log message
  // The event payload is accessible as the job `data` property
  console.log(job.data)
  job.progress(100)
  job.log('Done')
  // Call the `done` callback function to flag the job as done, accepts an error + an optional result
  done(null, {status: 'success'})
})

Validate events payload schema

Optionally, the Pub/Sub module can validate payloads received using a Joi schema:

// Include Joi as a dependency
const Joi = require('joi')
// Use the `payload_schema` property to define a Joi validation schema
module.exports = {
  server_started: {
    name: 'Server instance started',
    description: 'When a server instance has started',
    payload_schema: Joi.object({
      loaded_services_count: Joi.number().min(0).required(),
    }),
    log: true,
    log_level: 'info',
    telemetry: true,
    audit_trail: false,
    organizations_webhook: false,
    system_webhook: false,
    queue: false,
    schedule: false
  }
}

Specify additional job options

By default, when queue is true, this module creates a bull job in the related queue by letting bull automatically generate a job id. You can use a specific job id by passing a correlation_id property to the event payload:

// Emit an event leading to a specific job id
pubSub.emit('event_code', {
  correlation_id: '19ff75f4-3bda-4615-bee2-c47d8b975684',
  status: 'success'
})

You can also specify the job priority by usaing the priority property in the pevent payload:

// Emit an event leading to a specific job priority
pubSub.emit('event_code', {
  job_priority: 5, // By convention, use a value from 1 to 10, 1 being the highest priority
  status: 'success'
})

Emit organization webhook

Organization webhooks are dispatched across clients by using their tenant id. Any payload associated with an event with organizations_webhook as true MUST be comprised of a tenant.id property with a valid tenant UUID or it will be ignored:

// Emit an event that could be relayed as a webhook for clients
pubSub.emit('event_code', {
  tenant: {
    id: 'e5bce8f5-2ba0-4baf-a6e3-7a1891b07763'
  },
  status: 'success'
})

Additional resources

• 👮 Code of conduct
• ✍️ Contributing Guide
• 🔐 Security policy
• ©️ License