Skip to content

Feat/xread maxcount maxsize #3335

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
nkaradzhov wants to merge 2 commits into
base: master
Choose a base branch
from
+203 −7
Open

Feat/xread maxcount maxsize #3335

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
9120345
feat(client): add MAXCOUNT and MAXSIZE options to XREAD and XREADGROUP
nkaradzhov Jul 13, 2026
530b0b6
test(client): add MAXSIZE behavior tests; fix XREAD COUNT zero handling
nkaradzhov Jul 13, 2026
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
81 changes: 81 additions & 0 deletions packages/client/lib/commands/XREAD.spec.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,6 +89,45 @@ describe('XREAD', () => {
['XREAD', 'COUNT', '1', 'BLOCK', '0', 'STREAMS', 'key', '0-0']
);
});

it('with MAXCOUNT', () => {
assert.deepEqual(
parseArgs(XREAD, {
key: 'key',
id: '0-0'
}, {
MAXCOUNT: 3
}),
['XREAD', 'MAXCOUNT', '3', 'STREAMS', 'key', '0-0']
);
});

it('with MAXSIZE', () => {
assert.deepEqual(
parseArgs(XREAD, {
key: 'key',
id: '0-0'
}, {
MAXSIZE: 65536
}),
['XREAD', 'MAXSIZE', '65536', 'STREAMS', 'key', '0-0']
);
});

it('with COUNT, MAXCOUNT, MAXSIZE, BLOCK', () => {
assert.deepEqual(
parseArgs(XREAD, {
key: 'key',
id: '0-0'
}, {
COUNT: 2,
MAXCOUNT: 3,
MAXSIZE: 65536,
BLOCK: 0
}),
['XREAD', 'COUNT', '2', 'MAXCOUNT', '3', 'MAXSIZE', '65536', 'BLOCK', '0', 'STREAMS', 'key', '0-0']
);
});
});

