Give the held-message flow one place a reader can follow it
Mirror / push (push) Successful in 6s
Test / lint (pull_request) Successful in 24s
Test / test (18) (pull_request) Successful in 32s
Test / test (20) (pull_request) Successful in 31s
Test / test (24) (pull_request) Successful in 32s
Test / test (22) (pull_request) Successful in 32s
Test / test (26) (pull_request) Successful in 31s
Mirror / push (push) Successful in 6s
Test / lint (pull_request) Successful in 24s
Test / test (18) (pull_request) Successful in 32s
Test / test (20) (pull_request) Successful in 31s
Test / test (24) (pull_request) Successful in 32s
Test / test (22) (pull_request) Successful in 32s
Test / test (26) (pull_request) Successful in 31s
This commit is contained in:
+28
-1
@@ -25,7 +25,10 @@ type HoldEntry = {
|
||||
release: () => void;
|
||||
};
|
||||
|
||||
/** One message offered to the application, held until it is answered or every listener gives up. */
|
||||
/**
|
||||
* One message offered to the application. A drain waits on it until the first of: `answered()`,
|
||||
* every listener that took it rejecting, no listener taking it, its deadline, or the link going.
|
||||
*/
|
||||
export class MessageHold {
|
||||
private readonly entry: HoldEntry;
|
||||
private working: number;
|
||||
@@ -64,6 +67,8 @@ export class HeldMessages {
|
||||
private readonly idleWaiters = new IdleWaiters();
|
||||
private readonly log: SmppLog;
|
||||
private readonly maxOctets: number;
|
||||
/** A rejecting listener hands the message back as an `unknown`, so its hold is found by identity. */
|
||||
private readonly offered = new WeakMap<object, MessageHold>();
|
||||
|
||||
constructor(options: HeldMessagesOptions) {
|
||||
this.held = new ExpiringGroups({
|
||||
@@ -112,6 +117,28 @@ export class HeldMessages {
|
||||
return hold;
|
||||
}
|
||||
|
||||
/** Holds a message, builds what the application is handed, and emits it. */
|
||||
offer<T extends object>(
|
||||
pduObjs: PduObject[],
|
||||
listeners: number,
|
||||
build: (hold: MessageHold) => T,
|
||||
emit: (message: T) => boolean,
|
||||
): void {
|
||||
const hold = this.hold(pduObjs, listeners);
|
||||
const message = build(hold);
|
||||
|
||||
this.offered.set(message, hold);
|
||||
|
||||
if (!emit(message)) hold.release();
|
||||
}
|
||||
|
||||
/** One listener gave up on a message; the last one to do so is what releases it. */
|
||||
listenerRejected(message: unknown): void {
|
||||
if (typeof message !== 'object' || message === null) return;
|
||||
|
||||
this.offered.get(message)?.listenerGaveUp();
|
||||
}
|
||||
|
||||
private has(pduObjs: PduObject[]): boolean {
|
||||
const key = keyOf(pduObjs);
|
||||
|
||||
|
||||
@@ -10,7 +10,6 @@ import type { Result, VoidResult } from './result.ts';
|
||||
import type { SmppLog } from './log.ts';
|
||||
import type { Sms, SmsHandlers, SmsInput } from './sms.ts';
|
||||
import type { SmsIdFormat } from './sms-id.ts';
|
||||
import type { MessageHold } from './held-messages.ts';
|
||||
import { HeldMessages } from './held-messages.ts';
|
||||
import { Reassembler, decodeSegments } from './reassembly.ts';
|
||||
import { bindCommands, defaults, standsInFor } from './session-options.ts';
|
||||
@@ -85,8 +84,6 @@ export class IncomingRequests {
|
||||
private readonly deps: IncomingDeps;
|
||||
private readonly dlrMerger: DlrMerger;
|
||||
private readonly held: HeldMessages;
|
||||
/** The rejection handler is handed the Sms back as an `unknown`, so its hold is found by identity. */
|
||||
private readonly holds = new WeakMap<object, MessageHold>();
|
||||
private readonly log: SmppLog;
|
||||
private readonly reassembler: Reassembler;
|
||||
private readonly smsIdFormat: SmsIdFormat;
|
||||
@@ -172,11 +169,8 @@ export class IncomingRequests {
|
||||
this.reassembler.clear();
|
||||
}
|
||||
|
||||
/** One `sms` listener gave up on a message; the last one to do so is what releases the hold. */
|
||||
listenerRejected(sms: unknown): void {
|
||||
if (typeof sms !== 'object' || sms === null) return;
|
||||
|
||||
this.holds.get(sms)?.listenerGaveUp();
|
||||
this.held.listenerRejected(sms);
|
||||
}
|
||||
|
||||
/** Waits out the messages the application still holds, and says how many it never answered. */
|
||||
@@ -310,9 +304,8 @@ export class IncomingRequests {
|
||||
if (!first) return;
|
||||
|
||||
const generation = this.linkGeneration;
|
||||
const hold = this.held.hold(pduObjs, this.deps.smsListeners());
|
||||
|
||||
const sms = this.deps.createSms({
|
||||
this.held.offer(pduObjs, this.deps.smsListeners(), hold => this.deps.createSms({
|
||||
answeredAs,
|
||||
from: paramText(first.params.source_addr),
|
||||
message: decodeSegments(pduObjs),
|
||||
@@ -325,10 +318,6 @@ export class IncomingRequests {
|
||||
lostLink: () => this.linkGeneration !== generation,
|
||||
onAnswered: () => { hold.answered(); },
|
||||
send: input => (hold.isHeld() ? this.deps.sendPastDrain(input) : this.deps.send(input)),
|
||||
});
|
||||
|
||||
this.holds.set(sms, hold);
|
||||
|
||||
if (!this.deps.offerSms(sms)) hold.release();
|
||||
}), sms => this.deps.offerSms(sms));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -204,9 +204,6 @@ and 5. Every seat ranked the session's lifecycle hardest and least wanted to mod
|
||||
|
||||
- [ ] **Lift Locality to 7, and confirm it with a scoring run.** A run reading 7.0 or above also
|
||||
retires the #30 decision. The sub-items are what the 2026-09-28 run named, most seats first.
|
||||
- [ ] **Give the held-message flow one place a reader can follow it.** Whether a drain still waits
|
||||
on a message is spread over `emitSms()`, `MessageHold`, the session's rejection route and
|
||||
`Sms.isHeld()`. Four seats.
|
||||
- [ ] **Shrink the `IncomingDeps` closure bag.** 16 lambdas, six of them repeated in
|
||||
`SmsHandlers`, which makes every inbound call path indirect. Two seats.
|
||||
|
||||
|
||||
Reference in New Issue
Block a user