rascal npm

/**

* CUSTOM LIBS SETUP RASCAL FOR RABBITMQ BY RESTU WAHYU SAPUTRA

**/

import ‘dotenv/config’

import { CustomError } from ‘@helpers/helper.error’

import Broker, { PublicationSession, SubscriberSessionAsPromised } from ‘rascal’

import os from ‘os’
interface SetterConfig {

	key: string

	prefix: string

}
export class RabbitMQ {

	private brokerConfig: Broker.BrokerConfig

	private key: string

	private prefix: string
	constructor(key: string, prefix: string) {

		this.key = key

		this.prefix = prefix

		this.setConfig({ key, prefix })

	}
	private setConfig(config: SetterConfig) {

		const pubConfig: Broker.PublicationConfig = {

			vhost: process.env.RABBITMQ_VHOST,

			exchange: `rabbitmq_ex:${config.prefix}`,

			routingKey: ‘a.b.c’,

			autoCreated: true,

			options: {

				persistent: true,

				priority: os.cpus().length

			}

		}
		const subConfig: Broker.PublicationConfig = {

			vhost: process.env.RABBITMQ_VHOST,

			queue: `rabbitmq_eq:${config.prefix}`,

			autoCreated: true

		}
		const newPubConfig: Record = {}

		Object.assign(newPubConfig, { [`rabbitmq_pub:${config.key}:${config.prefix}`]: pubConfig })
		const newSubConfig: Record = {}

		Object.assign(newSubConfig, { [`rabbitmq_sub:${config.key}:${config.prefix}`]: subConfig })
		this.brokerConfig = {

			vhosts: {

				‘/’: {

					connectionStrategy: ‘random’,

					connection: {

						hostname: process.env.RABBITMQ_HOST,

						user: process.env.RABBITMQ_USERNAME,

						password: process.env.RABBITMQ_PASSWORD,

						port: process.env.RABBITMQ_PORT,

						protocol: process.env.RABBITMQ_PROTOCOL

					},

					exchanges: [`rabbitmq_ex:${config.prefix}`],

					queues: [`rabbitmq_eq:${config.prefix}`],

					bindings: [`rabbitmq_ex:${config.prefix}[a.b.c] -> rabbitmq_eq:${config.prefix}`],

					publications: newPubConfig,

					subscriptions: newSubConfig

				}

			},

			recovery: {

				[`rabbitmq_pub:${config.key}:${config.prefix}`]: { requeue: true, strategy: ‘ack’, xDeathFix: true },

				[`rabbitmq_sub:${config.key}:${config.prefix}`]: { strategy: ‘nack’ }

			}

		}

	}
	private getConfig(): Broker.BrokerConfig {

		return this.brokerConfig

	}
	private async connection(): Promise {

		try {

			const broker: Broker.BrokerAsPromised = await Broker.BrokerAsPromised.create(this.getConfig())

			broker.on(‘error’, console.error)

			return broker

		} catch (err: any) {

			return new CustomError(err.message)

		}

	}
	async publisher(data: Record | Record[]): Promise {

		try {

			const connection: Broker.BrokerAsPromised = await this.connection()

			const publisher: PublicationSession = await connection.publish(`rabbitmq_pub:${this.key}:${this.prefix}`, data)
			console.info(‘RabbitMQ publisher is called’)
			publisher.on(‘success’, (jobId: string) => console.log(`job ${jobId} is success`))

			publisher.on(‘error’, (_err: Error, jobId: string) => {

				console.log(`job ${jobId} is error`)

				connection.purge()

			})
			return true

		} catch (err: any) {

			return new CustomError(err.message)

		}

	}
	async subscriber(cb: (content: any, error?: Error) => any): Promise {

		try {

			const connection: Broker.BrokerAsPromised = await this.connection()

			const subscriber: SubscriberSessionAsPromised = await connection.subscribe(`rabbitmq_sub:${this.key}:${this.prefix}`)
			console.info(‘RabbitMQ subscriber is called’)
			subscriber

				.on(‘message’, (_message: any, content: any, ackOrNack: Broker.AckOrNack): void => {

					cb(content)

					ackOrNack()

				})

				.on(‘error’, console.error)

		} catch (err: any) {

			cb(null, err)

		}

	}

}
// publisher demo here

import { RabbitMQ } from ‘@libs/lib.rabbitmq’

;(async () => {

	try {

		const broker: InstanceType = new RabbitMQ(‘message:text’, ‘google’)

		setInterval(async () => {

			const res = await broker.publisher({ message: `hello wordl from publisher:${new Date().getTime()}` })

			console.log(res)

		}, 5000)

	} catch (err) {

		console.error(err)

	}

})()
// subscriber demo here

import { RabbitMQ } from ‘@libs/lib.rabbitmq’

;(async () => {

	try {

		const broker: InstanceType = new RabbitMQ(‘message:text’, ‘google’)

		broker.subscriber((content: string, error: Error) => {

			if (!error) console.log(content)

		})

	} catch (err) {

		console.error(err)

	}

})()