PubSub
pubsub is a wrapper around MongoDB change streams (per-Collection event listening) and Bull queues (scheduled jobs with per-collection querying) to support event subscriptions for simple or complex interactions between services
Installation
Use the package manager npm to install events.
npm install @postilion/pubsubUsage
- Import the PubSub constructor
import { PubSub } from '@postilion/pubsub';- Define subscriptions with a model, operation, handler, filters and queue options
[
{
name: 'SyncFilingsByTicker',
model: models.Company,
operation: Operation.named,
handler: filingManager.syncSecFilingFeedByTicker,
filters: [],
options: {}
},
{
name: 'GetFilingDocumentsForFiling',
model: models.Filing,
operation: Operation.create,
handler: filingManager.getDocumentsForFiling,
filters: [
{
$match: {
status: 'unseeded'
}
}
],
options: {}
}
]- Open a connection with a mongodb client
const mongoose = require('mongoose');
mongoose.connect('mongodb://localhost/db');-
Make sure that your redis instance is running
-
Create a new instance of
PubSubwithSubscriptionsandPubSubOptions
const pubsubOptions: PubSubOptions = {
redis: 'redis://localhost:6379',
mongodb: 'mongodb://localhost:27017'
}
const pubsub = new PubSub(subscriptions, pubsubOptions);- Change a document in a collection you've created a subscription for
db.filing.insert({
_id: "5e06528b29734aab3823235d",
status: "unseeded",
company: "5e065276aeee4f3833517b6b",
publishedAt: "2019-02-01T01:22:40.000Z",
fiscalYearEnd: "1231-01-01T00:00:00.000Z",
source: "sec",
type: "10-K",
refId: "0001018724-19-000004",
period: "2018-12-31T00:00:00.000Z",
url: "https://www.sec.gov/Archives/edgar/data/1018724/000101872419000004/0001018724-19-000004-index.htm",
name: "Form 10-K",
filedAt: "2019-02-01T00:00:00.000Z",
acceptedAt: "2019-02-01T04:22:40.000Z",
__v: 0
})- The handler attached to each matching subscription should receive a Job that matches the given filters and contains a Document of the
Modelspecificed in theSubscription.modelfield
{
id: "cde20f20-28d9-11ea-9735-99b5e82d5a99",
name: "GetFilingDocumentsForFiling",
operation: "insert",
data: Job {
_id: "5e06528b29734aab3823235d",
status: "unseeded",
company: "5e065276aeee4f3833517b6b",
publishedAt: "2019-02-01T01:22:40.000Z",
fiscalYearEnd: "1231-01-01T00:00:00.000Z",
source: "sec",
type: "10-K",
refId: "0001018724-19-000004",
period: "2018-12-31T00:00:00.000Z",
url: "https://www.sec.gov/Archives/edgar/data/1018724/000101872419000004/0001018724-19-000004-index.htm",
name: "Form 10-K",
filedAt: "2019-02-01T00:00:00.000Z",
acceptedAt: "2019-02-01T04:22:40.000Z",
__v: 0
}
}Contributing
Pull requests are welcome. For major changes, please open an issue first to discuss what you would like to change.