testUtils.testAll('client.xRead', async client => {
Expand Down Expand Up @@ -146,4 +185,46 @@ describe('XREAD', () => {
}
});

testUtils.testWithClientIfVersionWithinRange([[8, 10], 'LATEST'], 'client.xRead with MAXCOUNT caps entries cumulatively across streams', async client => {
await Promise.all([
client.xAdd('{t}s1', '1-0', { field: 'v1' }),
client.xAdd('{t}s1', '2-0', { field: 'v2' }),
client.xAdd('{t}s2', '1-0', { field: 'v3' }),
client.xAdd('{t}s2', '2-0', { field: 'v4' })
]);

const reply = await client.xRead([
{ key: '{t}s1', id: '0' },
{ key: '{t}s2', id: '0' }
], {
MAXCOUNT: 3
});

assert.ok(reply);
const total = reply.reduce((sum, stream) => sum + stream.messages.length, 0);
assert.equal(total, 3);
}, GLOBAL.SERVERS.OPEN);

testUtils.testWithClientIfVersionWithinRange([[8, 10], 'LATEST'], 'client.xRead with MAXSIZE caps reply size cumulatively across streams', async client => {
await Promise.all([
client.xAdd('{t}s1', '1-0', { field: 'v1' }),
client.xAdd('{t}s1', '2-0', { field: 'v2' }),
client.xAdd('{t}s2', '1-0', { field: 'v3' }),
client.xAdd('{t}s2', '2-0', { field: 'v4' })
]);

// MAXSIZE is a soft cumulative byte cap across all streams; the server always
// returns at least one entry, so MAXSIZE of 1 byte yields exactly one entry total.
const reply = await client.xRead([
{ key: '{t}s1', id: '0' },
{ key: '{t}s2', id: '0' }
], {
MAXSIZE: 1
});

assert.ok(reply);
const total = reply.reduce((sum, stream) => sum + stream.messages.length, 0);
assert.equal(total, 1);
}, GLOBAL.SERVERS.OPEN);

});
16 changes: 14 additions & 2 deletions packages/client/lib/commands/XREAD.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,12 +39,16 @@ export function pushXReadStreams(parser: CommandParser, streams: XReadStreams) {

/**
* Options for the XREAD command
*
*
* @property COUNT - Limit the number of entries returned per stream
* @property MAXCOUNT - Cumulative cap on the total number of entries returned across all streams (Redis 8.10+)
* @property MAXSIZE - Soft cumulative cap on the total server reply size in bytes across all streams (Redis 8.10+)
* @property BLOCK - Milliseconds to block waiting for new entries (0 for indefinite)
*/
export interface XReadOptions {
COUNT?: number;
MAXCOUNT?: number;
MAXSIZE?: number;
BLOCK?: number;
}

Expand All @@ -53,10 +57,18 @@ export default {
parseCommand(parser: CommandParser, streams: XReadStreams, options?: XReadOptions) {
parser.push('XREAD');

if (options?.COUNT) {
if (options?.COUNT !== undefined) {
parser.push('COUNT', options.COUNT.toString());
}

if (options?.MAXCOUNT !== undefined) {
parser.push('MAXCOUNT', options.MAXCOUNT.toString());
}

if (options?.MAXSIZE !== undefined) {
parser.push('MAXSIZE', options.MAXSIZE.toString());
}

if (options?.BLOCK !== undefined) {
parser.push('BLOCK', options.BLOCK.toString());
}
Expand Down
93 changes: 92 additions & 1 deletion packages/client/lib/commands/XREADGROUP.spec.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -117,7 +117,48 @@ describe('XREADGROUP', () => {
NOACK: true,
CLAIM: 100
}),
['XREADGROUP', 'GROUP', 'group', 'consumer', 'COUNT', '1', 'BLOCK', '0', 'NOACK', 'CLAIM', '100', 'STREAMS', 'key', '0-0']
['XREADGROUP', 'GROUP', 'group', 'consumer', 'COUNT', '1', 'BLOCK', '0', 'CLAIM', '100', 'NOACK', 'STREAMS', 'key', '0-0']
);
});

it('with MAXCOUNT', () => {
assert.deepEqual(
parseArgs(XREADGROUP, 'group', 'consumer', {
key: 'key',
id: '0-0'
}, {
MAXCOUNT: 3
}),
['XREADGROUP', 'GROUP', 'group', 'consumer', 'MAXCOUNT', '3', 'STREAMS', 'key', '0-0']
);
});

it('with MAXSIZE', () => {
assert.deepEqual(
parseArgs(XREADGROUP, 'group', 'consumer', {
key: 'key',
id: '0-0'
}, {
MAXSIZE: 65536
}),
['XREADGROUP', 'GROUP', 'group', 'consumer', 'MAXSIZE', '65536', 'STREAMS', 'key', '0-0']
);
});

it('with COUNT, MAXCOUNT, MAXSIZE, BLOCK, CLAIM, NOACK', () => {
assert.deepEqual(
parseArgs(XREADGROUP, 'group', 'consumer', {
key: 'key',
id: '0-0'
}, {
COUNT: 2,
MAXCOUNT: 3,
MAXSIZE: 65536,
BLOCK: 0,
CLAIM: 100,
NOACK: true
}),
['XREADGROUP', 'GROUP', 'group', 'consumer', 'COUNT', '2', 'MAXCOUNT', '3', 'MAXSIZE', '65536', 'BLOCK', '0', 'CLAIM', '100', 'NOACK', 'STREAMS', 'key', '0-0']
);
});
});
Expand Down Expand Up @@ -233,4 +274,54 @@ describe('XREADGROUP', () => {
assert.equal(typeof readGroupReply[0].messages[0].millisElapsedFromDelivery, 'number');
assert.equal(typeof readGroupReply[0].messages[0].deliveriesCounter, 'number');
}, GLOBAL.SERVERS.OPEN);

