Navy.gif/wrappers
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: Navy.gif:b68484d4c512096b5257b2dfdf84e12fe621d3fd
Branches Tags
Navy.gif:main
..
compare: Navy.gif:36d5339e15e1fe591f5d1f69e23cc5a495d6c5d9
Branches Tags
Navy.gif:main

No commits in common. "b68484d4c512096b5257b2dfdf84e12fe621d3fd" and "36d5339e15e1fe591f5d1f69e23cc5a495d6c5d9" have entirely different histories.
b68484d4c5 ... 36d5339e15

10 changed files with 166 additions and 404 deletions
Show Stats Expand all files Collapse all files

115
.eslintrc.json
View File

@ -15,38 +15,17 @@
"sourceType": "module"
},
"rules": {
"@typescript-eslint/no-unused-vars": "off",
"@typescript-eslint/no-non-null-assertion":"off",
"accessor-pairs": "warn",
"array-callback-return": "warn",
"array-bracket-newline": [
"warn",
"consistent"
],
"array-bracket-spacing": [
"warn",
"always",
{
"objectsInArrays": false,
"arraysInArrays": false
}
],
"array-bracket-newline": [ "warn", "consistent" ],
"array-bracket-spacing": [ "warn", "always", { "objectsInArrays": false, "arraysInArrays": false }],
"arrow-spacing": "warn",
"block-scoped-var": "warn",
"block-spacing": [
"warn",
"always"
],
"brace-style": [
"warn",
"allman"
],
"block-spacing": [ "warn", "always" ],
"brace-style": [ "warn", "1tbs" ],
"callback-return": "warn",
"camelcase": "warn",
"comma-dangle": [
"warn",
"only-multiline"
],
"comma-dangle": [ "warn", "only-multiline" ],
"comma-spacing": [
"warn",
{
@ -89,25 +68,9 @@
"implicit-arrow-linebreak": "warn",
"indent": "warn",
"init-declarations": "warn",
"quotes": ["error" , "single"],
"jsx-quotes": [
"warn",
"prefer-single"
],
"key-spacing": [
"warn",
{
"beforeColon": false,
"afterColon": true
}
],
"keyword-spacing": [
"warn",
{
"after": true,
"before": true
}
],
"jsx-quotes": [ "warn", "prefer-single" ],
"key-spacing": [ "warn", { "beforeColon": false, "afterColon": true }],
"keyword-spacing": [ "warn", { "after": true, "before": true }],
"linebreak-style": [
"error",
"unix"
@ -116,28 +79,10 @@
"lines-around-directive": "warn",
"max-classes-per-file": "warn",
"max-nested-callbacks": "warn",
"max-len": [
"warn",
{
"code": 140,
"ignoreComments": true,
"ignoreStrings": true,
"ignoreTemplateLiterals": true,
"ignoreRegExpLiterals": true
}
],
"max-lines-per-function": [
"warn",
100
],
"max-depth": [
"warn",
3
],
"new-parens": "warn",
"no-alert": "warn",
"no-array-constructor": "warn",
// "no-bitwise": "warn",
"no-bitwise": "warn",
"no-buffer-constructor": "warn",
"no-caller": "warn",
"no-console": "warn",
@ -151,6 +96,7 @@
"no-extend-native": "warn",
"no-extra-bind": "warn",
"no-extra-label": "warn",
"no-extra-parens": "warn",
"no-floating-decimal": "warn",
"no-implicit-coercion": "warn",
"no-implicit-globals": "warn",
@ -201,7 +147,7 @@
"no-unmodified-loop-condition": "warn",
"no-unneeded-ternary": "error",
"no-unused-expressions": "warn",
"@typescript-eslint/no-use-before-define": "error",
"no-use-before-define": "error",
"no-useless-call": "warn",
"no-useless-computed-key": "warn",
"no-useless-concat": "warn",
@ -209,39 +155,20 @@
"no-useless-rename": "warn",
"no-useless-return": "warn",
"no-var": "warn",
// "no-void": "warn",
"no-void": "warn",
"no-whitespace-before-property": "error",
"nonblock-statement-body-position": [
"warn",
"below"
],
"nonblock-statement-body-position": [ "warn", "below" ],
"object-curly-spacing": [
"warn",
"always"
],
"object-property-newline": [
"warn",
{
"allowAllPropertiesOnSameLine": true
}
],
"object-property-newline": [ "warn", { "allowAllPropertiesOnSameLine": true }],
"object-shorthand": "warn",
"one-var-declaration-per-line": "warn",
"operator-assignment": "warn",
"operator-linebreak": [
"warn",
"before"
],
"operator-linebreak": [ "warn", "before" ],
"padding-line-between-statements": "warn",
"padded-blocks": [
"warn",
{
"switches": "never"
},
{
"allowSingleLineBlocks": true
}
],
"padded-blocks": [ "warn", { "switches": "never" }, { "allowSingleLineBlocks": true }],
"prefer-arrow-callback": "warn",
"prefer-const": "warn",
"prefer-destructuring": "warn",
@ -262,18 +189,12 @@
"last"
],
"space-before-blocks": "warn",
"space-before-function-paren": [
"error",
"always"
],
"space-before-function-paren": [ "error", "always" ],
"space-in-parens": [
"warn",
"never"
],
"spaced-comment": [
"warn",
"always"
],
"spaced-comment": [ "warn", "always" ],
"strict": "warn",
"switch-colon-spacing": "warn",
"symbol-description": "warn",

2
index.ts
View File

@ -1,4 +1,4 @@
export { MessageBroker, BrokerOptions } from './src/MessageBroker.js';
export { MessageBroker, BrokerOptions } from "./src/MessageBroker.js";
export { MariaDB, MariaOptions } from './src/MariaDB.js';
export { MongoDB, MongoOptions } from './src/MongoDB.js';
export { ObjectId, Document, DeleteResult, Collection } from 'mongodb';

2
package.json
View File

@ -1,6 +1,6 @@
{
"name": "@navy.gif/wrappers",
"version": "1.4.3",
"version": "1.3.20",
"description": "Various wrapper classes I use in my projects",
"repository": "https://git.corgi.wtf/Navy.gif/wrappers.git",
"author": "Navy.gif",

158
src/MariaDB.ts
View File

@ -1,18 +1,17 @@
import { inspect } from 'node:util';
import { ILogger, IServer } from './interfaces/index.js';
import mysql, { PoolCluster, PoolClusterConfig, PoolConfig, PoolConnection, FieldInfo } from 'mysql';
import mysql, { Pool, PoolCluster, PoolClusterConfig, PoolConfig, PoolConnection, FieldInfo } from 'mysql';
import { LoggerClientOptions } from './interfaces/Logger.js';
const SAFE_TO_RETRY = [ 'ER_LOCK_DEADLOCK', 'PROTOCOL_CONNECTION_LOST' ];
const SAFE_TO_RETRY = [ 'ER_LOCK_DEADLOCK' ];
type Credentials = {
user: string,
password: string,
host: string,
port: number,
database: string,
node: string
database: string
}
export type MariaOptions = {
@ -32,28 +31,19 @@ type MariaError = {
sqlMessage:string
} & Error
type QueryOptions = {
node?: string,
timeout?: number
}
class MariaDB
{
class MariaDB {
#_activeQueries: number;
#afterLastQuery: (() => void) | null;
#logger: ILogger;
#_ready: boolean;
#load: boolean;
#config: MariaOptions;
#credentials: Credentials[];
#cluster: boolean;
#pool: PoolCluster | null;
#nodes: string[];
#pool: PoolCluster | Pool | null;
constructor (server: IServer, options: MariaOptions)
{
constructor (server: IServer, options: MariaOptions) {
if (!server)
throw new Error('Missing reference to server!');
@ -61,18 +51,11 @@ class MariaDB
throw new Error('No config options provided!');
this.#config = options;
this.#load = options.load ?? true;
const { host, user, port, password, database } = options.credentials;
const hosts = host.split(',');
this.#credentials = [];
this.#nodes = [];
for (const remote of hosts)
{
const [ node ] = remote.split('.');
if (typeof node === 'undefined' || !(/(([a-zA-Z]\w*[0-9]?)|([0-9]?\w*[a-zA-Z]))/u).test(node))
throw new Error('Invalid host config, expecting qualified domain names');
this.#credentials.push({ host: remote, user, port, password, database, node });
this.#nodes.push(node);
for (const remote of hosts) {
this.#credentials.push({ host: remote, user, port, password, database });
}
this.#pool = null;
@ -86,48 +69,42 @@ class MariaDB
}
get ready ()
{
get ready () {
return this.#_ready;
}
get activeQueries ()
{
get activeQueries () {
return this.#_activeQueries;
}
async init ()
{
async init () {
if (!this.#load)
if (!this.#config.load)
return this.#logger.info('Not loading MariaDB');
this.#logger.status(`Creating${this.#cluster ? ' cluster' : ''} connection pool`);
this.#pool = mysql.createPoolCluster(this.#config.cluster);
for (const creds of this.#credentials)
{
this.#pool.add(creds.node, { ...this.#config.client, ...creds });
this.#logger.info(`Added node ${creds.node} to pool cluster`);
if (this.#cluster) {
this.#pool = mysql.createPoolCluster(this.#config.cluster);
for (const creds of this.#credentials)
this.#pool.add({ ...this.#config.client, ...creds });
} else {
this.#pool = mysql.createPool({ ...this.#config.client, ...this.#credentials[0] });
}
this.#pool.on('connection', (connection) =>
{
this.#pool.on('connection', (connection) => {
this.#logger.debug(`New connection: ${connection?.threadId || null}`);
});
this.#pool.on('acquire', (connection) =>
{
this.#pool.on('acquire', (connection) => {
this.#_activeQueries++;
this.#logger.debug(`Connection acquired: ${connection?.threadId || null}`);
});
this.#pool.on('enqueue', () =>
{
this.#logger.debug('Query enqueued for connection');
this.#pool.on('enqueue', () => {
this.#logger.debug(`Query enqueued for connection`);
});
this.#pool.on('release', (connection) =>
{
this.#pool.on('release', (connection) => {
this.#_activeQueries--;
if (!this.ready && !this.#_activeQueries && this.#afterLastQuery)
@ -136,20 +113,11 @@ class MariaDB
this.#logger.debug(`Connection released: ${connection?.threadId || null}`);
});
this.#pool.on('remove', (nodeId) =>
{
this.#logger.status(`Node ${nodeId} was removed from pool`);
const index = this.#nodes.findIndex(n => n === nodeId);
this.#nodes.splice(index, 1);
});
this.#logger.info('Testing MariaDB connection');
await new Promise<void>((resolve, reject) =>
{
this.#logger.info(`Testing MariaDB connection`);
await new Promise<void>((resolve, reject) => {
if (!this.#pool)
return reject(new Error('Missing connection pool'));
this.#pool.getConnection((err, conn) =>
{
this.#pool.getConnection((err, conn) => {
if (err)
return reject(err);
conn.release();
@ -157,31 +125,27 @@ class MariaDB
});
});
this.#logger.status('Database connected');
this.#logger.status(`Database connected`);
this.#_ready = true;
return this;
}
async close ()
{
this.#logger.status('Shutting down database connections');
async close () {
this.#logger.status(`Shutting down database connections`);
if (!this.ready)
return Promise.resolve();
this.#_ready = false;
if (this.#_activeQueries)
{
if (this.#_activeQueries) {
this.#logger.info(`${this.#_activeQueries} active queries still running, letting them finish`);
await this.finishQueries();
this.#logger.info('Queries finished, shutting down');
this.#logger.info(`Queries finished, shutting down`);
}
return new Promise<void>(resolve =>
{
return new Promise<void>(resolve => {
if (!this.#pool)
return resolve();
this.#pool.end(() =>
{
this.#pool.end(() => {
this.#pool?.removeAllListeners();
resolve();
});
@ -189,34 +153,21 @@ class MariaDB
});
}
finishQueries ()
{
return new Promise<void>(resolve =>
{
finishQueries () {
return new Promise<void>(resolve => {
this.#afterLastQuery = resolve;
});
}
getConnection (node?: string): Promise<PoolConnection>
{
return new Promise((resolve, reject) =>
{
getConnection (): Promise<PoolConnection> {
return new Promise((resolve, reject) => {
if (!this.#pool)
return reject(new Error('Pool closed'));
// Get node by name
const pool = this.#pool;
if (node && !this.#nodes.includes(node))
{
this.#logger.warn(`Node ${node} is not available in pool, falling back to arbitrary node`);
node = '*';
}
return pool.of(node ?? '*').getConnection((err, conn) =>
{
this.#pool.getConnection((err, conn) => {
if (err)
return reject(err);
return resolve(conn);
resolve(conn);
});
});
}
@ -225,16 +176,11 @@ class MariaDB
* @throws {MariaError}
* @private
* */
async #_query<T> (query: string, values: (string | number | string[] | number[])[], { timeout, node }: QueryOptions = {}, attempts = 0):
Promise<T[] | FieldInfo[]>
{
const connection = await this.getConnection(node);
try
{
const result = await new Promise<T[] | FieldInfo[] | undefined>((resolve, reject) =>
{
const q = connection.query({ timeout, sql: query }, values, (err, results, fields) =>
{
async #_query<T> (query: string, values: (string | number | string[] | number[])[], timeout?: number, attempts = 0): Promise<T[] | FieldInfo[]> {
const connection = await this.getConnection();
try {
const result = await new Promise<T[] | FieldInfo[] | undefined>((resolve, reject) => {
const q = connection.query({ timeout, sql: query }, [ values ], (err, results, fields) => {
if (err)
reject(err);
else if (results)
@ -246,19 +192,16 @@ class MariaDB
this.#logger.debug(`Constructed query: ${q.sql}`);
});
return Promise.resolve(result || []);
}
catch (err)
{
} catch (err) {
const error = err as MariaError;
// Retry safe errors // (Galera) Instance not ready for query
if ((SAFE_TO_RETRY.includes(error.code) || error.errno === 1047) && attempts < 5) //
return this.#_query(query, values, { timeout, node }, ++attempts);
return this.#_query(query, values, timeout, ++attempts);
return Promise.reject(error);
}
}
async query<T> (query: string, values: (string | number | string[] | number[])[], opts?: QueryOptions): Promise<T[] | FieldInfo[]>
{
async query<T> (query: string, values: (string | number | string[] | number[])[], timeout?: number): Promise<T[] | FieldInfo[]> {
if (!this.ready)
return Promise.reject(new Error('MariaDB not ready'));
@ -268,13 +211,12 @@ class MariaDB
batch = values.some(val => val instanceof Array);
this.#logger.debug(`Incoming query (batch: ${batch})\n${query}\n${inspect(values)}`);
return this.#_query<T>(query, values, opts);
return this.#_query<T>(query, values, timeout);
}
q<T> (query: string, values: (string | number | string[] | number[])[], opts?: QueryOptions)
{
return this.query<T>(query, values, opts);
q<T> (query: string, values: (string | number | string[] | number[])[], timeout?: number) {
return this.query<T>(query, values, timeout);
}
}

141
src/MessageBroker.ts
View File

@ -1,4 +1,4 @@
import { ILogger, IServer } from './interfaces/index.js';
import { ILogger, IServer } from "./interfaces/index.js";
import amqp, { AmqpConnectionManager, ChannelWrapper } from 'amqp-connection-manager';
import { Channel, ConfirmChannel, ConsumeMessage, Options } from 'amqplib';
@ -56,8 +56,7 @@ type Consumer<T = any> = (content: T, msg: ConsumeMessage) => Promise<void>
// eslint-disable-next-line @typescript-eslint/no-explicit-any
type Subscriber<T = any> = (content: T, msg: ConsumeMessage) => Promise<void>
class MessageBroker
{
class MessageBroker {
// Broker definitions
#load: boolean;
@ -81,10 +80,9 @@ class MessageBroker
#_qQueue: InternalQueueMsg[];
#_qTO: NodeJS.Timeout | null;
constructor (server: IServer, options: BrokerOptions)
{
constructor (server: IServer, options: BrokerOptions) {
this.#load = options.load ?? true;
this.#load = options.load || false;
this.#hosts = options.host.split(',');
this.#username = options.user;
this.#password = options.pass;
@ -116,8 +114,7 @@ class MessageBroker
}
async init ()
{
async init () {
if (!this.#load)
return this.#logger.info('Not loading RabbitMQ');
@ -127,29 +124,23 @@ class MessageBroker
const connectionStrings = this.#hosts.map(host => `amqp://${credentials}${host}:${this.#port}/${this.#vhost}`);
this.#connection = await amqp.connect(connectionStrings);
this.#connection.on('disconnect', async ({ err }) =>
{
this.#connection.on('disconnect', async ({ err }) => {
this.#logger.status(`Disconnected: ${err.message}`);
await this.#channel?.close();
this.#channel = null;
});
this.#connection.on('blocked', ({ reason }) =>
{
this.#connection.on('blocked', ({ reason }) => {
this.#logger.status(`Blocked: ${reason}`);
});
this.#connection.on('connectFailed', ({ err }) =>
{
this.#connection.on('connectFailed', ({ err }) => {
this.#logger.error(`Message broker failed to connect: ${err.stack || err.message}`);
});
this.#connection.on('connect', async ({ url }) =>
{
this.#connection.on('connect', async ({ url }) => {
this.#logger.status(`Message broker connected to ${url}`);
});
await new Promise<void>((resolve, reject) =>
{
this.#connection?.once('connect', () =>
{
await new Promise<void>((resolve, reject) => {
this.#connection?.once('connect', () => {
this.#connection?.removeListener('connectFailed', reject);
resolve();
});
@ -161,15 +152,12 @@ class MessageBroker
}
async close ()
{
if (this.#channel)
{
async close () {
if (this.#channel) {
await this.#channel.close();
this.#channel.removeAllListeners();
}
if (this.#connection)
{
if (this.#connection) {
await this.#connection.close();
// No clue why the removeAllListeners function isn't exposed from the connection manager, but it exists since it's an EventEmitter
// eslint-disable-next-line @typescript-eslint/ban-ts-comment
@ -178,13 +166,11 @@ class MessageBroker
}
}
async createChannel ()
{
async createChannel () {
const exchanges = Object.entries(this.#exchanges);
const queues = Object.entries(this.#queues);
if (this.#channel)
{
if (this.#channel) {
this.#logger.debug('Closing old channel');
await this.#channel.close().catch(err => this.#logger.error(err.stack));
}
@ -193,14 +179,11 @@ class MessageBroker
throw new Error();
this.#logger.debug('Creating channel');
this.#channel = this.#connection.createChannel({
setup: async (channel: Channel | ConfirmChannel) =>
{
for (const [ name, props ] of exchanges)
{
setup: async (channel: Channel | ConfirmChannel) => {
for (const [ name, props ] of exchanges) {
await channel.assertExchange(name, props.type ?? 'fanout', props);
}
for (const [ name, props ] of queues)
{
for (const [ name, props ] of queues) {
await channel.assertQueue(name, props);
}
}
@ -210,12 +193,10 @@ class MessageBroker
this.#channel.on('connect', () => this.#logger.status('Channel connected'));
this.#channel.on('error', (err, info) => this.#logger.error(`${info.name}: ${err.stack}`));
await new Promise<void>((resolve, reject) =>
{
await new Promise<void>((resolve, reject) => {
if (!this.#channel)
return reject(new Error('Missing channel?'));
this.#channel.once('connect', () =>
{
this.#channel.once('connect', () => {
this.#channel?.removeListener('error', reject);
resolve();
});
@ -229,8 +210,7 @@ class MessageBroker
}
// Consume queue
async consume<T> (queue: string, consumer: Consumer<T>, options?: Options.Consume)
{
async consume<T> (queue: string, consumer: Consumer<T>, options?: Options.Consume) {
if (!this.#channel)
throw new Error('Channel does not exist');
@ -242,12 +222,10 @@ class MessageBroker
this.#consumers.set(queue, list);
}
private async _consume<T> (queue: string, consumer: Consumer<T>, options?: Options.Consume): Promise<void>
{
private async _consume<T> (queue: string, consumer: Consumer<T>, options?: Options.Consume): Promise<void> {
if (!this.#channel)
return Promise.reject(new Error('Channel doesn\'t exist'));
await this.#channel.consume(queue, async (msg: ConsumeMessage) =>
{
await this.#channel.consume(queue, async (msg: ConsumeMessage) => {
if (msg.content)
await consumer(JSON.parse(msg.content.toString()), msg);
this.#channel?.ack(msg);
@ -255,8 +233,7 @@ class MessageBroker
}
// Subscribe to exchange, ensures messages aren't lost by binding it to a queue
async subscribe<T> (name: string, listener: Subscriber<T | Buffer>)
{
async subscribe<T> (name: string, listener: Subscriber<T>) {
if (!this.#channel)
throw new Error('Channel does not exist');
@ -270,8 +247,7 @@ class MessageBroker
}
private async _subscribe<T> (name: string, listener: Subscriber<T | Buffer>): Promise<void>
{
private async _subscribe<T> (name: string, listener: Subscriber<T>): Promise<void> {
if (!this.#channel)
return Promise.reject(new Error('Channel doesn\'t exist'));
@ -279,19 +255,15 @@ class MessageBroker
const queue = await this.#channel.assertQueue('', { exclusive: true });
await this.#channel.bindQueue(queue.queue, name, '');
await this.#channel.consume(queue.queue, async (msg) =>
{
if (msg.content && msg.content.toString().startsWith('{'))
await this.#channel.consume(queue.queue, async (msg) => {
if (msg.content)
await listener(JSON.parse(msg.content.toString()), msg);
else
await listener(msg.content, msg);
this.#channel?.ack(msg);
});
}
// Add item to queue
async enqueue (queue: string, content: object, headers?: object): Promise<void | number>
{
async enqueue (queue: string, content: object, headers?: object): Promise<void | number> {
const properties = {
persistent: true,
contentType: 'application/json',
@ -302,12 +274,9 @@ class MessageBroker
if (!this.#channel)
return this.#_qQueue.push({ queue, content, properties });
try
{
try {
await this.#channel.sendToQueue(queue, Buffer.from(JSON.stringify(content)), properties);
}
catch (_)
{
} catch (_) {
this.#_qQueue.push({ queue, content, properties });
if (!this.#_qTO)
this.#_qTO = setTimeout(this.#_processQueues.bind(this), 5000);
@ -315,8 +284,7 @@ class MessageBroker
}
// Publish to exchange
async publish (exchange: string, content: object, { headers, routingKey = '' }: {headers?: string, routingKey?: string} = {}): Promise<void|number>
{
async publish (exchange: string, content: object, { headers, routingKey = '' }: {headers?: string, routingKey?: string} = {}): Promise<void|number> {
const properties = {
contentType: 'application/json',
headers
@ -326,12 +294,9 @@ class MessageBroker
if (!this.#channel)
return this.#_pQueue.push({ exchange, content, routingKey, properties });
try
{
try {
await this.#channel.publish(exchange, routingKey, Buffer.from(JSON.stringify(content)), properties);
}
catch (err)
{
} catch (err) {
const error = err as Error;
if (!error.message.includes('nack'))
this.#logger.error(`Error while publishing to ${exchange}:\n${error.stack}`);
@ -341,74 +306,62 @@ class MessageBroker
}
}
assertExchange (exchange: string, props?: ExchangeDef)
{
assertExchange (exchange: string, props?: ExchangeDef) {
if (!this.#channel)
throw new Error('Channel doesn\'t exist');
return this.#channel.assertExchange(exchange, props?.type ?? 'fanout', props);
}
assertQueue (queue: string, opts?: QueueDef)
{
assertQueue (queue: string, opts?: QueueDef) {
if (!this.#channel)
throw new Error('Channel doesn\'t exist');
return this.#channel.assertQueue(queue, opts);
}
// Processes messages queued up while the broker was unreachable
async #_processQueues ()
{
async #_processQueues () {
if (!this.#channel)
throw new Error('Channel doesn\'t exist');
this.#logger.status('Processing queues of unsent messages');
const pQ = [ ...this.#_pQueue ];
this.#_pQueue = [];
for (const msg of pQ)
{
for (const msg of pQ) {
const { exchange, content, routingKey, properties } = msg;
const result = await this.#channel.publish(exchange, routingKey, Buffer.from(JSON.stringify(content)), properties)
.catch(() => null);
const result = await this.#channel.publish(exchange, routingKey, Buffer.from(JSON.stringify(content)), properties).catch(() => null);
if (!result)
this.#_pQueue.push(msg);
}
const qQ = [ ...this.#_qQueue ];
for (const msg of qQ)
{
for (const msg of qQ) {
const { queue, content, properties } = msg;
const result = await this.#channel.sendToQueue(queue, Buffer.from(JSON.stringify(content)), properties).catch(() => null);
if (!result)
this.#_qQueue.push(msg);
}
this.#logger.status('Done processing');
if (this.#_qTO)
{
if (this.#_qTO) {
clearTimeout(this.#_qTO);
this.#_qTO = null;
}
}
async #_restoreListeners ()
{
async #_restoreListeners () {
this.#logger.status(`Restoring ${this.#consumers.size} consumers`);
for (const [ name, list ] of this.#consumers)
{
for (const [ name, list ] of this.#consumers) {
this.#logger.debug(`Processing consumer ${name}: ${list.length}`);
for (const { consumer, options } of list)
{
for (const { consumer, options } of list) {
await this._consume(name, consumer, options);
}
}
this.#logger.status(`Restoring ${this.#subscribers.size} subscribers`);
for (const [ name, list ] of this.#subscribers)
{
for (const [ name, list ] of this.#subscribers) {
this.#logger.debug(`Processing subscriber ${name}: ${list.length}`);
for (const subscriber of list)
{
for (const subscriber of list) {
await this._subscribe(name, subscriber);
}
}
this.#logger.status('Done restoring');
this.#logger.status(`Done restoring`);
}
}

117
src/MongoDB.ts
View File

@ -1,6 +1,6 @@
import { inspect } from 'node:util';
import { MongoClient, MongoClientOptions, Db, Document, WithId, ObjectId, Filter, IndexSpecification, CreateIndexesOptions, FindOptions } from 'mongodb';
import { IServer, ILogger, LoggerClientOptions } from './interfaces/index.js';
import { inspect } from "node:util";
import { MongoClient, MongoClientOptions, Db, Document, WithId, ObjectId, Filter, IndexSpecification, CreateIndexesOptions, FindOptions } from "mongodb";
import { IServer, ILogger, LoggerClientOptions } from "./interfaces/index.js";
type Credentials = {
URI?: string,
@ -29,8 +29,7 @@ export type MongoOptions = {
*
* @class MongoDB
*/
class MongoDB
{
class MongoDB {
#_database: string;
#config: MongoOptions;
@ -40,8 +39,7 @@ class MongoDB
#db: Db | null;
#_client: MongoClient;
constructor (server: IServer, config: MongoOptions)
{
constructor (server: IServer, config: MongoOptions) {
if (!server)
throw new Error('Missing reference to server!');
@ -50,7 +48,7 @@ class MongoDB
const { user, password, host, port, database, URI, authDb } = config.credentials;
if ((!host?.length || !port || !database?.length) && !URI)
throw new Error('Must provide host, port, and database OR URI parameters!');
throw new Error(`Must provide host, port, and database OR URI parameters!`);
this.#config = config;
this.#db = null; // DB connection
@ -58,22 +56,16 @@ class MongoDB
this.#logger = server.createLogger(this, config.loggerOptions);
if (URI)
{
if (URI) {
this.#URI = URI;
}
else
{
} else {
let AUTH_DB = authDb;
const auth = user ? `${user}:${password}@` : '';
if (!AUTH_DB && auth)
{
this.#logger.warn('No explicit auth db provided with MONGO_AUTH_DB, will attempt to use MONGO_DB for auth source');
if (!AUTH_DB && auth) {
this.#logger.warn(`No explicit auth db provided with MONGO_AUTH_DB, will attempt to use MONGO_DB for auth source`);
AUTH_DB = authDb;
}
else if (!auth)
{
this.#logger.warn('No auth provided, proceeding without');
} else if (!auth) {
this.#logger.warn(`No auth provided, proceeding without`);
}
this.#URI = `mongodb://${auth}${host}:${port}/${AUTH_DB || ''}?readPreference=secondaryPreferred`;
@ -83,19 +75,17 @@ class MongoDB
// TODO figure out reconnecting to DB when connection fails
this.#_client.on('error', (error) => this.#logger.error(`MongoDB error:\n${error.stack}`))
.on('timeout', () => this.#logger.warn('MongoDB timed out'))
.on('close', () => this.#logger.info('MongoDB client disconnected'))
.on('open', () => this.#logger.info('MongoDB client connected'));
.on('timeout', () => this.#logger.warn(`MongoDB timed out`))
.on('close', () => this.#logger.info(`MongoDB client disconnected`))
.on('open', () => this.#logger.info(`MongoDB client connected`));
}
get database ()
{
get database () {
return this.#_database;
}
get client ()
{
get client () {
return this.#_client;
}
@ -104,8 +94,7 @@ class MongoDB
*
* @memberof MongoDB
*/
async init ()
{
async init () {
if (!this.#config.load)
return this.#logger.info('Not loading MongoDB');
@ -116,7 +105,7 @@ class MongoDB
this.#logger.status(`Initializing database connection to ${this.#_client.options.hosts}`);
await this.#_client.connect();
this.#logger.debug('Connected, selecting DB');
this.#logger.debug(`Connected, selecting DB`);
this.#db = await this.#_client.db(this.#_database);
this.#logger.status('MongoDB ready');
@ -125,8 +114,7 @@ class MongoDB
}
async close ()
{
async close () {
if (!this.#db)
return;
@ -136,8 +124,7 @@ class MongoDB
this.#logger.status('Database closed');
}
get mongoClient ()
{
get mongoClient () {
return this.#_client;
}
@ -149,11 +136,10 @@ class MongoDB
* @returns {Array} An array containing the corresponding objects for the query
* @memberof Database
*/
async find<T extends Document> (db: string, query: MongoQuery, options?: FindOptions<T>): Promise<WithId<T>[]>
{
async find<T extends Document> (db: string, query: MongoQuery, options?: FindOptions<T>): Promise<WithId<T>[]> {
if (!this.#db)
throw new Error('MongoDB not connected');
throw new Error(`MongoDB not connected`);
if (typeof db !== 'string')
throw new TypeError('Expecting collection name for the first argument');
@ -176,11 +162,10 @@ class MongoDB
* @returns {Object} An object containing the queried data
* @memberof Database
*/
async findOne<T extends Document> (db: string, query: MongoQuery, options: FindOptions<T> = {}): Promise<WithId<T> | null>
{
async findOne<T extends Document> (db: string, query: MongoQuery, options: FindOptions<T> = {}): Promise<WithId<T> | null> {
if (!this.#db)
throw new Error('MongoDB not connected');
throw new Error(`MongoDB not connected`);
if (typeof db !== 'string')
throw new TypeError('Expecting collection name for the first argument');
@ -202,15 +187,14 @@ class MongoDB
* @returns {WriteResult} Object containing the followint counts: Matched, Upserted, Modified
* @memberof Database
*/
async updateMany<T extends Document> (db: string, filter: MongoQuery, data: T, upsert = false)
{
async updateMany<T extends Document> (db: string, filter: MongoQuery, data: T, upsert = false) {
if (!this.#db)
throw new Error('MongoDB not connected');
throw new Error(`MongoDB not connected`);
if (typeof db !== 'string')
throw new TypeError('Expecting collection name for the first argument');
if (!filter)
throw new Error('Cannot run update many without a filter, if you mean to update every single document, pass an empty object');
throw new Error(`Cannot run update many without a filter, if you mean to update every single document, pass an empty object`);
if (typeof filter._id === 'string')
filter._id = new ObjectId(filter._id);
@ -229,11 +213,10 @@ class MongoDB
* @returns {WriteResult} Object containing the followint counts: Matched, Upserted, Modified
* @memberof Database
*/
async updateOne (db: string, filter: MongoQuery, data: Document, upsert = false)
{
async updateOne (db: string, filter: MongoQuery, data: Document, upsert = false) {
if (!this.#db)
throw new Error('MongoDB not connected');
throw new Error(`MongoDB not connected`);
if (typeof db !== 'string')
throw new TypeError('Expecting collection name for the first argument');
if (typeof filter._id === 'string')
@ -254,11 +237,10 @@ class MongoDB
* @returns {WriteResult} Object containing the followint counts: Matched, Upserted, Modified
* @memberof Database
*/
async insertOne (db: string, data: Document)
{
async insertOne (db: string, data: Document) {
if (!this.#db)
throw new Error('MongoDB not connected');
throw new Error(`MongoDB not connected`);
if (typeof db !== 'string')
throw new TypeError('Expecting collection name for the first argument');
if (typeof data._id === 'string')
@ -270,11 +252,10 @@ class MongoDB
}
async deleteOne (db: string, filter: Document)
{
async deleteOne (db: string, filter: Document) {
if (!this.#db)
throw new Error('MongoDB not connected');
throw new Error(`MongoDB not connected`);
if (typeof db !== 'string')
throw new TypeError('Expecting collection name for the first argument');
if (typeof filter._id === 'string')
@ -296,11 +277,10 @@ class MongoDB
* @returns
* @memberof Database
*/
async push (db: string, filter: Document, data: object, upsert = false)
{
async push (db: string, filter: Document, data: object, upsert = false) {
if (!this.#db)
throw new Error('MongoDB not connected');
throw new Error(`MongoDB not connected`);
if (typeof db !== 'string')
throw new TypeError('Expecting collection name for the first argument');
if (typeof filter._id === 'string')
@ -321,11 +301,10 @@ class MongoDB
* @returns {object}
* @memberof Database
*/
random<T extends Document> (db: string, filter: Document = {}, amount = 1)
{
random<T extends Document> (db: string, filter: Document = {}, amount = 1) {
if (!this.#db)
throw new Error('MongoDB not connected');
throw new Error(`MongoDB not connected`);
if (typeof db !== 'string')
throw new TypeError('Expecting collection name for the first argument');
if (typeof filter._id === 'string')
@ -341,32 +320,28 @@ class MongoDB
}
stats (options = {})
{
stats (options = {}) {
if (!this.#db)
throw new Error('MongoDB not connected');
throw new Error(`MongoDB not connected`);
const result = this.#db.stats(options);
return result;
}
collection<T extends Document> (coll: string)
{
collection<T extends Document> (coll: string) {
if (!this.#db)
throw new Error('MongoDB not connected');
throw new Error(`MongoDB not connected`);
return this.#db.collection<T>(coll);
}
count (coll: string, query: Document)
{
count (coll: string, query: Document) {
if (!this.#db)
throw new Error('MongoDB not connected');
throw new Error(`MongoDB not connected`);
return this.#db.collection(coll).countDocuments(query);
}
async ensureIndex (collection: string, indices: IndexSpecification = [], options?: CreateIndexesOptions)
{
async ensureIndex (collection: string, indices: IndexSpecification = [], options?: CreateIndexesOptions) {
if (!this.#db)
throw new Error('MongoDB not connected');
throw new Error(`MongoDB not connected`);
if (!(indices instanceof Array))
indices = [ indices ];
await this.#db.collection(collection).createIndex(indices, options);

2
src/interfaces/Server.ts
View File

@ -1,5 +1,5 @@
import { ILogger, LoggerClientOptions } from "./Logger.js";
export interface IServer {
createLogger(obj: object, options?: Partial<LoggerClientOptions>): ILogger
createLogger(obj: object, options?: LoggerClientOptions): ILogger
}

27
tests/testBroker.js
View File

@ -1,27 +0,0 @@
import { MessageBroker } from '../build/esm/index.js';
const broker = new MessageBroker({
createLogger: () =>
{
return {
debug: console.log,
info: console.log,
status: console.log,
warn: console.log,
error: console.error
};
}
}, {
load: true,
host: 'rabbitmq-01.stylis.local',
user: 'stylis',
pass: 'RrwJyrfeXFMimDH3hjZ5xSreMAmXtQJj',
vhost: 'development',
port: 5672
});
await broker.init();
broker.subscribe('chatlogs', (message) =>
{
console.log(message);
});

3
tests/testMaria.js
View File

@ -4,8 +4,7 @@ import { MariaDB } from '../build/esm/index.js';
const credentials = JSON.parse(readFileSync('./credentials.json', { encoding: 'utf-8' }));
const maria = new MariaDB({
createLogger: () =>
{
createLogger: () => {
return {
debug: console.log,
info: console.log,

3
tests/testMongo.js
View File

@ -1,8 +1,7 @@
import { MongoDB } from '../build/esm/index.js';
const mongo = new MongoDB({
createLogger: () =>
{
createLogger: () => {
return {
debug: console.log,
info: console.log,