Message Bus SDK for AWS Resources

Table of Contents

• Configure your Environment
• Getting Started
• Test
• API

Configure your Environment

To test this package, we have two options, which are pointing to AWS Console, or using AWS Local Environment with localstack docker image.

To define which environment you will use, we can inform this envVar:

ENVIRONMENT_TEST="local"

Where you can use one of the values:

• "local": for use localstack image
• "aws": for use AWS Cloud

Did this, we need to inform some environment variables (you can use the 'dotenv' package) to use the AWS:

AWS_ACCOUNT="[YOUR_ACCOUNT]"
AWS_REGION="[YOUR_REGION]"
AWS_ACCESS_KEY_ID="[YOUR_ACCESS_KEY_ID]"
AWS_SECRET_ACCESS_KEY="[YOUR_SECRET_ACCESS_KEY]"
AWS_SESSION_TOKEN="[YOUR_SESSION_TOKEN]"
AWS_PLANNER_ROLE_ARN="[YOUR_SESSION_TOKEN]"

And, to configure the observability, you can inform the endpoint in this envVar:

OPENTELEMETRY_ENDPOINT="[YOUR_PROVIDER_ENDPOINT]"

Getting Started

1. Install the package

NPM:

npm i @vtexlab/planner-message-bus

Yarn:

yarn add @vtexlab/planner-message-bus

2. Tags File

You need to create a tags.json file in your root/src project and input some AWS information tags, like bellow:

[
    { "Key": "team", "Value": "your-team-name" },
    { "Key": "appName", "Value": "your-app-name" }
]

4. Sample Usage/Configuration

Configure your startup

const {  MessageBus } = require('@vtexlab/planner-message-bus');

/* 
 The default is console, but you can inform others 
 like Honeycomb, OpenSearhc and Datadog 
*/
MessageBus.configureLogger("{YOUR_DESTINATION_STREAM}")

// Each queue will bind to the referenced topic, informed
MessageBus.configureEndpoint([
    {
        queue: `[QUEUE_NAME]`,
        topic: `[TOPIC_NAME]`,
    }
]);

Test

npm test

We use the 'jest' framework to run the test. If you use the AWS environment, you just need to inform the credentials, otherwise, to the local environment, we use:

• Testcontainers: an open source framework for providing any resources that can run in a Docker container.
• Docker

Testcontainers use docker host url to connect to the docker and use it. And, we need some pay attention to how this works in each OS:

• If we use Docker Desktop, or Docker in unix distribution:
  • We don't need to do anything
• WSL (Windows Subsytem Linux):
  • You need to expose the Docker and inform your DOCKER_HOST as an environment variable
• Let's see how the URL format for each OS:
  • MAC: unix:///var/run/docker.sock
  • WINDOWS: npipe:////./pipe/docker_engine
    • WSL: tcp://localhost:2375
  • UNIX: unix:///var/run/docker.sock

API

MessageBus.configureEndpoint({queue: 'queue-name', topic: 'topic-name'})

Create a new SQS queue and SNS topic, and then bind to each other

MessageBus.service.publishMessage('topic-name', {CONTENT})

Send a message to SNS topic to do broadcast

• Content: any kind of type/object, this will be transformed into a JSON format

MessageBus.service.sendMessage('queue-name', {CONTENT}, {PARAMS})

Send a message to SQS queue direct

• Content: any kind of type/object, this will be transformed into a JSON format
• Params: SQS.Types.SendMessageRequest
  • You don't need to inform:
    • MessageBody
    • MessageAttributes
    • QueueUrl

MessageBus.service.schedule.send({PARAMS})

Send a message to EventBridge informing the SNS topic as the destination, and then, it will deliver in a SNS topic when arrives on the scheduled date

{
    "id": "Schedule rule ID",
    "message": "The object that will be sent",
    "scheduleDate": "Scheduled date to deliver the message",
    "topicName": "The destination which message will be deliver"
}

MessageBus.service.schedule.update({PARAMS})

Update a rule and message, already created, in EventBridge

{
    "id": "Schedule rule ID",
    "message": "The object that will be sent",
    "scheduleDate": "Scheduled date to deliver the message",
    "topicName": "The destination which message will be deliver"
}

MessageBus.service.schedule.delete("id")

Delete the rule (or cancel the scheduled message) in EventBridge

Service.handleConsumerMessage('queue-name', {RESILIENCE_PARAMS})

Handle a consumer to consume queue messages

• Resilience Params:
  • maxRetryCount: max number of attempts until send to DLQ
  • delaySeconds: the time that will wait between attempts