Sweep held messages before each hold, so expiry needs no live timer
This commit is contained in:
+11
-1
@@ -6,6 +6,8 @@ import { IdleWaiters } from './idle-waiters.ts';
|
|||||||
export type HeldMessagesOptions = {
|
export type HeldMessagesOptions = {
|
||||||
log: SmppLog;
|
log: SmppLog;
|
||||||
max: number;
|
max: number;
|
||||||
|
/** Injected so expiry can be exercised without a wall clock. */
|
||||||
|
now?: (() => number) | undefined;
|
||||||
timeout: number;
|
timeout: number;
|
||||||
};
|
};
|
||||||
|
|
||||||
@@ -26,6 +28,7 @@ export class HeldMessages {
|
|||||||
constructor(options: HeldMessagesOptions) {
|
constructor(options: HeldMessagesOptions) {
|
||||||
this.held = new ExpiringGroups({
|
this.held = new ExpiringGroups({
|
||||||
max: options.max,
|
max: options.max,
|
||||||
|
now: options.now,
|
||||||
onSweep: () => { this.sweep(); },
|
onSweep: () => { this.sweep(); },
|
||||||
timeout: options.timeout,
|
timeout: options.timeout,
|
||||||
});
|
});
|
||||||
@@ -33,12 +36,18 @@ export class HeldMessages {
|
|||||||
this.max = options.max;
|
this.max = options.max;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
get size(): number {
|
||||||
|
return this.held.size;
|
||||||
|
}
|
||||||
|
|
||||||
/** An application that answers no message at all may not grow this without end. */
|
/** An application that answers no message at all may not grow this without end. */
|
||||||
hold(pduObjs: PduObject[]): void {
|
hold(pduObjs: PduObject[]): void {
|
||||||
const key = keyOf(pduObjs);
|
const key = keyOf(pduObjs);
|
||||||
|
|
||||||
if (key === undefined) return;
|
if (key === undefined) return;
|
||||||
|
|
||||||
|
this.sweep();
|
||||||
|
|
||||||
if (this.held.get(key)) {
|
if (this.held.get(key)) {
|
||||||
this.log.warn('heldMessages - replacing a message on a re-used sequence number', { seqNr: key });
|
this.log.warn('heldMessages - replacing a message on a re-used sequence number', { seqNr: key });
|
||||||
} else if (this.held.full) {
|
} else if (this.held.full) {
|
||||||
@@ -86,7 +95,8 @@ export class HeldMessages {
|
|||||||
this.log.warn('heldMessages - buffer full, dropping the oldest message', { max: this.max, seqNr });
|
this.log.warn('heldMessages - buffer full, dropping the oldest message', { max: this.max, seqNr });
|
||||||
}
|
}
|
||||||
|
|
||||||
private sweep(): void {
|
/** Drops every message past its deadline. Runs before each hold and on its own timer. */
|
||||||
|
sweep(): void {
|
||||||
const expired = this.held.takeExpired();
|
const expired = this.held.takeExpired();
|
||||||
|
|
||||||
if (expired.length === 0) return;
|
if (expired.length === 0) return;
|
||||||
|
|||||||
@@ -879,7 +879,7 @@ describe('held message bounds', () => {
|
|||||||
return [heldPdu(seqNr)];
|
return [heldPdu(seqNr)];
|
||||||
}
|
}
|
||||||
|
|
||||||
test('drops the message held longest rather than holding every one', async () => {
|
test('drops the message held longest rather than holding every one', () => {
|
||||||
const held = new HeldMessages({ log: silentLog, max: 2, timeout: 10_000 });
|
const held = new HeldMessages({ log: silentLog, max: 2, timeout: 10_000 });
|
||||||
const oldest = message(1);
|
const oldest = message(1);
|
||||||
|
|
||||||
@@ -887,17 +887,25 @@ describe('held message bounds', () => {
|
|||||||
held.hold(message(2));
|
held.hold(message(2));
|
||||||
held.hold(message(3));
|
held.hold(message(3));
|
||||||
|
|
||||||
|
assert.equal(held.size, 2);
|
||||||
assert.equal(held.has(oldest), false);
|
assert.equal(held.has(oldest), false);
|
||||||
assert.equal(await held.idle(1, undefined), 2);
|
|
||||||
|
held.clear();
|
||||||
});
|
});
|
||||||
|
|
||||||
test('gives up on a message the application never answers', async () => {
|
test('gives up on a message the application never answers', () => {
|
||||||
const held = new HeldMessages({ log: silentLog, max: 10, timeout: 20 });
|
let now = 0;
|
||||||
|
const held = new HeldMessages({ log: silentLog, max: 10, now: () => now, timeout: 60 });
|
||||||
|
|
||||||
held.hold(message(1));
|
held.hold(message(1));
|
||||||
|
now = 61;
|
||||||
|
|
||||||
assert.equal(await held.idle(1, undefined), 1);
|
// The next message sweeps the one that expired, so only the new one is still waited for.
|
||||||
assert.equal(await held.idle(1000, undefined), 0);
|
held.hold(message(2));
|
||||||
|
|
||||||
|
assert.equal(held.size, 1);
|
||||||
|
|
||||||
|
held.clear();
|
||||||
});
|
});
|
||||||
|
|
||||||
// A receipt cannot be resent wholesale without duplicating the segments that landed, so sendDlr()
|
// A receipt cannot be resent wholesale without duplicating the segments that landed, so sendDlr()
|
||||||
|
|||||||
Reference in New Issue
Block a user