Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1,106 changes: 350 additions & 756 deletions package-lock.json

Large diffs are not rendered by default.

32 changes: 16 additions & 16 deletions package.json
Original file line number Diff line number Diff line change
Expand Up @@ -11,8 +11,9 @@
"json-message"
],
"scripts": {
"benchmark": "node benchmark -c $(( $(nproc) - 2 )) -m 100000",
"prepare": "./node_modules/.bin/tsc",
"test": "./node_modules/.bin/tsc && ./node_modules/.bin/nyc mocha && ./node_modules/.bin/nyc report --reporter=text-lcov && npm run test-coverage",
"test": "./node_modules/.bin/tsc && ./node_modules/.bin/nyc mocha && ./node_modules/.bin/nyc report --reporter=text-lcov",
"test-fast": "./node_modules/.bin/tsc && ./node_modules/.bin/nyc mocha && /usr/bin/env node -e \"import('open').then(open => open.default('file://`pwd`/coverage/index.html', { wait: false }))\"",
"test-local": "export COVERALLS_REPO_TOKEN=$IMQ_COVERALLS_TOKEN && npm test && /usr/bin/env node -e \"import('open').then(open => open.default('https://coveralls.io/github/imqueue/imq', { wait: false }))\"",
"test-dev": "npm run test && npm run clean-js && npm run clean-typedefs && npm run clean-maps",
Expand All @@ -37,37 +38,36 @@
"author": "imqueue.com <[email protected]> (https://imqueue.com)",
"license": "GPL-3.0-only",
"dependencies": {
"ioredis": "^5.6.1"
"ioredis": "^5.7.0"
},
"devDependencies": {
"@eslint/js": "^9.30.0",
"@eslint/js": "^9.33.0",
"@types/chai": "^5.2.2",
"@types/eslint__eslintrc": "^2.1.2",
"@types/mocha": "^10.0.0",
"@types/mock-require": "^3.0.0",
"@types/node": "^24.0.8",
"@types/node": "^24.2.1",
"@types/sinon": "^17.0.4",
"@types/yargs": "^17.0.33",
"@typescript-eslint/eslint-plugin": "^8.35.1",
"@typescript-eslint/parser": "^8.35.1",
"@typescript-eslint/typescript-estree": "^8.35.1",
"chai": "^5.2.0",
"codeclimate-test-reporter": "^0.5.1",
"coveralls-next": "^4.2.1",
"eslint": "^9.30.0",
"eslint-plugin-jsdoc": "^51.3.1",
"@typescript-eslint/eslint-plugin": "^8.39.0",
"@typescript-eslint/parser": "^8.39.0",
"@typescript-eslint/typescript-estree": "^8.39.0",
"chai": "^5.2.1",
"coveralls-next": "^5.0.0",
"eslint": "^9.33.0",
"eslint-plugin-jsdoc": "^52.0.4",
"mocha": "^11.7.1",
"mocha-lcov-reporter": "^1.3.0",
"mock-require": "^3.0.3",
"nyc": "^17.1.0",
"open": "^10.1.2",
"open": "^10.2.0",
"reflect-metadata": "^0.2.2",
"sinon": "^21.0.0",
"source-map-support": "^0.5.21",
"ts-node": "^10.9.2",
"typedoc": "^0.28.7",
"typescript": "^5.8.3",
"yargs": "^17.7.2"
"typedoc": "^0.28.9",
"typescript": "^5.9.2",
"yargs": "^18.0.0"
},
"main": "index.js",
"typescript": {
Expand Down
5 changes: 4 additions & 1 deletion src/UDPClusterManager.ts
Original file line number Diff line number Diff line change
Expand Up @@ -368,7 +368,10 @@ export class UDPClusterManager extends ClusterManager {
if (typeof socket.close === 'function') {
socket.removeAllListeners();
socket.close(() => {
socket?.unref();
// unref may be missing or not a function on mocked sockets
if (socket && typeof (socket as any).unref === 'function') {
socket.unref();
}

if (
socketKey
Expand Down
54 changes: 54 additions & 0 deletions test/ClusterManager.spec.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,54 @@
/*!
* ClusterManager additional tests
*
* I'm Queue Software Project
* Copyright (C) 2025 imqueue.com
*/
import './mocks';
import { expect } from 'chai';
import * as sinon from 'sinon';
import { ClusterManager, InitializedCluster } from '../src/ClusterManager';

class TestClusterManager extends ClusterManager {
public destroyed = false;
public constructor() { super(); }
public async destroy(): Promise<void> {
this.destroyed = true;
}
}

describe('ClusterManager.remove()', () => {
it('should call destroy when the last cluster is removed and destroy=true', async () => {
const cm = new TestClusterManager();
const cluster: InitializedCluster = cm.init({
add: () => undefined,
remove: () => undefined,
find: () => undefined,
});

// sanity: one cluster registered
expect((cm as any).clusters.length).to.equal(1);
const spy = sinon.spy(cm, 'destroy');

await cm.remove(cluster, true);

expect(spy.calledOnce).to.be.true;
expect((cm as any).clusters.length).to.equal(0);
expect(cm.destroyed).to.be.true;
});

it('should not call destroy when destroy=false', async () => {
const cm = new TestClusterManager();
const cluster: InitializedCluster = cm.init({
add: () => undefined,
remove: () => undefined,
find: () => undefined,
});

const spy = sinon.spy(cm, 'destroy');
await cm.remove(cluster.id, false);

expect(spy.called).to.be.false;
expect((cm as any).clusters.length).to.equal(0);
});
});
34 changes: 34 additions & 0 deletions test/ClusteredRedisQueue.addServer.defaultInit.spec.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,34 @@
/*!
* ClusteredRedisQueue.addServerWithQueueInitializing default param branch
*/
import './mocks';
import { expect } from 'chai';
import { ClusteredRedisQueue } from '../src';

describe('ClusteredRedisQueue.addServerWithQueueInitializing() default param', () => {
it('should use default initializeQueue=true when second param omitted', async () => {
const cq: any = new ClusteredRedisQueue('CQ-Default', {
logger: console,
cluster: [{ host: '127.0.0.1', port: 6379 }],
});
// prevent any actual start/subscription side-effects
(cq as any).state.started = false;
(cq as any).state.subscription = null;

const server = { host: '192.168.0.1', port: 6380 };
const initializedSpy = new Promise<void>((resolve) => {
cq['clusterEmitter'].once('initialized', () => resolve());
});

// Call without the second argument to hit default "true" branch
(cq as any).addServerWithQueueInitializing(server);

await initializedSpy; // should emit initialized when default is true

// Ensure the server added and queue length updated
expect((cq as any).servers.some((s: any) => s.host === server.host && s.port === server.port)).to.equal(true);
expect((cq as any).queueLength).to.equal((cq as any).imqs.length);

await cq.destroy();
});
});
32 changes: 32 additions & 0 deletions test/ClusteredRedisQueue.addServer.noInit.spec.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
/*!
* Cover ClusteredRedisQueue.addServerWithQueueInitializing with initializeQueue=false
*/
import './mocks';
import { expect } from 'chai';
import { ClusteredRedisQueue } from '../src';
import { ClusterManager } from '../src/ClusterManager';

const server = { host: '127.0.0.1', port: 6380 };

describe('ClusteredRedisQueue.addServerWithQueueInitializing(false)', () => {
it('should add server without initializing queue and not emit initialized', async () => {
const manager = new (ClusterManager as any)();
const cq: any = new ClusteredRedisQueue('NoInit', { clusterManagers: [manager] });

let initializedCalled = false;
(cq as any).clusterEmitter.on('initialized', () => { initializedCalled = true; });

// call private method via any to cover branch
(cq as any).addServerWithQueueInitializing(server, false);

// should have server and imq added
expect(cq.servers.length).to.be.greaterThan(0);
expect(cq.imqs.length).to.be.greaterThan(0);
// queueLength updated
expect(cq.queueLength).to.equal(cq.imqs.length);
// initialized not emitted
expect(initializedCalled).to.equal(false);

await cq.destroy();
});
});
48 changes: 48 additions & 0 deletions test/ClusteredRedisQueue.extra.spec.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,48 @@
/*!
* Additional tests for ClusteredRedisQueue event emitter proxy methods
*/
import './mocks';
import { expect } from 'chai';
import * as sinon from 'sinon';
import { ClusteredRedisQueue } from '../src';
import { ClusterManager } from '../src/ClusterManager';

const clusterConfig = {
cluster: [
{ host: '127.0.0.1', port: 6379 },
],
};

describe('ClusteredRedisQueue - EventEmitter proxy methods', () => {
it('should cover rawListeners/getMaxListeners/eventNames/listenerCount/emit', async () => {
const clusterManager = new (ClusterManager as any)();
const cq: any = new ClusteredRedisQueue('ProxyQueue', {
clusterManagers: [clusterManager],
});

// add underlying server and listener
cq.addServer(clusterConfig.cluster[0]);
const handler = sinon.spy();
cq.imqs[0].on('test', handler);

// set max listeners across emitters and verify getMaxListeners uses templateEmitter
cq.setMaxListeners(20);
expect(cq.getMaxListeners()).to.equal(20);

// collect raw listeners
const raw = cq.rawListeners('test');
expect(raw.length).to.be.greaterThan(0);

// event names come from underlying imq
const names = cq.eventNames();
expect(names).to.be.an('array');
expect(names.map(String)).to.include('test');

// listener count is aggregated via templateEmitter method applied on imq[0]
expect(cq.listenerCount('test')).to.equal(1);

// emit should return true
expect(cq.emit('test', 1, 2, 3)).to.equal(true);
expect(handler.calledOnce).to.be.true;
});
});
37 changes: 37 additions & 0 deletions test/ClusteredRedisQueue.initialize.spec.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,37 @@
/*!
* Additional tests for ClusteredRedisQueue.initializeQueue branches
*/
import './mocks';
import { expect } from 'chai';
import * as sinon from 'sinon';
import { ClusteredRedisQueue, RedisQueue } from '../src';
import { ClusterManager } from '../src/ClusterManager';

describe('ClusteredRedisQueue.initializeQueue()', () => {
it('should call imq.start when started and imq.subscribe when subscription is set', async () => {
const startStub = sinon.stub(RedisQueue.prototype as any, 'start').resolves(undefined);
const subscribeStub = sinon.stub(RedisQueue.prototype as any, 'subscribe').resolves();

const clusterManager = new (ClusterManager as any)();
const cq: any = new ClusteredRedisQueue('InitCover', { clusterManagers: [clusterManager] });

// mark started and set subscription using public APIs
await cq.start();
const channel = 'X';
const handler = () => undefined;
await cq.subscribe(channel, handler);

// adding a server triggers initializeQueue which should call start and subscribe
cq.addServer({ host: '127.0.0.1', port: 6453 });

// allow promises to resolve
await new Promise(res => setTimeout(res, 0));

expect(startStub.called).to.be.true;
expect(subscribeStub.called).to.be.true;

startStub.restore();
subscribeStub.restore();
await cq.destroy();
});
});
34 changes: 34 additions & 0 deletions test/ClusteredRedisQueue.matchServers.spec.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,34 @@
/*!
* Tests for ClusteredRedisQueue.matchServers combinations
*/
import './mocks';
import { expect } from 'chai';
import { ClusteredRedisQueue } from '../src';

// Access private static via casting
const match = (ClusteredRedisQueue as any).matchServers as (
source: any, target: any, strict?: boolean
) => boolean;

describe('ClusteredRedisQueue.matchServers()', () => {
it('should return sameAddress when no ids provided', () => {
expect(match({ host: 'h', port: 1 }, { host: 'h', port: 1 })).to.be.true;
expect(match({ host: 'h', port: 1 }, { host: 'h', port: 2 })).to.be.false;
});

it('should use strict logic when strict=true', () => {
// same id and same address -> true
expect(match({ id: 'a', host: 'h', port: 1 }, { id: 'a', host: 'h', port: 1 }, true)).to.be.true;
// same id but different address -> false
expect(match({ id: 'a', host: 'h', port: 1 }, { id: 'a', host: 'h', port: 2 }, true)).to.be.false;
// different id but same address -> false
expect(match({ id: 'a', host: 'h', port: 1 }, { id: 'b', host: 'h', port: 1 }, true)).to.be.false;
});

it('should use relaxed logic when strict=false', () => {
// id matches -> true even if address differs
expect(match({ id: 'a', host: 'h', port: 1 }, { id: 'a', host: 'h', port: 2 }, false)).to.be.true;
// address matches -> true even if id differs
expect(match({ id: 'a', host: 'h', port: 1 }, { id: 'b', host: 'h', port: 1 }, false)).to.be.true;
});
});
10 changes: 5 additions & 5 deletions test/ClusteredRedisQueue.ts
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@
* purchase a proprietary commercial license. Please contact us at
* <[email protected]> to get commercial licensing options.
*/
import * as mocks from './mocks';
import { logger } from './mocks';
import { expect } from 'chai';
import * as sinon from 'sinon';
import { ClusteredRedisQueue } from '../src';
Expand All @@ -30,7 +30,7 @@ import { ClusterManager } from '../src/ClusterManager';
process.setMaxListeners(100);

const clusterConfig = {
logger: mocks.logger,
logger,
cluster: [{
host: '127.0.0.1',
port: 7777
Expand Down Expand Up @@ -163,14 +163,14 @@ describe('ClusteredRedisQueue', function() {
'TestClusteredQueueOne',
{
clusterManagers: [clusterManager],
logger: mocks.logger,
logger,
},
);
const cqTwo: any = new ClusteredRedisQueue(
'TestClusteredQueueTwo',
{
clusterManagers: [clusterManager],
logger: mocks.logger,
logger,
},
);
const message = { 'hello': 'world' };
Expand Down Expand Up @@ -238,7 +238,7 @@ describe('ClusteredRedisQueue', function() {
'TestClusteredQueue',
{
clusterManagers: [clusterManager],
logger: mocks.logger,
logger,
},
);
const channel = 'TestChannel';
Expand Down
Loading