index.js

"use strict"

const assertArgs = require("assert-args")
const nats = require("nats")
const url = require("url")

const handlers = require("./handlers")
const utils = require("../utils")

module.exports = class ServiceBus
{
    constructor(env, config)
    {
        let args = assertArgs(arguments,
        {
            "env": "object",
            "[config]": "object"
        })

        this.env = args.env
        this.config = args.config
    }

    async connect()
    {
        switch (this.config.type)
        {
            case "nats":
                try
                {
                    this.connection = nats.connect(url.format(
                    {
                        protocol: this.config.protocol,
                        hostname: this.config.hostname,
                        port: this.config.port,
                        slashes: true
                    }))

                    return this.connection
                }
                catch (error)
                {
                    throw error
                }

                break

            case "rabbitmq":
                try
                {
                    this.connection = await amqp.connect(url.format(
                    {
                        protocol: this.config.protocol,
                        auth: this.config.username + ":" + this.config.password,
                        hostname: this.config.hostname,
                        port: this.config.port,
                        slashes: true
                    }))
                }
                catch (error)
                {
                    throw error
                }

                this.channel = await this.connection.createChannel()
                let queues = this.config.queues

                for (let name in queues)
                {
                    try
                    {
                        switch (queues[name].type)
                        {
                            // push -> only one consumer takes the job
                            case "push":
                                await this.channel.assertExchange(queues[name].exchange, "topic")
                                break
                        }
                    }
                    catch (error)
                    {
                        throw error
                    }
                }

                this.channel.on("error", (error) =>
                {
                    throw error
                })

                this.channel.on("close", () =>
                {
                    throw new Error("channel closed")
                })

                return this.channel
        }
    }

    async handleRequests()
    {
        for (let name in handlers)
        {
            await handlers[name].call(this.env)
        }
    }
}