agc-pipeline-node

agc-pipeline-node

Library provides pipeline element infrastructure for a cradle data pipeline

Motivation

The pipeline component designed to provide a RabbitMQ subscriber and publisher ecosystem around a transform() function and should simplify implementation of future Cradle components.

Installation

npm i --save agc-pipeline-node

Usage

// my-transform.ts
import { Action, IMessage, IResult } from 'agc-pipeline-node';

export default function(input: IMessage): IResult {
    const counter = input.data.counter;   // Simple transform function that increments a counter

    return {
        action: Action.Forward,
        target: '',     // can specify an exchange topic name
        message: {
            kind: 'myKind',
            data: {
                counter: counter + 1
            }
        }
    };
}

// start.ts
/**
 * Sub command to start a pipeline element with my-transform function
 */
import configProvider from '@vamship/config';
import loggerProvider from '@vamship/logger';
import { PipelineElement } from 'agc-pipeline-node';
import transform from '../lib/my-transform';

export const command = 'start';
export const describe = 'Run Pipeline Transform';
export const builder = {};
export const handler = (argv) => {
    const config = configProvider.getConfig();
    const logger = loggerProvider.getLogger('command:start');
    logger.trace('config', { config });

    const pipelineElement = new PipelineElement(transform, config);

    return pipelineElement.start();
};

// .myProjectNamerc
{
  "default": {
    "app": {},
    "log": {
        "level": "trace",
        "extremeLogging": false
    },
    "subscriber": {
      "hostname": "localhost",
      "port": "5672",
      "username": "user",
      "password": "password",
      "exchangeName": "inputExchange",
      "exchangeType": "topic",
      "isExchangeDurable": "true",
      "topicName": "myTopic",
      "queueName": "inputQueue",
      "isQueueDurable": "true",
      "isQueueExclusive": "false"
    },
    "publisher": {
      "hostname": "localhost",
      "port": "5672",
      "username": "user",
      "password": "password",
      "exchangeName": "outputExchange",
      "exchangeType": "topic",
      "isExchangeDurable": "true",
      "queueName": "outputQueue",
      "isQueueDurable": "true",
      "isQueueExclusive": "false"
    }
  },
  "development": {
    "app": {}
  },
  "test": {
    "app": {}
  },
  "production": {
    "app": { },
    "log": {
        "level": "info",
        "extremeLogging": true
    }
  }
}

Testing

Use Docker to spin up a RabbitMQ container: docker run -d -p 5672:5672 -p 15672:15672 --rm --hostname my-rabbit --name some-rabbit rabbitmq:3-management

Launch your CLI application: ./working/src/bin/my-project.js start 2>&1

Open RabbitMQ management console in a web browser: http://localhost:15672 User: guest. Password: guest.

Use the management console to send a message input the input exchange and to receive the message on the output exchange.

Tip: you can find the rabbit connection again using the

docker ps

Contributing

To ensure that your code works fine you will need to test it against running RabbitMQ instance.

API Documentation

API documentation can be found here.