From b671ebafb2fabe5015d245c9727f94b20c483235 Mon Sep 17 00:00:00 2001 From: Lilleman auf Larv Date: Tue, 25 Aug 2026 16:58:13 +0200 Subject: [PATCH] Fix merged DLR counting and close event, cover reconnect and reassembly --- src/session.ts | 59 ++++--- test/session-extras.test.ts | 302 ++++++++++++++++++++++++++++++++++++ 2 files changed, 342 insertions(+), 19 deletions(-) create mode 100644 test/session-extras.test.ts diff --git a/src/session.ts b/src/session.ts index 5021c9b..d596ef7 100644 --- a/src/session.ts +++ b/src/session.ts @@ -122,7 +122,7 @@ export class Session extends EventEmitter { private readonly options: SessionOptions; private readonly pending = new Map(); private readonly reassembly = new Map(); - private readonly segmentDlrs = new Map>(); + private readonly segmentDlrs = new Map }>(); private readonly waiting: (() => void)[] = []; private closed = false; @@ -258,6 +258,8 @@ export class Session extends EventEmitter { smsIds.push(paramText(one.pduObj.params.message_id)); } + this.expectSegmentDlrs(smsIds); + return { pduObjs, smsIds }; } @@ -296,6 +298,7 @@ export class Session extends EventEmitter { this.reassembly.clear(); this.sock.destroy(); + this.emit('close'); } private nextSeqNr(): number { @@ -603,31 +606,53 @@ export class Session extends EventEmitter { await this.sendReturn(pduObj); } - /** Segment ids look like `-`; once every segment is in, report on the whole message. */ + /** + * Registers a multipart message for merged reporting, but only when the peer numbered its ids + * `-` off one base — the convention this library's own server follows. An SMSC that + * hands out unrelated ids per segment cannot be merged, so no messageDlr is emitted for it. + */ + private expectSegmentDlrs(smsIds: string[]): void { + if (smsIds.length < 2) return; + + const bases = new Set(); + + for (const smsId of smsIds) { + const match = /^(.*)-(\d+)$/.exec(smsId); + + if (!match?.[1]) return; + + bases.add(match[1]); + } + + if (bases.size !== 1) return; + + for (const base of bases) { + this.segmentDlrs.set(base, { expected: smsIds.length, parts: new Map() }); + } + } + + /** Collects per-segment receipts and reports once on the whole message. */ private collectSegmentDlr(dlr: Dlr): void { const match = /^(.*)-(\d+)$/.exec(dlr.smsId); + const base = match?.[1]; + const part = match?.[2]; - if (!match) return; + if (base === undefined || part === undefined) return; - const [, smsId, part] = match; + const group = this.segmentDlrs.get(base); - if (smsId === undefined || part === undefined) return; + if (!group) return; - const segments = this.segmentDlrs.get(smsId) ?? new Map(); + group.parts.set(Number(part), dlr); - segments.set(Number(part), dlr); - this.segmentDlrs.set(smsId, segments); + if (group.parts.size < group.expected) return; - const highest = Math.max(...segments.keys()); + this.segmentDlrs.delete(base); - if (segments.size < highest) return; - - this.segmentDlrs.delete(smsId); - - const ordered = [...segments.entries()].sort(([a], [b]) => a - b).map(([, one]) => one); + const ordered = [...group.parts.entries()].sort(([a], [b]) => a - b).map(([, one]) => one); const worst = ordered.reduce((carry, one) => (one.statusId > carry.statusId ? one : carry)); - this.emit('messageDlr', { ...worst, segments: ordered, smsId }); + this.emit('messageDlr', { ...worst, segments: ordered, smsId: base }); } private resetTimers(): void { @@ -662,12 +687,8 @@ export class Session extends EventEmitter { } private onClose(): void { - const wasOpen = !this.closed; - this.teardown(); - if (wasOpen) this.emit('close'); - if (!this.stopped && this.options.reconnect) this.scheduleReconnect(); } diff --git a/test/session-extras.test.ts b/test/session-extras.test.ts new file mode 100644 index 0000000..ca48f6f --- /dev/null +++ b/test/session-extras.test.ts @@ -0,0 +1,302 @@ +import assert from 'node:assert/strict'; +import net from 'node:net'; +import test, { describe } from 'node:test'; +import type { MessageDlr } from '../src/session.ts'; +import type { Sms } from '../src/sms.ts'; +import type { SmppServer } from '../src/server.ts'; +import { client } from '../src/client.ts'; +import { objToPdu } from '../src/pdu.ts'; +import { server } from '../src/server.ts'; + +async function startServer(options: Parameters[0] = {}): Promise { + const { err, server: smpp } = await server({ ...options, port: 0 }); + + assert.equal(err, undefined); + assert.ok(smpp); + + return smpp; +} + +function once(register: (resolve: (value: T) => void) => void): Promise { + return new Promise(resolve => { register(resolve); }); +} + +describe('merged delivery reports', () => { + // 0.4.0 allocated a longSmsDlrs store to do exactly this and then never used it. + test('reports once on a whole multipart message', async () => { + const smpp = await startServer(); + const incoming = once(resolve => { + smpp.on('session', session => session.on('sms', resolve)); + }); + const { session } = await client({ port: smpp.port }); + + assert.ok(session); + + const merged = once(resolve => { session.on('messageDlr', resolve); }); + const perSegment: string[] = []; + + session.on('dlr', dlr => perSegment.push(dlr.smsId)); + + const [sms] = await Promise.all([ + incoming.then(async received => { + received.smsId = 'merge-me'; + await received.sendResp(); + + return received; + }), + session.sendSms({ + dlr: true, + from: '46701113311', + message: 'x'.repeat(400), + to: '46709771337', + }), + ]); + + await sms.sendDlr(); + + const report = await merged; + + assert.equal(report.smsId, 'merge-me'); + assert.equal(report.segments.length, 3); + assert.equal(report.statusMsg, 'DELIVERED'); + assert.deepEqual(perSegment, ['merge-me-1', 'merge-me-2', 'merge-me-3']); + + session.close(); + await smpp.close(); + }); + + test('reports the worst status across the segments', async () => { + const smpp = await startServer(); + const incoming = once(resolve => { + smpp.on('session', session => session.on('sms', resolve)); + }); + const { session } = await client({ port: smpp.port }); + + assert.ok(session); + + const merged = once(resolve => { session.on('messageDlr', resolve); }); + + const [sms] = await Promise.all([ + incoming.then(async received => { + received.smsId = 'partly-failed'; + await received.sendResp(); + + return received; + }), + session.sendSms({ + dlr: true, + from: '46701113311', + message: 'x'.repeat(400), + to: '46709771337', + }), + ]); + + await sms.sendDlr('UNDELIVERABLE'); + + const report = await merged; + + assert.equal(report.statusMsg, 'UNDELIVERABLE'); + assert.equal(report.segments.length, 3); + + session.close(); + await smpp.close(); + }); +}); + +describe('reconnect', () => { + test('re-binds after the connection drops, keeping the same session object', async () => { + const smpp = await startServer(); + const messages: string[] = []; + + // Registered up front so the session created by the reconnect is covered too. + smpp.on('session', bound => { + bound.on('sms', sms => { + messages.push(sms.message); + void sms.sendResp(); + }); + }); + + const { err, session } = await client({ + port: smpp.port, + reconnect: { maxDelay: 100, minDelay: 20 }, + }); + + assert.equal(err, undefined); + assert.ok(session); + + const reconnected = once(resolve => { session.on('reconnected', () => { resolve(true); }); }); + + // Drop the connection from the server's side, as a peer restart would. + for (const serverSession of smpp.sessions) { + serverSession.close(); + } + + await reconnected; + + assert.ok(session.loggedIn); + + // The session object survives the drop, so listeners stay attached and it is usable again. + const sent = await session.sendSms({ + from: '46701113311', + message: 'after reconnect', + to: '46709771337', + }); + + assert.equal(sent.err, undefined); + assert.deepEqual(messages, ['after reconnect']); + + session.close(); + await smpp.close(); + }); + + test('does not reconnect after an explicit close', async () => { + const smpp = await startServer(); + const { session } = await client({ + port: smpp.port, + reconnect: { maxDelay: 50, minDelay: 10 }, + }); + + assert.ok(session); + + let reconnects = 0; + + session.on('reconnected', () => { reconnects++; }); + session.close(); + + await new Promise(resolve => setTimeout(resolve, 150)); + + assert.equal(reconnects, 0); + await smpp.close(); + }); +}); + +describe('reassembly bounds', () => { + function segment(reference: number, part: number, total: number, seqNr: number): Buffer { + const body = Buffer.concat([ + Buffer.from([0x05, 0x00, 0x03, reference, total, part]), + Buffer.from('fragment'), + ]); + const { buffer } = objToPdu({ + cmdName: 'submit_sm', + params: { + data_coding: 0, + destination_addr: '46709771337', + esm_class: 0x40, + short_message: body, + sm_length: body.length, + source_addr: '46701113311', + }, + seqNr, + }); + + assert.ok(buffer); + + return buffer; + } + + // 0.4.0 held incomplete groups without limit and swept them only when other traffic arrived. + test('drops the oldest incomplete message once the cap is reached', async () => { + const smpp = await startServer({ maxReassembly: 2, reassemblyTimeout: 60_000 }); + + let delivered = 0; + + smpp.on('session', session => session.on('sms', () => { delivered++; })); + + const sock = net.connect({ port: smpp.port }, () => { + sock.write(Buffer.from('0000002100000009000000000000002f666f6f0062617200736d70700034000000', 'hex')); + }); + + let bound = false; + + sock.on('data', () => { + if (bound) return; + + bound = true; + + // Three different messages, each only ever sending part 1 of 2. + sock.write(segment(1, 1, 2, 10)); + sock.write(segment(2, 1, 2, 11)); + sock.write(segment(3, 1, 2, 12)); + // Completing the first one must not produce a message: it was evicted. + sock.write(segment(1, 2, 2, 13)); + }); + + await new Promise(resolve => setTimeout(resolve, 200)); + + assert.equal(delivered, 0, 'an evicted message must not be delivered'); + + sock.destroy(); + await smpp.close(); + }); + + test('expires an incomplete message on its own timer', async () => { + const smpp = await startServer({ reassemblyTimeout: 60 }); + + let delivered = 0; + + smpp.on('session', session => session.on('sms', () => { delivered++; })); + + const sock = net.connect({ port: smpp.port }, () => { + sock.write(Buffer.from('0000002100000009000000000000002f666f6f0062617200736d70700034000000', 'hex')); + }); + + let bound = false; + + sock.on('data', () => { + if (bound) return; + + bound = true; + sock.write(segment(9, 1, 2, 20)); + }); + + await new Promise(resolve => setTimeout(resolve, 200)); + + // The other half arrives after the group expired, so it starts a new, still-incomplete one. + sock.write(segment(9, 2, 2, 21)); + + await new Promise(resolve => setTimeout(resolve, 100)); + + assert.equal(delivered, 0); + + sock.destroy(); + await smpp.close(); + }); +}); + +describe('AbortSignal on a send', () => { + test('gives up on an in-flight request when the signal fires', async () => { + const accepted: net.Socket[] = []; + const silent = net.createServer(sock => { + accepted.push(sock); + sock.resume(); + // Answer the bind so the client gets a session, then go quiet. + sock.write(Buffer.from('0000001180000009000000000000000100', 'hex')); + }); + + await new Promise(resolve => silent.listen(0, resolve)); + + const address = silent.address(); + const port = typeof address === 'object' && address !== null ? address.port : 0; + const { err, session } = await client({ port, responseTimeout: 10_000 }); + + assert.equal(err, undefined); + assert.ok(session); + + const controller = new AbortController(); + + setTimeout(() => { controller.abort(); }, 50); + + const sent = await session.sendSms( + { from: '46701113311', message: 'never answered', to: '46709771337' }, + { signal: controller.signal }, + ); + + assert.ok(sent.err instanceof Error); + + session.close(); + + for (const sock of accepted) sock.destroy(); + + await new Promise(resolve => silent.close(() => { resolve(); })); + }); +});