From 2821d669431cec2df4cf2810eced4c6c98495a0a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Semy=C3=B3n?= Date: Tue, 6 Oct 2026 12:51:12 +0000 Subject: [PATCH 1/2] Answer enquire_link and drop links whose peer stops answering An incoming enquire_link was only emitted as an event, leaving every user to answer it by hand, and auto_enquire_link_period sent enquire_link without ever looking at the enquire_link_resp, so a dead link stayed open until TCP gave up on it, which can take many minutes. auto_enquire_link_period passed to createServer() was silently ignored. New options, accepted by both connect() and createServer(): - auto_enquire_link_response answers incoming enquire_link, before the 'pdu' and 'enquire_link' events are emitted. - enquire_link_timeout arms a timer for every keepalive enquire_link. When it fires, the session emits enquire_link_timeout with the unanswered PDU and is destroyed, unless close_on_enquire_link_timeout is false. - auto_enquire_link_period now also works on server sessions. The keepalive enquire_link only wait for enquire_link_timeout: response_timeout is for the requests the user sends, so they are tracked without its timer and never show up in 'response_timeout' or 'response_aborted' events. Otherwise a response_timeout shorter than enquire_link_timeout would settle them first and switch the link check off. startEnquireLink(period, [timeout]) / stopEnquireLink() and the autoEnquireLinkResponse property control the same at runtime. Everything defaults to the previous behaviour: answering automatically by default would double the response for everyone who already answers in their own handler. Lifecycle details: - A keepalive registers a response callback only when there is a timeout to clear, and the callback is removed on timeout, on stop and when the write fails, so nothing accumulates against a silent peer. - Any response with the right sequence number counts as an answer, including generic_nack, which some SMSCs send for an enquire_link before the bind. - close() stops sending but keeps awaiting an enquire_link already in flight: a peer that stopped answering doesn't send its FIN either, and that timeout is what drops the half-open session. destroy() stops the keepalive synchronously. Replacing the schedule keeps the enquire_link already sent awaited. - The schedule only starts on a writable socket, so it doesn't start when close() was called from a 'connect' listener, and a negative or non-numeric timeout or period is ignored instead of firing at once. --- .changeset/enquire-link-keepalive.md | 29 ++ README.md | 73 ++++ lib/smpp.ts | 184 +++++++++- lib/types.ts | 2 + test/smpp.js | 525 +++++++++++++++++++++++++++ 5 files changed, 795 insertions(+), 18 deletions(-) create mode 100644 .changeset/enquire-link-keepalive.md diff --git a/.changeset/enquire-link-keepalive.md b/.changeset/enquire-link-keepalive.md new file mode 100644 index 0000000..3899a4f --- /dev/null +++ b/.changeset/enquire-link-keepalive.md @@ -0,0 +1,29 @@ +--- +'@semyonf/smpp': minor +--- + +Add `enquire_link` handling to `Session`: answering incoming `enquire_link`, and dropping the link +when an outgoing one is not answered. + +Until now an incoming `enquire_link` was only emitted as an event, and `auto_enquire_link_period` +sent `enquire_link` without ever checking the `enquire_link_resp`, so a dead link stayed "open" +until TCP noticed, which can take many minutes. New options, accepted by both `smpp.connect()` and +`smpp.createServer()`: + +- `auto_enquire_link_response` (default `false`) answers every incoming `enquire_link`. +- `enquire_link_timeout` makes every `enquire_link` sent by `auto_enquire_link_period` wait that + many milliseconds for its response. If none arrives, an `enquire_link_timeout` event is emitted + with the unanswered `enquire_link`, and the session is destroyed unless + `close_on_enquire_link_timeout` is `false`. +- `auto_enquire_link_period` now also works on server sessions, and stops when `session.close()` or + `session.destroy()` is called instead of firing at a socket that can no longer be written to. + After `close()` the response to an `enquire_link` still in flight is still awaited, so if the + peer stopped answering and never closes its side, the timeout destroys the session instead of + leaving it half-open. + +These `enquire_link` are not subject to `response_timeout`, which is for the requests you send +yourself, and never show up in `'response_timeout'` or `'response_aborted'` events. + +All of them default to the previous behaviour. At runtime they are controlled with +`session.startEnquireLink(period, [timeout])`, `session.stopEnquireLink()` and the +`session.autoEnquireLinkResponse` property. diff --git a/README.md b/README.md index 96b9e21..cd9f23e 100644 --- a/README.md +++ b/README.md @@ -199,6 +199,49 @@ still emitted as a `'pdu'` event and as its command event (e.g. `'submit_sm_resp Without `response_timeout`, nothing changes: unanswered requests are never reported, and a `failureCallback` is only called for requests that could not be written to the socket. +### Keeping the link alive (enquire_link) + +Both sides of an SMPP session are expected to answer `enquire_link` with `enquire_link_resp`, and +to drop a link whose peer stops answering. Both are opt-in, on client sessions (`smpp.connect()` +options) and server sessions (`smpp.createServer()` options, applied to every accepted session): + +| Option | Default | Description | +|---|---|---| +| `auto_enquire_link_response` | `false` | Answer every incoming `enquire_link` with an `enquire_link_resp`. Don't also answer it from your own `enquire_link` handler, or the peer gets two responses. | +| `auto_enquire_link_period` | unset | Send an `enquire_link` every this many ms. Client sessions start one period after the TCP connection is established (so before the TLS handshake and the bind), server sessions one period after accepting the connection. | +| `enquire_link_timeout` | unset | How long, in ms, to wait for the `enquire_link_resp` to each of those `enquire_link`. When it does not arrive in time, an `enquire_link_timeout` event is emitted. Unset or `0` disables the check. | +| `close_on_enquire_link_timeout` | `true` | Destroy the session after emitting `enquire_link_timeout`. Set to `false` to decide on your own in the event handler. | + +``` javascript +var session = smpp.connect({ + url: 'smpp://example.com:2775', + auto_enquire_link_response: true, + auto_enquire_link_period: 30000, + enquire_link_timeout: 10000 +}); +session.on('enquire_link_timeout', function(pdu) { + console.log('no enquire_link_resp for sequence', pdu.sequence_number, ', dropping the link'); +}); +session.on('close', function() { + // open a new session with smpp.connect() here +}); +``` + +The same can be controlled at runtime with `session.startEnquireLink()`, `session.stopEnquireLink()` +and `session.autoEnquireLinkResponse`. + +The `enquire_link` sent by `auto_enquire_link_period` only wait for `enquire_link_timeout`: +`response_timeout` doesn't apply to them, and they never show up in `'response_timeout'` or +`'response_aborted'` events. An `enquire_link` you send yourself is an ordinary request. + +Any response carrying the sequence number of the `enquire_link` counts as an answer, including a +`generic_nack`: some SMSCs reject an `enquire_link` sent before the bind that way, but they still +answered, so the link is alive. + +Responses are only read while the session is not paused, so keep `session.pause()` shorter than +`enquire_link_timeout`, or the link is considered dead. A paused session doesn't answer incoming +`enquire_link` automatically either, so the peer's own timeout applies as well. + ### Proxy protocol [Proxy Protocol v1](https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt) is now @@ -301,6 +344,27 @@ Can be used to postpone incoming pdu events untill calling `session.resume()`. #### session.resume() Resumes the session after a call to `pause()`. +#### session.startEnquireLink(period, [timeout]) +Sends an `enquire_link` every `period` ms. If `timeout` is given (it defaults to the +`enquire_link_timeout` option), every `enquire_link` that gets no response within `timeout` ms +emits `enquire_link_timeout` and, unless `close_on_enquire_link_timeout` is `false`, destroys the +session. Replaces the schedule set by a previous call or by the `auto_enquire_link_period` option; +`enquire_link` already sent are still awaited with their original timeout. +On a client session that is not connected yet, the schedule starts once the connection is +established. + +#### session.stopEnquireLink() +Stops sending `enquire_link` and stops waiting for responses to the ones already sent. +`session.close()` and `session.destroy()` also stop sending them. After `session.close()` the +response to an `enquire_link` still in flight is awaited, so if the peer never closes its side of +the connection, the timeout destroys the session (unless `close_on_enquire_link_timeout` is +`false`). With nothing in flight, `session.close()` waits for the peer like before: call +`session.destroy()` if the `'close'` event doesn't follow in time. + +#### session.autoEnquireLinkResponse +Whether incoming `enquire_link` are answered automatically. Initialized from the +`auto_enquire_link_response` option and can be changed at any time. + #### session.getPeerCertificate([detailed]) Returns the certificate presented by the TLS peer, as returned by node's [`tls.TLSSocket.getPeerCertificate()`](https://nodejs.org/api/tls.html#tlssocketgetpeercertificatedetailed). @@ -349,6 +413,12 @@ Emitted when a pdu is being sent over the session with the pdu as the argument. #### Event: 'pdu' `(pdu)` Emitted upon receiving a pdu. +#### Event: 'enquire_link_timeout' `(pdu)` +Emitted when an `enquire_link` sent by `auto_enquire_link_period` or `session.startEnquireLink()` +got no `enquire_link_resp` within the timeout. `pdu` is that `enquire_link`. Unless +`close_on_enquire_link_timeout` is `false`, the session is destroyed right after, so a `'close'` +event follows. + #### Event: 'unknown' `(pdu)` Emitted upon receiving an unknown pdu. @@ -369,6 +439,9 @@ Creates a new SMPP server. The `sessionListener` argument is automatically set as a listener for the 'session' event. If options include `key` and `cert`, a TLS secured server will be created. Include `rejectUnauthorized: false` to disable the certificate validation. +The `enquire_link` options described in +[Keeping the link alive](#keeping-the-link-alive-enquire_link) apply to every session the server +accepts. ### smpp.Server The base object for a SMPP server created with `smpp.createServer()`. diff --git a/lib/smpp.ts b/lib/smpp.ts index 40e3313..930ff3e 100644 --- a/lib/smpp.ts +++ b/lib/smpp.ts @@ -5,7 +5,7 @@ import { parse } from 'url'; import * as defs from './defs'; import { PDU } from './pdu'; import { EventEmitter } from 'events'; -import { SessionEventMap, AnyPDU } from './types'; +import { SessionEventMap, AnyPDU, EnquireLinkPDU } from './types'; const proxy = require('findhit-proxywrap').proxy; @@ -50,6 +50,15 @@ export interface ConnectOptions { * (default: unset, wait forever). See README, "Response timeout". */ response_timeout?: number; + /** + * How long to wait for the enquire_link_resp to an auto-sent enquire_link (milliseconds). + * When it does not arrive in time, 'enquire_link_timeout' is emitted. Unset or 0 disables the check. + */ + enquire_link_timeout?: number; + /** Destroy the session when an enquire_link_resp does not arrive in time (default: true) */ + close_on_enquire_link_timeout?: boolean; + /** Answer incoming enquire_link with enquire_link_resp automatically (default: false) */ + auto_enquire_link_response?: boolean; /** Alternative: connection URL (smpp://host:port or ssmpp://host:port for TLS) */ url?: string; /** Additional options passed to net.connect or tls.connect */ @@ -82,6 +91,14 @@ export interface ServerOptions { debugListener?: DebugListener; /** response_timeout for every session accepted by this server, see ConnectOptions */ response_timeout?: number; + /** Auto-send enquire_link to every connected client at this interval (milliseconds) */ + auto_enquire_link_period?: number; + /** See ConnectOptions.enquire_link_timeout */ + enquire_link_timeout?: number; + /** See ConnectOptions.close_on_enquire_link_timeout */ + close_on_enquire_link_timeout?: boolean; + /** Answer incoming enquire_link with enquire_link_resp automatically (default: false) */ + auto_enquire_link_response?: boolean; /** Auto-prepend buffer for testing (internal use) */ autoPrependBuffer?: Buffer; /** Additional options passed to net.Server or tls.Server */ @@ -174,11 +191,16 @@ export class Session extends EventEmitter { public proxyProtocolProxy: any = null; /** Whether this session runs over TLS */ public readonly tls: boolean; + /** Whether incoming enquire_link PDUs are answered automatically */ + public autoEnquireLinkResponse: boolean; private _busy: boolean = false; private _callbacks: Record = {}; private _responseTimeout: number; private _lastSocketError: Error | null = null; private _interval: NodeJS.Timeout | 0 = 0; + private _enquireLink: { period: number; timeout: number } | null = null; + // Awaited enquire_link: timeout timer -> the request and the response callback registered for it + private _enquireLinkTimers = new Map void }>(); private _command_length: number | null = null; private _mode: string | null = null; private _id: number = Math.floor(Math.random() * (999999 - 100000)) + 100000; // random session id @@ -296,6 +318,13 @@ export class Session extends EventEmitter { this.tls = options.tls === true; this._responseTimeout = normalizeResponseTimeout(options.response_timeout); + this.autoEnquireLinkResponse = options.auto_enquire_link_response === true; + if (options.auto_enquire_link_period > 0) { + this._enquireLink = { + period: options.auto_enquire_link_period, + timeout: options.enquire_link_timeout > 0 ? options.enquire_link_timeout : 0, + }; + } const self = this; let connectTimeout; @@ -346,11 +375,7 @@ export class Session extends EventEmitter { self.debug('server.connected', 'connected to server', { secure: options.tls }); self.emitMetric('server.connected', 1); self.emit('connect'); // @todo should emit the session, but it would break BC - if (self.options.auto_enquire_link_period) { - self._interval = setInterval(function () { - self.enquire_link(); - }, self.options.auto_enquire_link_period); - } + self._startEnquireLinkInterval(); }.bind(this) ); this.socket.on( @@ -380,27 +405,25 @@ export class Session extends EventEmitter { self.debug('server.disconnected', 'disconnected from server'); self.emitMetric('server.disconnected', 1); } + self._stopEnquireLinkInterval(); // Settled before 'close' is emitted, so that a listener which reconnects // never sees requests of the old connection as pending. self._abortPending(); self.emit('close'); - if (self._interval) { - clearInterval(self._interval); - self._interval = 0; - } }); this.socket.on('error', function (e) { clearTimeout(connectTimeout); - if (self._interval) { - clearInterval(self._interval); - self._interval = 0; - } + self._stopEnquireLinkInterval(); self.debug('socket.error', e.message, e); self.emitMetric('socket.error', 1, { error: e }); // 'close' follows; the error becomes the cause of the requests aborted there. self._lastSocketError = e; self.emit('error', e); // Emitted errors will kill the program if they're not captured. }); + + if (this._mode === 'server') { + this._startEnquireLinkInterval(); + } } emitMetric(event, value, payload?) { @@ -429,6 +452,7 @@ export class Session extends EventEmitter { 'pdu.command.out': '\x1b[32m', 'pdu.command.error': '\x1b[41m\x1b[30m', 'socket.error': '\x1b[41m\x1b[30m', + 'enquire_link.timeout': '\x1b[41m\x1b[30m', 'socket.data.in': '\x1b[2m', 'socket.data.out': '\x1b[2m', metrics: '\x1b[2m', @@ -479,7 +503,8 @@ export class Session extends EventEmitter { private _addPending( pdu: PDU, responseCallback: PDUResponseCallback, - failureCallback?: PDUFailureCallback + failureCallback: PDUFailureCallback | undefined, + responseTimeout: number ): PendingRequest { const sequenceNumber = pdu.sequence_number; const displaced = this._callbacks[sequenceNumber]; @@ -499,8 +524,8 @@ export class Session extends EventEmitter { settled: false, }; this._callbacks[sequenceNumber] = entry; - if (this._responseTimeout > 0) { - entry.timer = setTimeout(() => this._timeoutPending(entry), this._responseTimeout); + if (responseTimeout > 0) { + entry.timer = setTimeout(() => this._timeoutPending(entry), responseTimeout); // The socket keeps the event loop alive while a response can still arrive, and // the timers are cleared when it closes, so a timer has no reason to do it too. entry.timer.unref(); @@ -624,6 +649,9 @@ export class Session extends EventEmitter { return; } this._command_length = null; + if (pdu.command === 'enquire_link' && this.autoEnquireLinkResponse) { + this.send(pdu.response()); + } this.emit('pdu', pdu); this.emit(pdu.command, pdu); if (pdu.isResponse()) { @@ -638,6 +666,16 @@ export class Session extends EventEmitter { responseCallback?: PDUResponseCallback, sendCallback?: PDUSendCallback, failureCallback?: PDUFailureCallback + ): boolean { + return this._send(pdu, responseCallback, sendCallback, failureCallback, this._responseTimeout); + } + + private _send( + pdu: PDU, + responseCallback: PDUResponseCallback | undefined, + sendCallback: PDUSendCallback | undefined, + failureCallback: PDUFailureCallback | undefined, + responseTimeout: number ): boolean { if (!this.socket.writable) { const errorObject = { @@ -664,7 +702,7 @@ export class Session extends EventEmitter { pdu.sequence_number = ++this.sequence; } if (responseCallback) { - pending = this._addPending(pdu, responseCallback, failureCallback); + pending = this._addPending(pdu, responseCallback, failureCallback, responseTimeout); } } else if (responseCallback && !sendCallback) { sendCallback = responseCallback; @@ -705,6 +743,107 @@ export class Session extends EventEmitter { return true; } + /** + * Start sending enquire_link every `period` milliseconds. When `timeout` is given (defaults to the + * enquire_link_timeout option), every enquire_link that is not answered within `timeout` + * milliseconds emits 'enquire_link_timeout' and, unless close_on_enquire_link_timeout is false, + * destroys the session. + * + * Replaces any previous schedule. On a client session that is not connected yet, the first + * enquire_link goes out one period after the TCP connection is established. + */ + startEnquireLink(period: number, timeout: number = this.options.enquire_link_timeout): void { + if (!(period > 0)) { + throw new TypeError('enquire_link period must be a positive number of milliseconds'); + } + this._enquireLink = { period, timeout: timeout > 0 ? timeout : 0 }; + if (!this.socket.destroyed && !this.socket.connecting) { + this._startEnquireLinkInterval(); + } + } + + /** + * Stop sending enquire_link and forget the enquire_link_resp that are still awaited. + */ + stopEnquireLink(): void { + this._enquireLink = null; + this._stopEnquireLinkInterval(); + } + + private _startEnquireLinkInterval(): void { + // Only the schedule is replaced: the enquire_link already sent stay awaited. + this._clearEnquireLinkSchedule(); + // Not writable: close() was already called, e.g. from a 'connect' listener. + if (!this._enquireLink || !this.socket.writable) { + return; + } + const { period, timeout } = this._enquireLink; + this._interval = setInterval(() => this._sendEnquireLink(timeout), period); + } + + private _stopEnquireLinkInterval(): void { + this._clearEnquireLinkSchedule(); + for (const [timer, { pdu, settle }] of this._enquireLinkTimers) { + clearTimeout(timer); + this._forgetEnquireLink(pdu, settle); + } + this._enquireLinkTimers.clear(); + } + + private _clearEnquireLinkSchedule(): void { + if (this._interval) { + clearInterval(this._interval); + this._interval = 0; + } + } + + private _sendEnquireLink(timeout: number): void { + const pdu = new PDU('enquire_link'); + if (!timeout) { + // Nothing waits for the response, so don't leave a callback behind for a peer that never answers. + this.send(pdu); + return; + } + const timer = setTimeout(() => { + this._enquireLinkTimers.delete(timer); + this._forgetEnquireLink(pdu, settle); + if (this.socket.destroyed) { + // The socket was destroyed directly (not through destroy(), which clears the timers) and + // 'close' hasn't run yet: the session is going away anyway. + return; + } + this.debug('enquire_link.timeout', 'no enquire_link_resp within ' + timeout + 'ms', { + sequence_number: pdu.sequence_number, + timeout: timeout, + }); + this.emitMetric('enquire_link.timeout', 1, { pdu: pdu, timeout: timeout }); + this.emit('enquire_link_timeout', pdu as unknown as EnquireLinkPDU); + if (this.options.close_on_enquire_link_timeout !== false) { + this.destroy(); + } + }, timeout); + const settle = (): void => { + clearTimeout(timer); + this._enquireLinkTimers.delete(timer); + }; + // Registered before send(): a socket that isn't writable fails the send synchronously. + this._enquireLinkTimers.set(timer, { pdu, settle }); + // A failed write was never seen by the peer, so there is no response to wait for. The + // keepalive has a timeout of its own, so response_timeout doesn't apply to it. + this._send(pdu, settle, undefined, settle, 0); + } + + /** + * Stop tracking the response to a keepalive enquire_link, if it is still the one awaited under + * its sequence number. + */ + private _forgetEnquireLink(pdu: PDU, settle: () => void): void { + const entry = this._callbacks[pdu.sequence_number]; + if (entry && entry.responseCallback === settle) { + this._settlePending(entry); + } + } + pause() { this.paused = true; } @@ -722,6 +861,10 @@ export class Session extends EventEmitter { this.socket.once('close', callback); } } + // The socket stops being writable, so further enquire_link could only fail. The ones already + // sent stay awaited: a peer that stopped answering will not close its side either, and their + // timeout is what destroys the half-open session then. + this._clearEnquireLinkSchedule(); this.socket.end(); } @@ -733,6 +876,7 @@ export class Session extends EventEmitter { this.socket.once('close', callback); } } + this._stopEnquireLinkInterval(); this.socket.destroy(); } } @@ -825,6 +969,10 @@ function ServerConstructor(this: any, options: ServerOptions | SessionListener, debug: self.options.debug, debugListener: self.options.debugListener || undefined, response_timeout: self.options.response_timeout, + auto_enquire_link_period: self.options.auto_enquire_link_period, + enquire_link_timeout: self.options.enquire_link_timeout, + close_on_enquire_link_timeout: self.options.close_on_enquire_link_timeout, + auto_enquire_link_response: self.options.auto_enquire_link_response, }); session.server = self; if (socket.savedEmit) { diff --git a/lib/types.ts b/lib/types.ts index b5ef13a..63ae607 100644 --- a/lib/types.ts +++ b/lib/types.ts @@ -519,4 +519,6 @@ export type SessionEventMap = { response_timeout: AnyPDU; /** The request pdu that was still waiting for its response when the session closed */ response_aborted: AnyPDU; + /** An auto-sent enquire_link got no enquire_link_resp in time; carries that enquire_link */ + enquire_link_timeout: EnquireLinkPDU; }; diff --git a/test/smpp.js b/test/smpp.js index 3d76508..b239050 100644 --- a/test/smpp.js +++ b/test/smpp.js @@ -1038,3 +1038,528 @@ describe('Client/Server simulations', function() { }); }); }); + +describe('enquire_link', function() { + var server, port, serverSessions; + + function listen(options, done) { + serverSessions = []; + server = smpp.createServer(options, function(session) { + session.on('error', function() {}); // the tests tear clients down abruptly + serverSessions.push(session); + }); + server.listen(0, done); + port = server.address().port; + } + + // Connect a client and call back with the server's side of the session. The client's 'connect' + // can come before the server has accepted the connection (seen on macOS), so wait for both. + function connect(options, callback) { + return smpp.connect(options, function() { + if (serverSessions.length) { + callback(serverSessions[0]); + } else { + server.once('session', callback); + } + }); + } + + afterEach(function(done) { + server.sessions.forEach(function(session) { + session.destroy(); + }); + server.close(done); + }); + + describe('auto_enquire_link_response', function() { + beforeEach(function(done) { + listen({ auto_enquire_link_response: true }, done); + }); + + it('should answer an incoming enquire_link without a handler', function(done) { + var session = smpp.connect({ port: port }, function() { + session.enquire_link(function(pdu) { + assert.equal(pdu.command, 'enquire_link_resp'); + assert.equal(pdu.command_status, smpp.ESME_ROK); + session.destroy(done); + }); + }); + }); + + it('should still emit the enquire_link event', function(done) { + var session = connect({ port: port }, function(serverSession) { + serverSession.on('enquire_link', function(pdu) { + assert.equal(pdu.command, 'enquire_link'); + session.destroy(done); + }); + session.enquire_link(); + }); + }); + + it('should answer before emitting the events', function(done) { + var session = connect({ port: port }, function(serverSession) { + var sent = []; + var send = serverSession.send; + serverSession.send = function(pdu) { + sent.push(pdu.command); + return send.apply(this, arguments); + }; + serverSession.on('pdu', function() { + assert.deepEqual(sent, ['enquire_link_resp']); + }); + serverSession.on('enquire_link', function() { + assert.deepEqual(sent, ['enquire_link_resp']); + session.destroy(done); + }); + session.enquire_link(); + }); + }); + + it('should be switchable at runtime', function(done) { + var session = connect({ port: port }, function(serverSession) { + serverSession.autoEnquireLinkResponse = false; + session.enquire_link(function() { + done(new Error('enquire_link should not have been answered')); + }); + setTimeout(function() { + session.destroy(done); + }, 100); + }); + }); + }); + + describe('enquire_link_timeout', function() { + beforeEach(function(done) { + // Nothing on the server side answers enquire_link. + listen({}, done); + }); + + it('should not answer an incoming enquire_link by default', function(done) { + var session = smpp.connect({ port: port }, function() { + session.enquire_link(function() { + done(new Error('enquire_link should not have been answered')); + }); + setTimeout(function() { + session.destroy(done); + }, 100); + }); + }); + + it('should emit enquire_link_timeout and close the session when no response arrives', function(done) { + var timedOut = null; + var session = smpp.connect({ + port: port, + auto_enquire_link_period: 20, + enquire_link_timeout: 50 + }); + session.on('enquire_link_timeout', function(pdu) { + assert.equal(timedOut, null, 'only one timeout is expected before the session closes'); + timedOut = pdu; + }); + session.on('close', function() { + assert.ok(timedOut, 'enquire_link_timeout was not emitted'); + assert.equal(timedOut.command, 'enquire_link'); + assert.equal(timedOut.sequence_number, 1); + done(); + }); + }); + + it('should keep the session open when close_on_enquire_link_timeout is false', function(done) { + var timeouts = 0; + var session = smpp.connect({ + port: port, + auto_enquire_link_period: 20, + enquire_link_timeout: 30, + close_on_enquire_link_timeout: false + }); + session.on('close', function() { + if (timeouts < 3) done(new Error('session closed on enquire_link timeout')); + }); + session.on('enquire_link_timeout', function(pdu) { + assert.equal(session._callbacks[pdu.sequence_number], undefined, 'the timed out callback was kept'); + if (++timeouts === 3) { + // At most the enquire_link sent after the one that just timed out is still awaited. + assert.ok(session._enquireLinkTimers.size <= 1, 'fired timers were kept: ' + session._enquireLinkTimers.size); + session.destroy(done); + } + }); + }); + + it('should not keep callbacks for enquire_link when no timeout is set', function(done) { + var session = connect({ port: port, auto_enquire_link_period: 10 }, function(serverSession) { + var received = 0; + serverSession.on('enquire_link', function() { + if (++received === 3) { + assert.deepEqual(Object.keys(session._callbacks), []); + session.destroy(done); + } + }); + }); + }); + + it('should take the timeout from the option when startEnquireLink() is not given one', function(done) { + var session = smpp.connect({ port: port, enquire_link_timeout: 30 }, function() { + session.startEnquireLink(10); + }); + session.on('close', done); + }); + + it('should ignore a negative timeout option', function(done) { + var session = connect({ + port: port, + auto_enquire_link_period: 10, + enquire_link_timeout: -1 + }, function(serverSession) { + serverSession.on('enquire_link', function(pdu) { + serverSession.send(pdu.response()); + }); + }); + session.on('enquire_link_timeout', function() { + done(new Error('enquire_link timed out')); + }); + setTimeout(function() { + session.destroy(done); + }, 150); + }); + + it('should start a schedule requested before the connection is established', function(done) { + var connected = false; + var session = connect({ port: port }, function(serverSession) { + serverSession.once('enquire_link', function() { + session.destroy(done); + }); + }); + session.on('connect', function() { + connected = true; + }); + session.on('debug', function(type) { + if (type === 'pdu.command.out') { + assert.ok(connected, 'enquire_link was sent before the connection was established'); + } + }); + session.startEnquireLink(1); + assert.equal(session._interval, 0, 'the schedule started before the connection was established'); + }); + + it('should ignore a negative timeout given to startEnquireLink()', function(done) { + var session = connect({ port: port }, function(serverSession) { + serverSession.on('enquire_link', function(pdu) { + serverSession.send(pdu.response()); + }); + session.startEnquireLink(10, -5); + }); + session.on('enquire_link_timeout', function() { + done(new Error('enquire_link timed out')); + }); + setTimeout(function() { + session.destroy(done); + }, 100); + }); + + it('should not wait for a response to an enquire_link that could not be written', function(done) { + var session = smpp.connect({ port: port }, function() { + session.socket.write = function(buffer, callback) { + process.nextTick(callback, new Error('write failed')); + return false; + }; + session.startEnquireLink(10, 30); + }); + session.on('enquire_link_timeout', function() { + done(new Error('enquire_link_timeout emitted for an enquire_link that was never written')); + }); + setTimeout(function() { + assert.equal(session._enquireLinkTimers.size, 0); + session.destroy(done); + }, 120); + }); + + // A peer that stopped answering and doesn't close its side of the connection either. + function withDeadPeer(test) { + return function(done) { + var net = require('net'); + var sockets = []; + var deadPeer = net.createServer({ allowHalfOpen: true }, function(socket) { + sockets.push(socket); + }); + deadPeer.listen(0, function() { + test(deadPeer.address().port, function(err) { + sockets.forEach(function(socket) { + socket.destroy(); + }); + deadPeer.close(function() { + done(err); + }); + }); + }); + }; + } + + it('should keep waiting for the responses after close() and drop a half-open peer', withDeadPeer(function(deadPort, done) { + var timedOut = false; + var session = smpp.connect({ + port: deadPort, + auto_enquire_link_period: 10, + enquire_link_timeout: 50 + }); + session.once('send', function() { + session.close(); + }); + session.on('enquire_link_timeout', function() { + timedOut = true; + }); + session.on('close', function() { + assert.ok(timedOut, 'the session was not dropped by the enquire_link timeout'); + done(); + }); + })); + + it('should keep waiting for the responses when the schedule is replaced', withDeadPeer(function(deadPort, done) { + var timedOut = false; + var session = smpp.connect({ + port: deadPort, + auto_enquire_link_period: 10, + enquire_link_timeout: 50 + }); + session.once('send', function() { + session.close(); + session.startEnquireLink(10, 50); + assert.equal(session._enquireLinkTimers.size, 1, 'the awaited enquire_link was forgotten'); + }); + session.on('enquire_link_timeout', function() { + timedOut = true; + }); + session.on('close', function() { + assert.ok(timedOut, 'the session was not dropped by the enquire_link timeout'); + done(); + }); + })); + + it('should not wait for a response to an enquire_link that failed to send synchronously', withDeadPeer(function(deadPort, done) { + var session = smpp.connect({ + port: deadPort, + auto_enquire_link_period: 10, + enquire_link_timeout: 20 + }, function() { + setImmediate(function() { + // Unlike close(), this leaves the keepalive running against a socket that is no longer writable. + session.socket.end(); + }); + }); + session.on('enquire_link_timeout', function() { + done(new Error('enquire_link_timeout emitted for an enquire_link that was never sent')); + }); + setTimeout(function() { + assert.equal(session._enquireLinkTimers.size, 0); + session.destroy(); + done(); + }, 100); + })); + + it('should stop the keepalive synchronously on destroy()', function(done) { + var session = smpp.connect({ port: port, auto_enquire_link_period: 10, enquire_link_timeout: 1000 }, function() { + session.once('send', function() { + session.destroy(); + assert.equal(session._interval, 0); + assert.equal(session._enquireLinkTimers.size, 0); + done(); + }); + }); + session.on('enquire_link_timeout', function() { + done(new Error('enquire_link_timeout emitted after destroy()')); + }); + }); + + it('should not start sending enquire_link when closed from the connect listener', function(done) { + var session = smpp.connect({ port: port, auto_enquire_link_period: 10 }, function() { + session.close(); + setImmediate(function() { + assert.equal(session._interval, 0, 'the interval was started after close()'); + done(); + }); + }); + }); + + it('should stop sending enquire_link once the peer closes the session', function(done) { + var session = connect({ port: port, auto_enquire_link_period: 10 }, function(serverSession) { + serverSession.destroy(); + }); + session.on('close', function() { + setTimeout(function() { + assert.equal(session._interval, 0, 'the interval survived the close'); + done(); + }, 30); + }); + }); + + it('should stop sending enquire_link once the session is closed', function(done) { + var session = smpp.connect({ port: port, auto_enquire_link_period: 10 }, function() { + setTimeout(function() { + assert.notEqual(session._interval, 0, 'the interval did not start'); + session.close(); + assert.equal(session._interval, 0, 'close() kept the interval'); + }, 5); + }); + session.on('close', function() { + setTimeout(function() { + assert.equal(session._interval, 0, 'the interval survived the close'); + done(); + }, 30); + }); + }); + + it('should not time out when the peer answers', function(done) { + var responses = 0; + var session = connect({ + port: port, + auto_enquire_link_period: 20, + enquire_link_timeout: 200 + }, function(serverSession) { + serverSession.on('enquire_link', function(pdu) { + serverSession.send(pdu.response()); + }); + }); + session.on('enquire_link_timeout', function() { + done(new Error('enquire_link timed out')); + }); + session.on('enquire_link_resp', function() { + responses++; + }); + setTimeout(function() { + assert.ok(responses >= 3, 'expected at least 3 responses, got ' + responses); + session.destroy(done); + }, 250); + }); + + it('should not apply response_timeout to the keepalive', function(done) { + var responses = 0; + var session = connect({ + port: port, + response_timeout: 20, + auto_enquire_link_period: 10, + enquire_link_timeout: 500 + }, function(serverSession) { + serverSession.on('enquire_link', function(pdu) { + // Slower than response_timeout, well within enquire_link_timeout. + setTimeout(function() { + serverSession.send(pdu.response()); + }, 60); + }); + }); + session.on('response_timeout', function() { + done(new Error('response_timeout was applied to a keepalive enquire_link')); + }); + session.on('enquire_link_timeout', function() { + done(new Error('enquire_link timed out')); + }); + session.on('enquire_link_resp', function() { + if (++responses === 3) { + session.on('response_aborted', function() { + done(new Error('a keepalive enquire_link was reported as aborted')); + }); + session.destroy(done); + } + }); + }); + + it('should be startable and stoppable at runtime', function(done) { + var sent = 0; + var session = connect({ port: port }, function(serverSession) { + serverSession.on('enquire_link', function(pdu) { + sent++; + serverSession.send(pdu.response()); + }); + session.startEnquireLink(20, 50); + setTimeout(function() { + assert.ok(sent >= 2, 'expected enquire_link to be sent, got ' + sent); + session.stopEnquireLink(); + var sentWhenStopped = sent; + setTimeout(function() { + // One enquire_link may have been in flight when stopping. + assert.ok(sent <= sentWhenStopped + 1, 'enquire_link kept being sent after stopEnquireLink()'); + session.destroy(done); + }, 100); + }, 100); + }); + }); + + it('should forget the awaited response when stopped', function(done) { + var session = smpp.connect({ port: port }, function() { + session.startEnquireLink(10, 50); + setTimeout(function() { + session.stopEnquireLink(); + assert.deepEqual(Object.keys(session._callbacks), [], 'stopEnquireLink() kept response callbacks'); + }, 15); + setTimeout(function() { + assert.ok(!session.closed, 'session was closed after stopEnquireLink()'); + session.destroy(done); + }, 150); + }); + session.on('enquire_link_timeout', function() { + done(new Error('enquire_link_timeout emitted after stopEnquireLink()')); + }); + }); + + it('should reject a non-positive period', function(done) { + var session = smpp.connect({ port: port }, function() { + assert.throws(function() { + session.startEnquireLink(0); + }, TypeError); + session.destroy(done); + }); + }); + }); + + describe('server sessions', function() { + beforeEach(function(done) { + listen({ auto_enquire_link_period: 20, enquire_link_timeout: 50 }, done); + }); + + it('should drop a client that does not answer enquire_link', function(done) { + var serverTimedOut = false; + var session = connect({ port: port }, function(serverSession) { + serverSession.on('enquire_link_timeout', function() { + serverTimedOut = true; + }); + }); + session.on('error', function() {}); + session.on('close', function() { + assert.ok(serverTimedOut, 'server session did not report the timeout'); + done(); + }); + }); + + it('should keep a silent client when close_on_enquire_link_timeout is false', function(done) { + server.close(); + listen({ auto_enquire_link_period: 10, enquire_link_timeout: 20, close_on_enquire_link_timeout: false }, function() { + var timeouts = 0; + var session = connect({ port: port }, function(serverSession) { + serverSession.on('enquire_link_timeout', function() { + if (++timeouts === 3) { + session.destroy(done); + } + }); + }); + session.on('error', function() {}); + session.on('close', function() { + if (timeouts < 3) done(new Error('server dropped the client')); + }); + }); + }); + + it('should keep a client that answers enquire_link', function(done) { + var received = 0; + var session = smpp.connect({ port: port, auto_enquire_link_response: true }); + session.on('enquire_link', function() { + received++; + }); + session.on('close', function() { + done(new Error('session was closed')); + }); + setTimeout(function() { + assert.ok(received >= 3, 'expected at least 3 enquire_link, got ' + received); + session.removeAllListeners('close'); + session.destroy(done); + }, 250); + }); + }); +}); From 7ee10170a47626d8fe73ae4327ef955bf5af9454 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Semy=C3=B3n?= Date: Tue, 6 Oct 2026 12:51:12 +0000 Subject: [PATCH 2/2] Give every session.connect() a socket of its own connect() reconnected by calling connect() again on the previous socket. A net.Socket that was closed can connect again, but never emits 'readable' again, so a reconnected session sent PDUs and never read a single one: responses and incoming requests were silently lost. The connection was also only half reset: a partly received PDU kept its command length and corrupted the first PDU of the new connection, the bytesRead baseline stayed at the old socket's count, closed stayed true, and the connect timeout was not re-armed. connect() now opens a new socket (net or tls, as in the constructor) and resets all per-connection state. Socket listeners only act while their socket is the session's socket: reconnecting from an 'error' listener used to be followed by the failed connection's 'close', which made the session look closed again. They stay attached so that a late error of a replaced socket is swallowed instead of crashing the process. A previous connection that is still open or still connecting is destroyed instead of failing with EISCONN / EALREADY, and a server session throws a clear error instead of Node's ERR_MISSING_ARGS. The enquire_link schedule and all listeners carry over to the new connection. The new socket is created before anything else is touched, so options that net.connect() or tls.connect() reject leave the session as it was. Requests waiting for a response fail with ESESSIONCLOSED when response_timeout is set, as on close, and the socket to replace is only picked after their failure callbacks ran, since those may reconnect on their own. A connection replaced from such a callback, which runs while its 'close' is handled, no longer emits that 'close'; and a connection opened from an enquire_link_timeout listener is no longer destroyed by close_on_enquire_link_timeout right after. A write that fails on the replaced socket can't drop the callback of a new request reusing its sequence number, since pending entries are only removed by identity. --- .changeset/session-reconnect.md | 26 ++ README.md | 22 +- lib/smpp.ts | 202 +++++++++------ test/smpp.js | 445 ++++++++++++++++++++++++++++++++ 4 files changed, 616 insertions(+), 79 deletions(-) create mode 100644 .changeset/session-reconnect.md diff --git a/.changeset/session-reconnect.md b/.changeset/session-reconnect.md new file mode 100644 index 0000000..d117132 --- /dev/null +++ b/.changeset/session-reconnect.md @@ -0,0 +1,26 @@ +--- +'@semyonf/smpp': patch +--- + +Fix `session.connect()`, which could not reconnect a client session: after reconnecting, the session +sent PDUs but never received one, so every response and incoming request was silently lost. + +`connect()` reused the socket of the previous connection, and a closed `net.Socket` that connects +again never emits `'readable'`. Each call now opens a new socket, and also: + +- resets everything bound to the previous connection: sequence numbers, a partly received PDU + (which used to corrupt the first PDU of the new connection), the read counters behind the + `socket.data.in` debug and metrics events, and the `closed` state, so `close(callback)` after a + reconnect waits for the new connection to close. Requests still awaiting a response fail with + `ESESSIONCLOSED` when `response_timeout` is set, and are dropped otherwise, as before; +- re-arms `connectTimeout` for every attempt; +- stops emitting events of the previous connection. Reconnecting from an `'error'` listener used to + be followed by that connection's `'close'`, which made the session look closed again; +- destroys the previous connection first if it is still open or still being established, instead + of connecting its socket again, which failed with `EALREADY` while it was still connecting; +- throws a clear error on a server session, instead of Node's `ERR_MISSING_ARGS`, and leaves the + session untouched when `net.connect()` or `tls.connect()` reject the options. + +Listeners, options and the `enquire_link` schedule are kept across the reconnect, and reconnecting +from an `enquire_link_timeout` listener keeps the new connection instead of having +`close_on_enquire_link_timeout` destroy it. diff --git a/README.md b/README.md index cd9f23e..b133f81 100644 --- a/README.md +++ b/README.md @@ -223,7 +223,9 @@ session.on('enquire_link_timeout', function(pdu) { console.log('no enquire_link_resp for sequence', pdu.sequence_number, ', dropping the link'); }); session.on('close', function() { - // open a new session with smpp.connect() here + setTimeout(function() { + session.connect(); // the enquire_link options apply to the new connection too + }, 5000); }); ``` @@ -336,7 +338,20 @@ that don't honor gracefull tear-down. ( Looking at you SMPPSim ) If supplied, the `callback` is called once the session is fully closed. #### session.connect() -Can be used to reconnect a closed connection. +Connects a client session again, typically from a `'close'`, `'error'` or `'enquire_link_timeout'` +listener. The session gets a new connection: sequence numbers start over, and a PDU the previous +connection delivered only partly is discarded. Requests still waiting for a response on the +previous connection fail with `ESESSIONCLOSED` if `response_timeout` is set (see +[Response timeout](#response-timeout)), and are dropped otherwise. Event listeners, options and +the `enquire_link` schedule are kept, and the connect listener passed to `smpp.connect()` runs +again. + +If the previous connection is still open or still being established, it is destroyed first. Either +way none of its events are emitted any more, so a `'close'` that the previous connection emits +after the error that you reconnected on does not reach your listeners. + +Throws when called on a server session, and when `net.connect()` or `tls.connect()` reject the +options, in which case the session is left as it was. #### session.pause() Can be used to postpone incoming pdu events untill calling `session.resume()`. @@ -349,7 +364,8 @@ Sends an `enquire_link` every `period` ms. If `timeout` is given (it defaults to `enquire_link_timeout` option), every `enquire_link` that gets no response within `timeout` ms emits `enquire_link_timeout` and, unless `close_on_enquire_link_timeout` is `false`, destroys the session. Replaces the schedule set by a previous call or by the `auto_enquire_link_period` option; -`enquire_link` already sent are still awaited with their original timeout. +`enquire_link` already sent are still awaited with their original timeout. The schedule is kept +across `session.connect()`. On a client session that is not connected yet, the schedule starts once the connection is established. diff --git a/lib/smpp.ts b/lib/smpp.ts index 930ff3e..bd1f9df 100644 --- a/lib/smpp.ts +++ b/lib/smpp.ts @@ -198,6 +198,7 @@ export class Session extends EventEmitter { private _responseTimeout: number; private _lastSocketError: Error | null = null; private _interval: NodeJS.Timeout | 0 = 0; + private _connectTimeout: NodeJS.Timeout | undefined; private _enquireLink: { period: number; timeout: number } | null = null; // Awaited enquire_link: timeout timer -> the request and the response callback registered for it private _enquireLinkTimers = new Map void }>(); @@ -326,104 +327,126 @@ export class Session extends EventEmitter { }; } - const self = this; - let connectTimeout; - this._extractPDUs = this._extractPDUs.bind(self); + this._extractPDUs = this._extractPDUs.bind(this); if (options.socket) { // server mode / socket is already connected. this._mode = 'server'; this.socket = options.socket; - this.remoteAddress = self.rootSocket().remoteAddress || self.remoteAddress; + this.remoteAddress = this.rootSocket().remoteAddress || this.remoteAddress; this.remotePort = this.rootSocket().remotePort; this.proxyProtocolProxy = this.rootSocket().proxyAddress ? { address: this.rootSocket().proxyAddress, port: this.rootSocket().proxyPort } : false; + this._attachSocket(this.socket); + this._startEnquireLinkInterval(); } else { // client mode this._mode = 'client'; - if (options.hasOwnProperty('connectTimeout') && options.connectTimeout > 0) { - connectTimeout = setTimeout(function () { - if (self.socket) { - const e = new Error( - 'Timeout of ' + - options.connectTimeout + - 'ms while connecting to ' + - self.options.host + - ':' + - self.options.port - ); - e['code'] = 'ETIMEOUT'; - e['timeout'] = options.connectTimeout; - self.socket.destroy(e); - } - }, options.connectTimeout); - } + this._useClientSocket(this._createClientSocket()); + } + } - if (options.tls) { - this.socket = tls.connect(this.options as tls.ConnectionOptions); - } else { - this.socket = net.connect(this.options as net.NetConnectOpts); - } + /** + * Every connection gets a socket of its own: a net.Socket that was closed can connect again, but + * never emits 'readable' again. + */ + private _createClientSocket(): net.Socket | tls.TLSSocket { + if (this.options.tls) { + return tls.connect(this.options as tls.ConnectionOptions); + } + return net.connect(this.options as net.NetConnectOpts); + } - this.socket.on( - 'connect', - function () { - clearTimeout(connectTimeout); - self.remoteAddress = self.rootSocket().remoteAddress || self.remoteAddress; - self.remotePort = self.rootSocket().remotePort || self.remoteAddress; - self.debug('server.connected', 'connected to server', { secure: options.tls }); - self.emitMetric('server.connected', 1); - self.emit('connect'); // @todo should emit the session, but it would break BC - self._startEnquireLinkInterval(); - }.bind(this) - ); - this.socket.on( - 'secureConnect', - function () { - self.emit('secureConnect'); // @todo should emit the session, but it would break BC - }.bind(this) - ); + /** + * Make the socket the session's socket, with the connect timeout and the listeners of a client. + */ + private _useClientSocket(socket): void { + const options = this.options; + this.socket = socket; + + if (options.hasOwnProperty('connectTimeout') && options.connectTimeout > 0) { + this._connectTimeout = setTimeout(() => { + if (socket !== this.socket) return; + const e = new Error( + 'Timeout of ' + + options.connectTimeout + + 'ms while connecting to ' + + options.host + + ':' + + options.port + ); + e['code'] = 'ETIMEOUT'; + e['timeout'] = options.connectTimeout; + socket.destroy(e); + }, options.connectTimeout); } - this.socket.on('readable', function () { - const bytesRead = self.socket.bytesRead - self._prevBytesRead; + + socket.on('connect', () => { + if (socket !== this.socket) return; + clearTimeout(this._connectTimeout); + this.remoteAddress = this.rootSocket().remoteAddress || this.remoteAddress; + this.remotePort = this.rootSocket().remotePort || this.remoteAddress; + this.debug('server.connected', 'connected to server', { secure: options.tls }); + this.emitMetric('server.connected', 1); + this.emit('connect'); // @todo should emit the session, but it would break BC + this._startEnquireLinkInterval(); + }); + socket.on('secureConnect', () => { + if (socket !== this.socket) return; + this.emit('secureConnect'); // @todo should emit the session, but it would break BC + }); + this._attachSocket(socket); + } + + /** + * Listen to the socket for as long as it is the session's socket. Once connect() has replaced it, + * its events are ignored, which keeps a late 'close' or 'error' of the previous connection from + * being taken for the new one. The listeners stay attached so that such an error is swallowed + * instead of crashing the process. + */ + private _attachSocket(socket): void { + socket.on('readable', () => { + if (socket !== this.socket) return; + const bytesRead = socket.bytesRead - this._prevBytesRead; if (bytesRead > 0) { // on disconnections the readable event receives 0 bytes, we do not want to debug that - self.debug('socket.data.in', null, { bytes: bytesRead }); - self.emitMetric('socket.data.in', bytesRead, { bytes: bytesRead }); - self._prevBytesRead = self.socket.bytesRead; + this.debug('socket.data.in', null, { bytes: bytesRead }); + this.emitMetric('socket.data.in', bytesRead, { bytes: bytesRead }); + this._prevBytesRead = socket.bytesRead; } - self._extractPDUs(); + this._extractPDUs(); }); - this.socket.on('close', function () { - self.closed = true; - clearTimeout(connectTimeout); - if (self._mode === 'server') { - self.debug('client.disconnected', 'client has disconnected'); - self.emitMetric('client.disconnected', 1); + socket.on('close', () => { + if (socket !== this.socket) return; + this.closed = true; + clearTimeout(this._connectTimeout); + if (this._mode === 'server') { + this.debug('client.disconnected', 'client has disconnected'); + this.emitMetric('client.disconnected', 1); } else { - self.debug('server.disconnected', 'disconnected from server'); - self.emitMetric('server.disconnected', 1); + this.debug('server.disconnected', 'disconnected from server'); + this.emitMetric('server.disconnected', 1); } - self._stopEnquireLinkInterval(); + this._stopEnquireLinkInterval(); // Settled before 'close' is emitted, so that a listener which reconnects // never sees requests of the old connection as pending. - self._abortPending(); - self.emit('close'); + this._abortPending(); + // A failureCallback that reconnected has replaced this connection: its 'close' is not + // emitted, like any other event of a replaced connection. + if (socket !== this.socket) return; + this.emit('close'); }); - this.socket.on('error', function (e) { - clearTimeout(connectTimeout); - self._stopEnquireLinkInterval(); - self.debug('socket.error', e.message, e); - self.emitMetric('socket.error', 1, { error: e }); + socket.on('error', (e) => { + if (socket !== this.socket) return; + clearTimeout(this._connectTimeout); + this._stopEnquireLinkInterval(); + this.debug('socket.error', e.message, e); + this.emitMetric('socket.error', 1, { error: e }); // 'close' follows; the error becomes the cause of the requests aborted there. - self._lastSocketError = e; - self.emit('error', e); // Emitted errors will kill the program if they're not captured. + this._lastSocketError = e; + this.emit('error', e); // Emitted errors will kill the program if they're not captured. }); - - if (this._mode === 'server') { - this._startEnquireLinkInterval(); - } } emitMetric(event, value, payload?) { @@ -483,12 +506,37 @@ export class Session extends EventEmitter { this.emit('debug', type, msg, payload); } - connect() { + /** + * Connect a client session again, typically after it was closed. A connection that is still + * open or still being established is destroyed first, and from then on none of its events are + * emitted any more: no 'close' or 'error' of the previous connection follows. + * + * Everything bound to the previous connection is reset: sequence numbers start over, requests + * still waiting for a response fail with ESESSIONCLOSED if they have a response_timeout and are + * dropped otherwise, and a partly received PDU is discarded. The enquire_link schedule and the + * event listeners are kept. + */ + connect(): void { + if (this._mode !== 'client') { + throw new Error('connect() can only be called on a client session'); + } + // First, so that options net.connect() or tls.connect() reject leave the session untouched. + const socket = this._createClientSocket(); + clearTimeout(this._connectTimeout); + this._stopEnquireLinkInterval(); + // Can call failure callbacks, which may reconnect on their own: the socket to replace is only + // picked after that. + this._resetPending(); + const previous = this.socket; this.sequence = 0; this.paused = false; + this.closed = false; this._busy = false; - this._resetPending(); - this.socket.connect(this.options); + this._command_length = null; + this._prevBytesRead = 0; + this._useClientSocket(socket); + // Its events are ignored from here on, as it is no longer this.socket. + previous.destroy(); } // ========== Pending requests ========== @@ -817,8 +865,10 @@ export class Session extends EventEmitter { timeout: timeout, }); this.emitMetric('enquire_link.timeout', 1, { pdu: pdu, timeout: timeout }); + const socket = this.socket; this.emit('enquire_link_timeout', pdu as unknown as EnquireLinkPDU); - if (this.options.close_on_enquire_link_timeout !== false) { + // Unless a listener has already replaced the connection with connect(). + if (this.options.close_on_enquire_link_timeout !== false && socket === this.socket) { this.destroy(); } }, timeout); diff --git a/test/smpp.js b/test/smpp.js index b239050..917e562 100644 --- a/test/smpp.js +++ b/test/smpp.js @@ -1563,3 +1563,448 @@ describe('enquire_link', function() { }); }); }); + +describe('Session#connect() reconnecting', function() { + var net = require('net'); + var server, port, connections; + + beforeEach(function(done) { + connections = 0; + server = smpp.createServer({}, function(session) { + connections++; + session.on('error', function() {}); // the tests tear connections down abruptly + session.on('enquire_link', function(pdu) { + session.send(pdu.response()); + }); + }); + server.listen(0, done); + port = server.address().port; + }); + + afterEach(function(done) { + server.sessions.forEach(function(session) { + session.destroy(); + }); + server.close(done); + }); + + // The server's side of the first connection. The client's 'connect' can come before the server has + // accepted the connection (seen on macOS), so it may have to be waited for. + function firstServerSession(callback) { + if (server.sessions.length) { + callback(server.sessions[0]); + } else { + server.once('session', callback); + } + } + + // Drop the first connection from the server side, once it is up on both sides. + function dropFirstConnection() { + server.once('session', function(serverSession) { + setImmediate(function() { + serverSession.destroy(); + }); + }); + } + + it('should receive PDUs again after reconnecting from the close event', function(done) { + var session = smpp.connect({ port: port }); + session.once('connect', function() { + // Some traffic on the first connection, so that its counters are not at zero. + session.enquire_link(function() { + firstServerSession(function(serverSession) { + serverSession.destroy(); + }); + }); + }); + var closes = 0; + var bytesIn = 0; + session.on('debug', function(type, msg, payload) { + if (type === 'socket.data.in') bytesIn += payload.bytes; + }); + session.on('close', function() { + if (++closes > 1) return; + bytesIn = 0; + assert.equal(session.closed, true); + session.connect(); + assert.equal(session.closed, false); + session.once('connect', function() { + session.enquire_link(function(pdu) { + assert.equal(pdu.command, 'enquire_link_resp'); + assert.equal(pdu.sequence_number, 1, 'sequence numbers did not start over'); + assert.equal(bytesIn, 16, 'socket.data.in did not count the bytes of the new connection'); + assert.equal(connections, 2); + session.close(function() { + assert.equal(closes, 2); + done(); + }); + }); + }); + }); + }); + + it('should reconnect from the error event and ignore the events of the failed connection', function(done) { + // Find a port nobody listens on, fail to connect to it, then start listening there. + var probe = net.createServer(); + probe.listen(0, function() { + var freePort = probe.address().port; + probe.close(function() { + var session = smpp.connect({ port: freePort }); + var errors = 0, closes = 0, other; + session.on('close', function() { + closes++; + }); + session.on('error', function(e) { + errors++; + assert.equal(e.code, 'ECONNREFUSED'); + other = smpp.createServer({}, function(serverSession) { + serverSession.on('error', function() {}); + serverSession.on('enquire_link', function(pdu) { + serverSession.send(pdu.response()); + }); + }); + other.listen(freePort, function() { + session.connect(); + session.once('connect', function() { + // The failed connection's 'close' has had its turn by now. + setTimeout(function() { + assert.equal(errors, 1); + assert.equal(closes, 0, "the failed connection's 'close' was emitted"); + assert.equal(session.closed, false); + session.enquire_link(function() { + session.close(function() { + assert.equal(closes, 1); + other.close(done); + }); + }); + }, 20); + }); + }); + }); + }); + }); + }); + + it('should replace a connection that is still open, without its close or error events', function(done) { + var session = smpp.connect({ port: port }); + session.once('connect', function() { + firstServerSession(function(first) { + session.on('close', function() { + done(new Error("the replaced connection's 'close' was emitted")); + }); + session.on('error', function(e) { + done(e); + }); + session.connect(); + first.on('close', function() { + session.enquire_link(function() { + assert.equal(connections, 2); + session.removeAllListeners('close'); + session.destroy(done); + }); + }); + }); + }); + }); + + it('should ignore an error of the replaced connection', function(done) { + var session = smpp.connect({ port: port }); + session.once('connect', function() { + var previous = session.socket; + var connects = 0; + session.connect(); + session.on('connect', function() { + connects++; + }); + session.on('secureConnect', function() { + done(new Error("the replaced connection's secureConnect was emitted")); + }); + session.on('close', function() { + done(new Error("the replaced connection's close was emitted")); + }); + session.on('error', function() { + done(new Error("the replaced connection's error was emitted")); + }); + previous.emit('error', new Error('late error of the previous connection')); + previous.emit('connect'); + previous.emit('secureConnect'); + previous.emit('readable'); + previous.emit('close'); + assert.equal(connects, 0); + assert.equal(session.closed, false); + session.enquire_link(function() { + assert.equal(connects, 1); + session.removeAllListeners('close'); + session.destroy(done); + }); + }); + }); + + it('should forget the enquire_link awaited on the replaced connection', function(done) { + var session = smpp.connect({ port: port, enquire_link_timeout: 1000 }); + session.once('connect', function() { + // The server answers, but not before the connection is replaced. + session.startEnquireLink(1); + session.once('send', function() { + assert.equal(session._enquireLinkTimers.size, 1); + session.connect(); + assert.equal(session._enquireLinkTimers.size, 0); + assert.equal(session._interval, 0); + session.destroy(done); + }); + }); + }); + + it('should not let a write failing on the replaced connection drop a callback of the new one', function(done) { + var session = smpp.connect({ port: port }); + var failures = 0; + // Queued while connecting; it fails when connect() destroys that socket. + session.enquire_link({}, function() { + done(new Error('the replaced connection delivered a response')); + }, null, function() { + failures++; + }); + session.connect(); + // The same sequence number, 1, on the new connection. + session.enquire_link(function(pdu) { + assert.equal(pdu.sequence_number, 1); + setTimeout(function() { + assert.equal(failures, 1, 'the write on the replaced connection did not fail'); + session.destroy(done); + }, 20); + }); + }); + + it('should leave the session untouched when the options are rejected', function(done) { + var session = smpp.connect({ port: port }); + session.once('connect', function() { + session.options.port = 99999; + assert.throws(function() { + session.connect(); + }, /port/i); + session.options.port = port; + assert.equal(session.closed, false); + session.enquire_link(function() { + assert.equal(connections, 1); + session.destroy(done); + }); + }); + }); + + it('should replace a connection that is still being established', function(done) { + var session = smpp.connect({ port: port }); + session.on('error', done); + session.connect(); + session.on('connect', function() { + session.enquire_link(function() { + session.destroy(done); + }); + }); + }); + + it('should discard a PDU the previous connection delivered only partly', function(done) { + var enquireLink = new smpp.PDU('enquire_link', { sequence_number: 42 }).toBuffer(); + var sockets = []; + var raw = net.createServer(function(socket) { + sockets.push(socket); + socket.on('error', function() {}); + if (sockets.length === 1) { + // The length and part of the header; the rest never comes. + socket.write(enquireLink.slice(0, 10)); + } else { + socket.write(enquireLink); + } + }); + raw.listen(0, function() { + var session = smpp.connect({ port: raw.address().port }); + session.on('error', done); + session.on('debug', function(type) { + if (type === 'socket.data.in' && sockets.length === 1) { + setImmediate(function() { + assert.ok(session._command_length, 'the partial PDU was not pending'); + session.connect(); + }); + } + }); + session.on('enquire_link', function(pdu) { + assert.equal(pdu.sequence_number, 42); + session.destroy(function() { + sockets.forEach(function(socket) { + socket.destroy(); + }); + raw.close(function() { + done(); + }); + }); + }); + }); + }); + + it('should reconnect a secure session', function(done) { + var secureServer = smpp.createServer({ + key: fs.readFileSync(__dirname + '/fixtures/server.key'), + cert: fs.readFileSync(__dirname + '/fixtures/server.crt') + }, function(serverSession) { + serverSession.on('error', function() {}); + serverSession.on('enquire_link', function(pdu) { + serverSession.send(pdu.response()); + }); + }); + secureServer.listen(0, function() { + var secureConnects = 0; + secureServer.once('session', function(serverSession) { + setImmediate(function() { + serverSession.destroy(); + }); + }); + var session = smpp.connect({ port: secureServer.address().port, tls: true }); + session.on('secureConnect', function() { + if (++secureConnects === 1) { + return; + } + session.enquire_link(function(pdu) { + assert.equal(pdu.command, 'enquire_link_resp'); + session.destroy(function() { + secureServer.close(done); + }); + }); + }); + session.once('close', function() { + session.connect(); + }); + }); + }); + + it('should restart the enquire_link schedule on the new connection', function(done) { + dropFirstConnection(); + var session = smpp.connect({ port: port, auto_enquire_link_period: 10, enquire_link_timeout: 200 }); + session.on('enquire_link_timeout', function() { + done(new Error('enquire_link timed out')); + }); + session.once('close', function() { + assert.equal(session._interval, 0); + session.connect(); + var responses = 0; + session.on('enquire_link_resp', function() { + if (++responses === 3) { + session.destroy(done); + } + }); + }); + }); + + it('should arm the connect timeout again', function(done) { + dropFirstConnection(); + var session = smpp.connect({ port: port }); + session.once('close', function() { + // A host name that never resolves, so only the connect timeout ends the attempt. + session.options.host = 'example.invalid'; + session.options.lookup = function() {}; + session.options.connectTimeout = 25; + session.connect(); + session.on('error', function(e) { + assert.equal(e.code, 'ETIMEOUT'); + done(); + }); + }); + }); + + it('should fail requests with a response_timeout that the replaced connection was waiting on', function(done) { + var session = smpp.connect({ port: port, response_timeout: 1000 }); + session.once('connect', function() { + // Nothing answers a submit_sm on this server. + var failures = []; + session.on('close', function() { + done(new Error("the replaced connection's 'close' was emitted")); + }); + session.submit_sm({ destination_addr: '123' }, function() { + done(new Error('unexpected response')); + }, null, function(pdu, err) { + failures.push(err.code); + }); + session.connect(); + assert.deepEqual(failures, ['ESESSIONCLOSED'], 'not failed before connect() returned'); + session.once('connect', function() { + setTimeout(function() { + assert.deepEqual(failures, ['ESESSIONCLOSED']); + session.removeAllListeners('close'); + session.destroy(done); + }, 20); + }); + }); + }); + + it('should not emit close for a connection a failure callback has already replaced', function(done) { + var session = smpp.connect({ port: port, response_timeout: 1000 }); + var failures = 0; + session.on('error', function() {}); // the server resets the first connection + session.on('close', function() { + done(new Error("'close' was emitted after the failure callback reconnected")); + }); + session.once('connect', function() { + session.submit_sm({ destination_addr: '123' }, function() { + done(new Error('unexpected response')); + }, function() { + firstServerSession(function(serverSession) { + serverSession.destroy(); + }); + }, function(pdu, err) { + failures++; + assert.equal(err.code, 'ESESSIONCLOSED'); + session.connect(); + session.once('connect', function() { + session.enquire_link(function() { + assert.equal(failures, 1); + session.removeAllListeners('close'); + session.destroy(done); + }); + }); + }); + }); + }); + + it('should keep a connection opened from the enquire_link_timeout listener', function(done) { + var sockets = []; + var deadPeer = net.createServer(function(socket) { + sockets.push(socket); + }); + deadPeer.listen(0, function() { + var connects = 0; + var session = smpp.connect({ + port: deadPeer.address().port, + auto_enquire_link_period: 10, + enquire_link_timeout: 30 + }); + session.on('connect', function() { + connects++; + }); + session.on('close', function() { + done(new Error('the new connection was closed')); + }); + session.once('enquire_link_timeout', function() { + session.options.port = port; + session.connect(); + session.once('enquire_link_resp', function() { + assert.equal(connects, 2); + session.removeAllListeners('close'); + sockets.forEach(function(socket) { + socket.destroy(); + }); + session.destroy(function() { + deadPeer.close(done); + }); + }); + }); + }); + }); + + it('should refuse to connect a server session', function(done) { + var session = smpp.connect({ port: port }); + server.once('session', function(serverSession) { + assert.throws(function() { + serverSession.connect(); + }, /client session/); + session.destroy(done); + }); + }); +});