Compare commits
No commits in common. "b68484d4c512096b5257b2dfdf84e12fe621d3fd" and "36d5339e15e1fe591f5d1f69e23cc5a495d6c5d9" have entirely different histories.
b68484d4c5
...
36d5339e15
115
.eslintrc.json
115
.eslintrc.json
@ -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
2
index.ts
@ -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';
|
||||
@ -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
158
src/MariaDB.ts
@ -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);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@ -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
117
src/MongoDB.ts
@ -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);
|
||||
|
||||
@ -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
|
||||
}
|
||||
@ -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);
|
||||
});
|
||||
@ -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,
|
||||
|
||||
@ -1,8 +1,7 @@
|
||||
import { MongoDB } from '../build/esm/index.js';
|
||||
|
||||
const mongo = new MongoDB({
|
||||
createLogger: () =>
|
||||
{
|
||||
createLogger: () => {
|
||||
return {
|
||||
debug: console.log,
|
||||
info: console.log,
|
||||
|
||||
Loading…
Reference in New Issue
Block a user