testUtils.testWithClientIfVersionWithinRange([[8, 10], 'LATEST'], 'xReadGroup - MAXCOUNT caps entries cumulatively across streams', async client => {
await Promise.all([
client.xGroupCreate('{t}s1', 'group', '0', { MKSTREAM: true }),
client.xGroupCreate('{t}s2', 'group', '0', { MKSTREAM: true })
]);
await Promise.all([
client.xAdd('{t}s1', '1-0', { field: 'v1' }),
client.xAdd('{t}s1', '2-0', { field: 'v2' }),
client.xAdd('{t}s2', '1-0', { field: 'v3' }),
client.xAdd('{t}s2', '2-0', { field: 'v4' })
]);

const reply = await client.xReadGroup('group', 'consumer', [
{ key: '{t}s1', id: '>' },
{ key: '{t}s2', id: '>' }
], {
MAXCOUNT: 3
});

assert.ok(reply);
const total = reply.reduce((sum, stream) => sum + stream.messages.length, 0);
assert.equal(total, 3);
}, GLOBAL.SERVERS.OPEN);

testUtils.testWithClientIfVersionWithinRange([[8, 10], 'LATEST'], 'xReadGroup - MAXSIZE caps reply size cumulatively across streams', async client => {
await Promise.all([
client.xGroupCreate('{t}s1', 'group', '0', { MKSTREAM: true }),
client.xGroupCreate('{t}s2', 'group', '0', { MKSTREAM: true })
]);
await Promise.all([
client.xAdd('{t}s1', '1-0', { field: 'v1' }),
client.xAdd('{t}s1', '2-0', { field: 'v2' }),
client.xAdd('{t}s2', '1-0', { field: 'v3' }),
client.xAdd('{t}s2', '2-0', { field: 'v4' })
]);

// MAXSIZE is a soft cumulative byte cap across all streams; the server always
// returns at least one entry, so MAXSIZE of 1 byte yields exactly one entry total.
const reply = await client.xReadGroup('group', 'consumer', [
{ key: '{t}s1', id: '>' },
{ key: '{t}s2', id: '>' }
], {
MAXSIZE: 1
});

assert.ok(reply);
const total = reply.reduce((sum, stream) => sum + stream.messages.length, 0);
assert.equal(total, 1);
}, GLOBAL.SERVERS.OPEN);
});
20 changes: 16 additions & 4 deletions packages/client/lib/commands/XREADGROUP.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,12 +7,16 @@ import { transformStreamsMessagesReplyResp2, transformStreamsMessagesReplyResp3C
* Options for the XREADGROUP command
*
* @property COUNT - Limit the number of entries returned per stream
* @property MAXCOUNT - Cumulative cap on the total number of entries returned across all streams (Redis 8.10+)
* @property MAXSIZE - Soft cumulative cap on the total server reply size in bytes across all streams (Redis 8.10+)
* @property BLOCK - Milliseconds to block waiting for new entries (0 for indefinite)
* @property NOACK - Skip adding the message to the PEL (Pending Entries List)
* @property CLAIM - Prepend PEL entries that are at least this many milliseconds old
*/
export interface XReadGroupOptions {
COUNT?: number;
MAXCOUNT?: number;
MAXSIZE?: number;
BLOCK?: number;
NOACK?: boolean;
CLAIM?: number;
Expand All @@ -33,18 +37,26 @@ export default {
parser.push('COUNT', options.COUNT.toString());
}

if (options?.BLOCK !== undefined) {
parser.push('BLOCK', options.BLOCK.toString());
if (options?.MAXCOUNT !== undefined) {
parser.push('MAXCOUNT', options.MAXCOUNT.toString());
}

if (options?.NOACK) {
parser.push('NOACK');
if (options?.MAXSIZE !== undefined) {
parser.push('MAXSIZE', options.MAXSIZE.toString());
}

if (options?.BLOCK !== undefined) {
parser.push('BLOCK', options.BLOCK.toString());
}

if (options?.CLAIM !== undefined) {
parser.push('CLAIM', options.CLAIM.toString());
}

if (options?.NOACK) {
parser.push('NOACK');
}

pushXReadStreams(parser, streams);
},
/**
Expand Down