dave/smsgw-tester
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor smppsession

Browse Source
This commit is contained in:
David Majdandžić
2023-03-29 18:55:31 +02:00
parent 330051382e
commit ad63061209
7 changed files with 96 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
src/Center/Center.ts
View File

@@ -21,14 +21,13 @@ export class Center extends SmppSession {
port: number; port: number;
pduProcessors: PduProcessor[] = []; pduProcessors: PduProcessor[] = [];
defaultSingleJob: Job = Job.createEmptySingle(); defaultSingleJob!: Job;
defaultMultipleJob: Job = Job.createEmptyMultiple(); defaultMultipleJob!: Job;
readonly logger: Logger;
private pendingSessions: any[] = []; private pendingSessions: any[] = [];
private sessions: any[] = []; private sessions: any[] = [];
private nextSession: number = 0; private nextSession: number = 0;
private server: any; private server: any;
readonly logger: Logger;
constructor(id: number, port: number, username: string, password: string) { constructor(id: number, port: number, username: string, password: string) {
super(); super();
@@ -37,6 +36,9 @@ export class Center extends SmppSession {
this.password = password; this.password = password;
this.port = port; this.port = port;
this.setDefaultSingleJob(Job.createEmptySingle());
this.setDefaultMultipleJob(Job.createEmptyMultiple());
this.logger = new Logger(`Center-${id}`); this.logger = new Logger(`Center-${id}`);
this.initialize(); this.initialize();
@@ -113,6 +115,10 @@ export class Center extends SmppSession {
}; };
} }
getPort(): number {
return this.port;
}
private validateSessions(reject: (reason?: any) => void) { private validateSessions(reject: (reason?: any) => void) {
if (this.sessions.length === 0) { if (this.sessions.length === 0) {
reject(`No clients connected`); reject(`No clients connected`);

266
src/Client/Client.ts
View File

@@ -1,10 +1,8 @@
import EventEmitter from "events";
import {Job} from "../Job/Job"; import {Job} from "../Job/Job";
import {JobEvents} from "../Job/JobEvents";
import Logger from "../Logger"; import Logger from "../Logger";
import {PduProcessor} from "../PDUProcessor/PduProcessor";
import PersistentPromise from "../PersistentPromise"; import PersistentPromise from "../PersistentPromise";
import {SmppSession} from "../SmppSession"; import {SmppSession} from "../SmppSession";
import ClientStatus from "./ClientStatus";
const NanoTimer = require('nanotimer'); const NanoTimer = require('nanotimer');
const smpp = require("smpp"); const smpp = require("smpp");
@@ -12,140 +10,60 @@ const smpp = require("smpp");
const AUTO_ENQUIRE_LINK_PERIOD: number = Number(process.env.AUTO_ENQUIRE_LINK_PERIOD) || 30000; const AUTO_ENQUIRE_LINK_PERIOD: number = Number(process.env.AUTO_ENQUIRE_LINK_PERIOD) || 30000;
const MESSAGE_SEND_UPDATE_DELAY: number = Number(process.env.MESSAGE_SEND_UPDATE_DELAY) || 500; const MESSAGE_SEND_UPDATE_DELAY: number = Number(process.env.MESSAGE_SEND_UPDATE_DELAY) || 500;
export class Client implements SmppSession { export class Client extends SmppSession {
static EVENTS: any = { readonly STATUS: string[] = [
STATUS_CHANGED: "STATUS_CHANGED", "NOT CONNECTED",
STATE_CHANGED: "STATE_CHANGED", "CONNECTING",
ANY_PDU: "ANY_PDU", "CONNECTED",
MESSAGE_SEND_COUNTER_UPDATE_EVENT: "MESSAGE_SEND_COUNTER_UPDATE_EVENT", "BINDING",
} "BOUND",
"BUSY",
]
id: number;
username: string;
password: string;
status: string = this.STATUS[0];
url: string;
pduProcessors: PduProcessor[] = [];
defaultSingleJob!: Job; defaultSingleJob!: Job;
defaultMultipleJob!: Job; defaultMultipleJob!: Job;
UPDATE_WS: string = "UPDATE_WS"; readonly logger: Logger;
private readonly eventEmitter: EventEmitter = new EventEmitter();
private readonly logger: Logger;
private readonly _id: number;
private session?: any; private session?: any;
private connectPromise: PersistentPromise | null = null; private connectPromise: PersistentPromise | null = null;
// TODO: Implement close promise
private bindPromise: PersistentPromise | null = null; private bindPromise: PersistentPromise | null = null;
private closePromise: PersistentPromise | null = null;
// TODO: Implement close promise
// Apparently the sessions are not closed on a dime but instead a .close() call causes eventSessionClose // Apparently the sessions are not closed on a dime but instead a .close() call causes eventSessionClose
private sendTimer: any | null = null;
private counterUpdateTimer: any | null = null;
constructor(id: number, url: string, username: string, password: string) { constructor(id: number, url: string, username: string, password: string) {
this._id = id; super();
this._url = url; this.id = id;
this._username = username; this.username = username;
this._password = password; this.password = password;
this.url = url;
this.setDefaultSingleJob(Job.createEmptySingle());
this.setDefaultMultipleJob(Job.createEmptyMultiple());
this.logger = new Logger(`Client-${id}`); this.logger = new Logger(`Client-${id}`);
this.status = ClientStatus.NOT_CONNECTED;
this.eventEmitter.on(Client.EVENTS.STATE_CHANGED, () => this.updateWs(Client.EVENTS.STATE_CHANGED));
this.eventEmitter.on(Client.EVENTS.STATUS_CHANGED, () => this.updateWs(Client.EVENTS.STATUS_CHANGED));
this.eventEmitter.on(Client.EVENTS.ANY_PDU, (pdu: any) => this.updateWs(Client.EVENTS.ANY_PDU, [pdu]));
this.eventEmitter.on(Client.EVENTS.MESSAGE_SEND_COUNTER_UPDATE_EVENT, (count: number) => this.updateWs(Client.EVENTS.MESSAGE_SEND_COUNTER_UPDATE_EVENT, [count]));
this.initialize();
}
private _username: string;
set username(value: string) {
this._username = value;
}
private _password: string;
set password(value: string) {
this._password = value;
}
private _url: string;
set url(value: string) {
this._url = value;
}
private _status: ClientStatus = ClientStatus.NOT_CONNECTED;
set status(value: ClientStatus) {
this._status = value;
this.eventEmitter.emit(Client.EVENTS.STATUS_CHANGED, this._status);
this.eventEmitter.emit(Client.EVENTS.STATE_CHANGED, this.serialize());
}
updateWs(event: string, args?: any[]): void {
this.logger.log1(`Update WS: ${event}`);
let message: {
type: string,
data?: string
} = {
type: event,
};
switch (event) {
case Client.EVENTS.STATE_CHANGED:
message.data = JSON.stringify(this.serialize());
break;
case Client.EVENTS.STATUS_CHANGED:
message.data = JSON.stringify(this._status);
break;
case Client.EVENTS.ANY_PDU:
message.data = JSON.stringify(args![0]);
break;
case Client.EVENTS.MESSAGE_SEND_COUNTER_UPDATE_EVENT:
message.data = JSON.stringify(args![0]);
break;
}
this.eventEmitter.emit(this.UPDATE_WS, message);
}
getUrl(): string {
return this._url;
}
setDefaultSingleJob(job: Job): void {
this.defaultSingleJob = job;
this.eventEmitter.emit(Client.EVENTS.STATE_CHANGED, this.serialize());
job.on(JobEvents.STATE_CHANGED, () => this.eventEmitter.emit(Client.EVENTS.STATE_CHANGED, this.serialize()));
}
setDefaultMultipleJob(job: Job): void {
this.defaultMultipleJob = job;
this.eventEmitter.emit(Client.EVENTS.STATE_CHANGED, this.serialize());
job.on(JobEvents.STATE_CHANGED, () => this.eventEmitter.emit(Client.EVENTS.STATE_CHANGED, this.serialize()));
}
getDefaultSingleJob(): Job {
return this.defaultSingleJob;
}
getDefaultMultipleJob(): Job {
return this.defaultMultipleJob;
}
initialize(): void {
this.defaultSingleJob = Job.createEmptySingle();
this.defaultMultipleJob = Job.createEmptyMultiple();
this.defaultSingleJob.on(JobEvents.STATE_CHANGED, () => this.eventEmitter.emit(Client.EVENTS.STATE_CHANGED, this.serialize()));
this.defaultMultipleJob.on(JobEvents.STATE_CHANGED, () => this.eventEmitter.emit(Client.EVENTS.STATE_CHANGED, this.serialize()));
} }
doConnect(): PersistentPromise { doConnect(): PersistentPromise {
this.connectPromise = new PersistentPromise((resolve, reject) => { this.connectPromise = new PersistentPromise((resolve, reject) => {
if (this._status !== ClientStatus.NOT_CONNECTED) { if (this.status !== this.STATUS[0]) {
let errorString = `Client-${this._id} already connected`; let errorString = `Client-${this.getId()} already connected`;
this.logger.log1(errorString); this.logger.log1(errorString);
reject(errorString); reject(errorString);
return; return;
} }
this.logger.log1(`Client-${this._id} connecting to ${this._url}`); this.logger.log1(`Client-${this.getId()} connecting to ${this.url}`);
this.status = ClientStatus.CONNECTING; this.setStatus(1);
this.connectSession().then(resolve, ((err: any) => { this.connectSession().then(resolve, ((err: any) => {
this.logger.log1(`Client-${this._id} connection failed: ${err}`); this.logger.log1(`Client-${this.getId()} connection failed: ${err}`);
this.status = ClientStatus.NOT_CONNECTED; this.setStatus(0);
this.session.close(); this.session.close();
reject(err); reject(err);
})); }));
@@ -158,31 +76,37 @@ export class Client implements SmppSession {
this.validateFields(reject); this.validateFields(reject);
this.session.bind_transceiver({ this.session.bind_transceiver({
system_id: this._username, password: this._password, system_id: this.username,
password: this.password,
}, this.eventBindReply.bind(this)); }, this.eventBindReply.bind(this));
this.status = ClientStatus.BINDING; this.setStatus(3);
}); });
return this.bindPromise; return this.bindPromise;
} }
connectAndBind(): Promise<void> { connectAndBind(): Promise<void> {
return this.doConnect().then(this.doBind.bind(this), (error) => { return this.doConnect().then(this.doBind.bind(this), (error) => {
this.logger.log1(`Client-${this._id} connectAndBind failed: ${error}`); this.logger.log1(`Client-${this.getId()} connectAndBind failed: ${error}`);
}); });
} }
serialize(): object { serialize(): object {
return { return {
id: this._id, url: this._url, username: this._username, password: this._password, status: this._status, id: this.getId(),
defaultSingleJob: this.defaultSingleJob, defaultMultipleJob: this.defaultMultipleJob, url: this.url,
username: this.username,
password: this.password,
status: this.status,
defaultSingleJob: this.defaultSingleJob,
defaultMultipleJob: this.defaultMultipleJob,
}; };
} }
close(): Promise<void> { close(): Promise<void> {
return new Promise((resolve, reject) => { return new Promise((resolve, reject) => {
this.logger.log1(`Client-${this._id} closing connection`); this.logger.log1(`Client-${this.getId()} closing connection`);
this.session.close(); this.session.close();
this.status = ClientStatus.NOT_CONNECTED; this.setStatus(0);
resolve(); resolve();
}); });
} }
@@ -193,7 +117,7 @@ export class Client implements SmppSession {
this.validateSession(reject); this.validateSession(reject);
this.validateBound(reject); this.validateBound(reject);
} }
this.logger.log5(`Client-${this._id} sending PDU: ${JSON.stringify(pdu)}`); this.logger.log5(`Client-${this.getId()} sending PDU: ${JSON.stringify(pdu)}`);
this.session.send(pdu, (replyPdu: object) => resolve(replyPdu)); this.session.send(pdu, (replyPdu: object) => resolve(replyPdu));
}); });
} }
@@ -203,26 +127,24 @@ export class Client implements SmppSession {
this.validateSession(reject); this.validateSession(reject);
this.validateBound(reject); this.validateBound(reject);
if (!job.count || !job.perSecond) { if (!job.count || !job.perSecond) {
reject(`Client-${this._id} sendMultiple failed: invalid job, missing fields`); reject(`Client-${this.getId()} sendMultiple failed: invalid job, missing fields`);
} }
this.logger.log1(`Client-${this._id} sending multiple messages: ${JSON.stringify(job)}`); this.logger.log1(`Client-${this.getId()} sending multiple messages: ${JSON.stringify(job)}`);
this.status = ClientStatus.BUSY; this.setStatus(4);
let counter = 0; let counter = 0;
let previousUpdateCounter = 0; let previousUpdateCounter = 0;
this.counterUpdateTimer = new NanoTimer();
this.counterUpdateTimer.setInterval(() => { this.counterUpdateTimer.setInterval(() => {
if (previousUpdateCounter !== counter) { if (previousUpdateCounter !== counter) {
this.eventEmitter.emit(Client.EVENTS.MESSAGE_SEND_COUNTER_UPDATE_EVENT, counter); this.eventEmitter.emit(this.EVENT.MESSAGE_SEND_COUNTER_UPDATE_EVENT, counter);
previousUpdateCounter = counter; previousUpdateCounter = counter;
} }
}, '', `${MESSAGE_SEND_UPDATE_DELAY / 1000} s`); }, '', `${MESSAGE_SEND_UPDATE_DELAY / 1000} s`);
let count = job.count || 1; let count = job.count || 1;
let interval = 1 / (job.perSecond || 1); let interval = 1 / (job.perSecond || 1);
this.sendTimer = new NanoTimer();
this.sendTimer.setInterval(() => { this.sendTimer.setInterval(() => {
if (count > 0 && counter >= count) { if (count > 0 && counter >= count) {
this.cancelSendInterval(); this.cancelSendInterval();
@@ -236,43 +158,17 @@ export class Client implements SmppSession {
}); });
} }
sendSingle(job: Job): Promise<object> { getUrl(): string {
return this.sendPdu(job.pdu); return this.url;
}
cancelSendInterval(): void {
if (this.sendTimer) {
this.sendTimer.clearInterval();
this.counterUpdateTimer.clearInterval();
this.sendTimer = null;
this.counterUpdateTimer = null;
}
this.status = ClientStatus.BOUND;
}
on(event: string, callback: (...args: any[]) => void): void {
this.eventEmitter.on(event, callback);
}
getId(): number {
return this._id;
}
sendMultipleDefault(): Promise<void> {
return this.sendMultiple(this.getDefaultMultipleJob());
}
sendSingleDefault(): Promise<object> {
return this.sendSingle(this.getDefaultSingleJob());
} }
private connectSession(): Promise<void> { private connectSession(): Promise<void> {
return new Promise<void>((resolve, reject) => { return new Promise<void>((resolve, reject) => {
this.validateFields(reject); this.validateFields(reject);
this.logger.log1(`Client-${this._id} connecting to ${this._url}`); this.logger.log1(`Client-${this.getId()} connecting to ${this.url}`);
this.session = smpp.connect({ this.session = smpp.connect({
url: this._url, auto_enquire_link_period: AUTO_ENQUIRE_LINK_PERIOD, url: this.url, auto_enquire_link_period: AUTO_ENQUIRE_LINK_PERIOD,
}, this.eventSessionConnected.bind(this)); }, this.eventSessionConnected.bind(this));
this.session.on('error', this.eventSessionError.bind(this)); this.session.on('error', this.eventSessionError.bind(this));
this.session.on('close', this.eventSessionClose.bind(this)); this.session.on('close', this.eventSessionClose.bind(this));
@@ -281,40 +177,35 @@ export class Client implements SmppSession {
} }
private eventSessionConnected(): void { private eventSessionConnected(): void {
this.logger.log1(`Client-${this._id} connected to ${this._url}`); this.logger.log1(`Client-${this.getId()} connected to ${this.url}`);
this.status = ClientStatus.CONNECTED; this.setStatus(2);
if (this.connectPromise) { if (this.connectPromise) {
this.connectPromise.resolve(); this.connectPromise.resolve();
} }
} }
private eventAnyPdu(pdu: any): void {
this.logger.log6(`Client-${this._id} received PDU: ${JSON.stringify(pdu)}`);
this.eventEmitter.emit(Client.EVENTS.ANY_PDU, pdu);
}
private eventSessionError(pdu: any): void { private eventSessionError(pdu: any): void {
this.logger.log1(`Client-${this._id} error on ${this._url}`); this.logger.log1(`Client-${this.getId()} error on ${this.url}`);
this.status = ClientStatus.NOT_CONNECTED; this.setStatus(0);
this.rejectPromises(pdu); this.rejectPromises();
} }
private eventSessionClose(): void { private eventSessionClose(): void {
this.logger.log1(`Client-${this._id} closed on ${this._url}`); this.logger.log1(`Client-${this.getId()} closed on ${this.url}`);
this.status = ClientStatus.NOT_CONNECTED; this.setStatus(0);
this.rejectPromises(); this.rejectPromises();
} }
private eventBindReply(pdu: any): void { private eventBindReply(pdu: any): void {
if (pdu.command_status === 0) { if (pdu.command_status === 0) {
this.logger.log1(`Client-${this._id} bound to ${this._url}`); this.logger.log1(`Client-${this.getId()} bound to ${this.url}`);
this.status = ClientStatus.BOUND; this.setStatus(4);
if (this.bindPromise) { if (this.bindPromise) {
this.bindPromise.resolve(); this.bindPromise.resolve();
} }
} else { } else {
this.logger.log1(`Client-${this._id} bind failed to ${this.url}`); this.logger.log1(`Client-${this.getId()} bind failed to ${this.url}`);
this.status = ClientStatus.CONNECTED; this.setStatus(2);
if (this.bindPromise) { if (this.bindPromise) {
this.bindPromise.reject(pdu); this.bindPromise.reject(pdu);
} }
@@ -328,21 +219,24 @@ export class Client implements SmppSession {
if (this.bindPromise) { if (this.bindPromise) {
this.bindPromise.reject(err); this.bindPromise.reject(err);
} }
if (this.closePromise) {
this.closePromise.resolve();
}
} }
private validateFields(reject: (reason?: any) => void) { private validateFields(reject: (reason?: any) => void) {
if (!this._url) { if (!this.url) {
let error = `Client-${this._id} has no url set`; let error = `Client-${this.getId()} has no url set`;
this.logger.log1(error); this.logger.log1(error);
reject(error); reject(error);
} }
if (!this._username) { if (!this.username) {
let error = `Client-${this._id} has no username set`; let error = `Client-${this.getId()} has no username set`;
this.logger.log1(error); this.logger.log1(error);
reject(error); reject(error);
} }
if (!this._password) { if (!this.password) {
let error = `Client-${this._id} has no password set`; let error = `Client-${this.getId()} has no password set`;
this.logger.log1(error); this.logger.log1(error);
reject(error); reject(error);
} }
@@ -350,15 +244,15 @@ export class Client implements SmppSession {
private validateSession(reject: (reason?: any) => void) { private validateSession(reject: (reason?: any) => void) {
if (!this.session) { if (!this.session) {
let errorMessage = `Client-${this._id} session is not defined`; let errorMessage = `Client-${this.getId()} session is not defined`;
this.logger.log1(errorMessage); this.logger.log1(errorMessage);
reject(errorMessage); reject(errorMessage);
} }
} }
private validateBound(reject: (reason?: any) => void) { private validateBound(reject: (reason?: any) => void) {
if (this._status !== ClientStatus.BOUND) { if (this.status !== this.STATUS[4]) {
let errorMessage = `Client-${this._id} is not bound`; let errorMessage = `Client-${this.getId()} is not bound`;
this.logger.log1(errorMessage); this.logger.log1(errorMessage);
reject(errorMessage); reject(errorMessage);
} }

1
src/Client/ClientSessionManager.ts
View File

@@ -1,4 +1,3 @@
import EventEmitter from "events";
import Logger from "../Logger"; import Logger from "../Logger";
import {SessionManager} from "../SessionManager"; import {SessionManager} from "../SessionManager";
import {SmppSession} from "../SmppSession"; import {SmppSession} from "../SmppSession";

6
src/Client/ClientStatus.ts
View File

@@ -1,8 +1,2 @@
export default class ClientStatus { export default class ClientStatus {
static readonly CONNECTING: string = "CONNECTING";
static readonly CONNECTED: string = "CONNECTED";
static readonly BUSY: string = "BUSY";
static readonly BINDING: string = "BINDING";
static readonly BOUND: string = "BOUND";
static readonly NOT_CONNECTED: string = "NOT CONNECTED";
} }

8
src/Job/Job.ts
View File

@@ -1,9 +1,9 @@
import EventEmitter from "events"; import EventEmitter from "events";
import {JobEvents} from "./JobEvents";
const smpp = require("smpp"); const smpp = require("smpp");
export class Job { export class Job {
static readonly STATE_CHANGED: string = "STATE_CHANGED";
private eventEmitter: EventEmitter = new EventEmitter(); private eventEmitter: EventEmitter = new EventEmitter();
constructor(pdu: any, perSecond?: number, count?: number) { constructor(pdu: any, perSecond?: number, count?: number) {
@@ -20,7 +20,7 @@ export class Job {
set pdu(value: any) { set pdu(value: any) {
this._pdu = value; this._pdu = value;
this.eventEmitter.emit(JobEvents.STATE_CHANGED, {}); this.eventEmitter.emit(Job.STATE_CHANGED, {});
} }
private _perSecond?: number; private _perSecond?: number;
@@ -31,7 +31,7 @@ export class Job {
set perSecond(value: number) { set perSecond(value: number) {
this._perSecond = value; this._perSecond = value;
this.eventEmitter.emit(JobEvents.STATE_CHANGED, {}); this.eventEmitter.emit(Job.STATE_CHANGED, {});
} }
private _count?: number; private _count?: number;
@@ -42,7 +42,7 @@ export class Job {
set count(value: number) { set count(value: number) {
this._count = value; this._count = value;
this.eventEmitter.emit(JobEvents.STATE_CHANGED, {}); this.eventEmitter.emit(Job.STATE_CHANGED, {});
} }
static deserialize(serialized: any): Job { static deserialize(serialized: any): Job {

2
src/SessionManager.ts
View File

@@ -58,7 +58,7 @@ export abstract class SessionManager {
} }
serialize(): object { serialize(): object {
this.logger.log1(`Serializing ${this.sessions.length} clients`) this.logger.log1(`Serializing ${this.sessions.length} clients`);
return this.sessions.map((session: SmppSession) => { return this.sessions.map((session: SmppSession) => {
return session.serialize(); return session.serialize();
}); });

4
src/SmppSession.ts
View File

@@ -61,8 +61,6 @@ export abstract class SmppSession {
abstract close(): Promise<void>; abstract close(): Promise<void>;
abstract initialize(): void;
abstract serialize(): object; abstract serialize(): object;
on(event: string, callback: (...args: any[]) => void): void { on(event: string, callback: (...args: any[]) => void): void {
@@ -168,7 +166,7 @@ export abstract class SmppSession {
}); });
}); });
if (successful === 0) { if (successful === 0) {
return Promise.reject("No PDU processor was able to process the PDU"); return Promise.resolve("No PDU processor was able to process the PDU");
} else { } else {
return Promise.resolve(); return Promise.resolve();
} }