Package Exports

nats-topology-runner
nats-topology-runner/dist/index.js

This package does not declare an exports field, so the exports above have been automatically detected and optimized by JSPM instead. If any package subpath is missing, it is recommended to post an issue to the original package (nats-topology-runner) to support the "exports" field. If that is not possible, create a JSPM override to customize the exports field for this package.

Readme

NATS Topology Runner

Run a job using nats-jobs and topology-runner. Exports the function runTopologyWithNats that runs a topology and resumes the topology based on loadSnapshot. Uniqueness can be determined by using getStreamDataFromMsg to extract stream and streamSequence from the message.

runTopologyWithNats

Returns a fn that takes a JsMsg and runs the topology with the data off the message. Automatically resumes a topology if a snapshot exists with the topologyId and it was not run to completion. Regardless of whether the topology succeeds or fails, the last snapshot will be persisted and awaited.

The example below uses an in-memory database called loki. In a real-world scenario you would want to use something like MongoDB or Redis.

If you topology executes fast enough you may want to use the debounceMs option to prevent the possibility of out-of-order writes to the datastore.

import {
  runTopologyWithNats,
  getStreamDataFromMsg,
  StreamSnapshot,
  Fns,
} from 'nats-topology-runner'
import { JsMsg, JSONCodec } from 'nats'
import { expBackoff, JobDef, jobProcessor } from 'nats-jobs'
import { DAG, RunFn, Spec } from 'topology-runner'
import loki from 'lokijs'
import _ from 'lodash/fp'

const db = new loki('test.db')
const topology = db.addCollection<StreamSnapshot & { topologyId: string }>(
  'topology'
)
const scraper = db.addCollection('scraper')
const jc = JSONCodec()

const dag: DAG = {
  api: { deps: [] },
  details: { deps: ['api'] },
  attachments: { deps: ['api'] },
  writeToDB: { deps: ['details', 'attachments'] },
}

const spec: Spec = {
  nodes: {
    api: {
      run: async () => [1, 2, 3],
    },
    details: {
      run: async ({ data }) => {
        const ids: number[] = data[0]
        return ids.reduce(
          (acc, n) => _.set(n, { description: `description ${n}` }, acc),
          {}
        )
      },
    },
    attachments: {
      run: async ({ data }) => {
        const ids: number[] = data[0]
        return ids.reduce(
          (acc, n) => _.set(n, { file: `file${n}.jpg` }, acc),
          {}
        )
      },
    },
    writeToDB: {
      run: async ({ data }) => {
        scraper.insert(_.mergeAll(data))
      },
    },
  },
}
const loadSnapshot = (topologyId: string) =>
  topology.findOne({ topologyId }) || undefined

const persistSnapshot = (topologyId: string, snapshot: StreamSnapshot) => {
  const existing = topology.findOne({ topologyId })
  if (existing) {
    // Loki adds a meta field which must be stripped from the snapshot
    topology.update({ ...existing, ...stripLoki(snapshot) })
  } else {
    topology.insertOne({ ...snapshot, topologyId })
  }
}

const fns: Fns = {
  unpack: jc.decode,
  persistSnapshot,
  loadSnapshot,
}
const perform = runTopologyWithNats(spec, dag, fns)

const def: JobDef = {
  stream: 'scraper',
  backoff: expBackoff(1000),
  perform,
}
const processor = await jobProcessor()
processor.start(def)