ami: TCP client with parser, list actions, ping and reconnect backoff
Parser tests replay a real Asterisk 24 session recorded byte for byte, in every chunking from 1 byte up. Client tests use an in-process fake AMI. Checked live against the container: login, command, hard kill rejects the pending action, reconnect follows the 1/2/4/8/16/30 s backoff. Task: M2-T1
This commit is contained in:
iamdoubz committed
1 parent
9a81e37725
commit
70b0a92666
7 files changed
+800
No files matched your search
@@ -0,0 +1,194 @@
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
// AmiClient against an in-process fake AMI server.
|
||||
import { createServer, type AddressInfo, type Socket } from 'node:net';
|
||||
import { afterEach, describe, expect, it } from 'vitest';
|
||||
import { AmiParser, serialize } from './events.ts';
|
||||
import { AmiActionError, AmiClient, AmiDisconnectedError, type AmiOptions } from './client.ts';
|
||||
|
||||
const silent = { info() {}, warn() {}, error() {}, debug() {} };
|
||||
|
||||
type Action = Record<string, string>;
|
||||
type Handler = (a: Action, reply: (f: Action) => void, raw: (s: string) => void) => void;
|
||||
|
||||
/** A fake Asterisk: banner, password check on Login, then `handler` answers each action. */
|
||||
async function fakeAmi(handler: Handler, port = 0) {
|
||||
const sockets = new Set<Socket>();
|
||||
const logins: Action[] = [];
|
||||
const server = createServer((s) => {
|
||||
sockets.add(s);
|
||||
s.on('close', () => sockets.delete(s));
|
||||
s.setEncoding('utf8');
|
||||
s.write('Asterisk Call Manager/13.0.0\r\n');
|
||||
const p = new AmiParser();
|
||||
p.push('Asterisk Call Manager/13.0.0\r\n'); // client→server traffic has no banner
|
||||
s.on('data', (d: string) => {
|
||||
for (const m of p.push(d)) {
|
||||
const a = m as Action;
|
||||
const reply = (f: Action) => s.write(serialize({ ...f, ActionID: a.ActionID! }));
|
||||
if (a.Action !== 'Login') handler(a, reply, (raw) => s.write(raw));
|
||||
else {
|
||||
logins.push(a);
|
||||
reply(
|
||||
a.Secret === 'right'
|
||||
? { Response: 'Success', Message: 'Authentication accepted' }
|
||||
: { Response: 'Error', Message: 'Authentication failed' },
|
||||
);
|
||||
}
|
||||
}
|
||||
});
|
||||
});
|
||||
await new Promise<void>((r) => server.listen(port, '127.0.0.1', r));
|
||||
return {
|
||||
port: (server.address() as AddressInfo).port,
|
||||
logins,
|
||||
broadcast: (f: Action) => sockets.forEach((s) => s.write(serialize(f))),
|
||||
dropAll: () => sockets.forEach((s) => s.destroy()),
|
||||
close: () =>
|
||||
new Promise<void>((r) => {
|
||||
sockets.forEach((s) => s.destroy());
|
||||
server.close(() => r());
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
const standard: Handler = (a, reply, raw) => {
|
||||
if (a.Action === 'Command')
|
||||
raw(
|
||||
`Response: Success\r\nActionID: ${a.ActionID}\r\nMessage: Command output follows\r\nOutput: line a\r\nOutput: line b\r\n\r\n`,
|
||||
);
|
||||
else if (a.Action === 'Ping') reply({ Response: 'Success', Ping: 'Pong' });
|
||||
else reply({ Response: 'Error', Message: `Invalid/unknown command: ${a.Action}` });
|
||||
};
|
||||
|
||||
let cleanups: (() => unknown)[] = [];
|
||||
afterEach(async () => {
|
||||
for (const c of cleanups.reverse()) await c();
|
||||
cleanups = [];
|
||||
});
|
||||
|
||||
async function setup(handler: Handler, extra: Partial<AmiOptions> = {}) {
|
||||
const srv = await fakeAmi(handler);
|
||||
const ami = new AmiClient({
|
||||
host: '127.0.0.1',
|
||||
port: srv.port,
|
||||
username: 'u',
|
||||
secret: 'right',
|
||||
logger: silent,
|
||||
backoffMs: [20, 40],
|
||||
...extra,
|
||||
});
|
||||
cleanups.push(
|
||||
() => srv.close(),
|
||||
() => ami.stop(),
|
||||
);
|
||||
return { srv, ami };
|
||||
}
|
||||
|
||||
const once = (ami: AmiClient, ev: string) => new Promise((r) => ami.once(ev, r));
|
||||
const until = async (cond: () => boolean, ms = 3000) => {
|
||||
for (const end = Date.now() + ms; !cond(); await new Promise((r) => setTimeout(r, 10)))
|
||||
if (Date.now() > end) throw new Error('timeout');
|
||||
};
|
||||
|
||||
describe('AmiClient', () => {
|
||||
it('logs in with Events: on and emits connected', async () => {
|
||||
const { srv, ami } = await setup(standard);
|
||||
ami.start();
|
||||
await once(ami, 'connected');
|
||||
expect(ami.isConnected()).toBe(true);
|
||||
expect(srv.logins[0]).toMatchObject({
|
||||
Action: 'Login',
|
||||
Username: 'u',
|
||||
Secret: 'right',
|
||||
Events: 'on',
|
||||
ActionID: 'sp-1',
|
||||
});
|
||||
});
|
||||
|
||||
it('command() joins repeated Output lines; errors reject with AmiActionError', async () => {
|
||||
const { ami } = await setup(standard);
|
||||
ami.start();
|
||||
await once(ami, 'connected');
|
||||
expect(await ami.command('core show version')).toBe('line a\nline b');
|
||||
await expect(ami.action({ Action: 'Nope' })).rejects.toBeInstanceOf(AmiActionError);
|
||||
});
|
||||
|
||||
it('collects list actions until EventList: Complete, and maps "No … found" to []', async () => {
|
||||
const { srv, ami } = await setup((a, reply) => {
|
||||
if (a.Action === 'CoreShowChannels') {
|
||||
reply({ Response: 'Success', EventList: 'start', Message: 'Channels will follow' });
|
||||
reply({ Event: 'CoreShowChannel', Channel: 'PJSIP/101-0001' });
|
||||
reply({ Event: 'CoreShowChannel', Channel: 'PJSIP/102-0002' });
|
||||
reply({ Event: 'CoreShowChannelsComplete', EventList: 'Complete', ListItems: '2' });
|
||||
} else if (a.Action === 'PJSIPShowContacts')
|
||||
reply({ Response: 'Error', Message: 'No Contacts found' });
|
||||
else reply({ Response: 'Error', Message: 'Permission denied' });
|
||||
});
|
||||
ami.start();
|
||||
await once(ami, 'connected');
|
||||
const events: string[] = [];
|
||||
ami.on('event', (m) => events.push(String(m.Event)));
|
||||
const list = await ami.actionList({ Action: 'CoreShowChannels' });
|
||||
expect(list.map((m) => m.Channel)).toEqual(['PJSIP/101-0001', 'PJSIP/102-0002']);
|
||||
expect(await ami.actionList({ Action: 'PJSIPShowContacts' })).toEqual([]);
|
||||
await expect(ami.actionList({ Action: 'QueueStatus' })).rejects.toBeInstanceOf(AmiActionError);
|
||||
expect(events).toEqual([]); // list items don't leak out as normal events
|
||||
srv.broadcast({ Event: 'ExtensionStatus', Exten: '101', Status: '1' });
|
||||
await until(() => events.length === 1);
|
||||
});
|
||||
|
||||
it('emits event and event:<Name>', async () => {
|
||||
const { srv, ami } = await setup(standard);
|
||||
ami.start();
|
||||
await once(ami, 'connected');
|
||||
const got = once(ami, 'event:ContactStatus');
|
||||
srv.broadcast({ Event: 'ContactStatus', AOR: '101', ContactStatus: 'Reachable' });
|
||||
await expect(got).resolves.toMatchObject({ AOR: '101', ContactStatus: 'Reachable' });
|
||||
});
|
||||
|
||||
it('rejects pending actions on disconnect, then reconnects with backoff', async () => {
|
||||
const { srv, ami } = await setup(() => {}); // never answers
|
||||
ami.start();
|
||||
await once(ami, 'connected');
|
||||
const pending = ami.action({ Action: 'Status' });
|
||||
const disconnected = once(ami, 'disconnected');
|
||||
srv.dropAll();
|
||||
await expect(pending).rejects.toBeInstanceOf(AmiDisconnectedError);
|
||||
await disconnected;
|
||||
expect(ami.isConnected()).toBe(false);
|
||||
await expect(ami.action({ Action: 'Ping' })).rejects.toBeInstanceOf(AmiDisconnectedError);
|
||||
await once(ami, 'connected');
|
||||
expect(srv.logins).toHaveLength(2);
|
||||
});
|
||||
|
||||
it('keeps retrying while Asterisk is down', async () => {
|
||||
const { srv, ami } = await setup(standard);
|
||||
ami.start();
|
||||
await once(ami, 'connected');
|
||||
await srv.close();
|
||||
await new Promise((r) => setTimeout(r, 150)); // several refused attempts
|
||||
const back = await fakeAmi(standard, srv.port);
|
||||
cleanups.push(() => back.close());
|
||||
await once(ami, 'connected');
|
||||
expect(back.logins).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('times out actions, and reconnects when Ping gets no answer', async () => {
|
||||
const { srv, ami } = await setup(() => {}, { pingEveryMs: 50, pingTimeoutMs: 50 });
|
||||
ami.start();
|
||||
await once(ami, 'connected');
|
||||
await expect(ami.action({ Action: 'Status' }, { timeoutMs: 30 })).rejects.toThrow(/timed out/);
|
||||
await once(ami, 'disconnected');
|
||||
await once(ami, 'connected');
|
||||
expect(srv.logins.length).toBeGreaterThanOrEqual(2);
|
||||
});
|
||||
|
||||
it('never connects with a wrong secret, but keeps trying', async () => {
|
||||
const { srv, ami } = await setup(standard, { secret: 'wrong' });
|
||||
let connected = false;
|
||||
ami.on('connected', () => (connected = true));
|
||||
ami.start();
|
||||
await until(() => srv.logins.length >= 2);
|
||||
expect(connected).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,231 @@
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
// Our own AMI client (04 §4): connect, log in, reconnect forever, match replies by ActionID.
|
||||
import { EventEmitter } from 'node:events';
|
||||
import { Socket } from 'node:net';
|
||||
import type { FastifyBaseLogger } from 'fastify';
|
||||
import { AmiParser, field, fields, serialize, type AmiMessage } from './events.ts';
|
||||
|
||||
export class AmiDisconnectedError extends Error {
|
||||
constructor() {
|
||||
super('AMI disconnected');
|
||||
}
|
||||
}
|
||||
export class AmiActionError extends Error {
|
||||
readonly response: AmiMessage;
|
||||
constructor(response: AmiMessage) {
|
||||
super(`AMI action failed: ${field(response, 'Message') ?? 'unknown error'}`);
|
||||
this.response = response;
|
||||
}
|
||||
}
|
||||
|
||||
type Pending = {
|
||||
resolve: (m: AmiMessage) => void;
|
||||
reject: (e: Error) => void;
|
||||
timer: NodeJS.Timeout;
|
||||
list?: { items: AmiMessage[]; done: (items: AmiMessage[]) => void };
|
||||
};
|
||||
|
||||
export type AmiOptions = {
|
||||
host: string;
|
||||
port: number;
|
||||
username: string;
|
||||
secret: string;
|
||||
logger: Pick<FastifyBaseLogger, 'info' | 'warn' | 'error' | 'debug'>;
|
||||
/** Overridable for tests. */
|
||||
backoffMs?: number[];
|
||||
pingEveryMs?: number;
|
||||
pingTimeoutMs?: number;
|
||||
};
|
||||
|
||||
const DEFAULT_BACKOFF = [1, 2, 4, 8, 16, 30].map((s) => s * 1000);
|
||||
|
||||
export class AmiClient extends EventEmitter {
|
||||
private readonly o: Required<AmiOptions>;
|
||||
private socket: Socket | null = null;
|
||||
private readonly parser = new AmiParser();
|
||||
private readonly pending = new Map<string, Pending>();
|
||||
private nextId = 0;
|
||||
private attempt = 0;
|
||||
private connected = false;
|
||||
private stopped = true;
|
||||
private pingTimer: NodeJS.Timeout | null = null;
|
||||
private reconnectTimer: NodeJS.Timeout | null = null;
|
||||
|
||||
constructor(opts: AmiOptions) {
|
||||
super();
|
||||
this.o = {
|
||||
backoffMs: DEFAULT_BACKOFF,
|
||||
pingEveryMs: 20_000,
|
||||
pingTimeoutMs: 10_000,
|
||||
...opts,
|
||||
};
|
||||
}
|
||||
|
||||
start(): void {
|
||||
this.stopped = false;
|
||||
this.connect();
|
||||
}
|
||||
|
||||
stop(): void {
|
||||
this.stopped = true;
|
||||
if (this.reconnectTimer) clearTimeout(this.reconnectTimer);
|
||||
this.socket?.destroy();
|
||||
}
|
||||
|
||||
isConnected(): boolean {
|
||||
return this.connected;
|
||||
}
|
||||
|
||||
private connect(): void {
|
||||
this.parser.reset();
|
||||
const s = new Socket();
|
||||
this.socket = s;
|
||||
s.setEncoding('utf8');
|
||||
s.setNoDelay(true);
|
||||
s.on('data', (chunk: string) => {
|
||||
for (const msg of this.parser.push(chunk)) this.onMessage(msg);
|
||||
});
|
||||
s.on('error', (err) => this.o.logger.debug({ err: err.message }, 'ami socket error'));
|
||||
s.on('close', () => this.onClose(s));
|
||||
s.connect(this.o.port, this.o.host, () => void this.login());
|
||||
}
|
||||
|
||||
private async login(): Promise<void> {
|
||||
try {
|
||||
// Login fields carry the secret: send() never logs them (04 §4).
|
||||
await this.send({
|
||||
Action: 'Login',
|
||||
Username: this.o.username,
|
||||
Secret: this.o.secret,
|
||||
Events: 'on',
|
||||
});
|
||||
} catch (err) {
|
||||
this.o.logger.error({ err: (err as Error).message }, 'ami login failed');
|
||||
this.socket?.destroy();
|
||||
return;
|
||||
}
|
||||
this.connected = true;
|
||||
this.attempt = 0;
|
||||
this.o.logger.info({ banner: this.parser.banner }, 'ami connected');
|
||||
this.startPing();
|
||||
this.emit('connected');
|
||||
}
|
||||
|
||||
private startPing(): void {
|
||||
this.pingTimer = setInterval(() => {
|
||||
this.action({ Action: 'Ping' }, { timeoutMs: this.o.pingTimeoutMs }).catch(() => {
|
||||
this.o.logger.warn('ami ping timed out, reconnecting');
|
||||
this.socket?.destroy();
|
||||
});
|
||||
}, this.o.pingEveryMs);
|
||||
}
|
||||
|
||||
private onClose(s: Socket): void {
|
||||
if (s !== this.socket) return;
|
||||
if (this.pingTimer) clearInterval(this.pingTimer);
|
||||
this.pingTimer = null;
|
||||
const wasConnected = this.connected;
|
||||
this.connected = false;
|
||||
for (const [id, p] of this.pending) {
|
||||
clearTimeout(p.timer);
|
||||
p.reject(new AmiDisconnectedError());
|
||||
this.pending.delete(id);
|
||||
}
|
||||
if (wasConnected) {
|
||||
this.o.logger.warn('ami disconnected');
|
||||
this.emit('disconnected');
|
||||
}
|
||||
if (this.stopped) return;
|
||||
const delay = this.o.backoffMs[Math.min(this.attempt, this.o.backoffMs.length - 1)]!;
|
||||
this.attempt++;
|
||||
this.reconnectTimer = setTimeout(() => this.connect(), delay);
|
||||
}
|
||||
|
||||
private onMessage(msg: AmiMessage): void {
|
||||
const id = field(msg, 'ActionID');
|
||||
const p = id ? this.pending.get(id) : undefined;
|
||||
if (p && id) {
|
||||
if (msg.Response !== undefined) {
|
||||
if (field(msg, 'Response') === 'Error') {
|
||||
this.finish(id);
|
||||
p.reject(new AmiActionError(msg));
|
||||
} else if (!p.list) {
|
||||
this.finish(id);
|
||||
p.resolve(msg);
|
||||
} else if (field(msg, 'EventList') !== 'start') {
|
||||
// A list action that answered without a list (e.g. nothing to report).
|
||||
this.finish(id);
|
||||
p.list.done([]);
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (p.list && msg.Event !== undefined) {
|
||||
if (field(msg, 'EventList') === 'Complete') {
|
||||
this.finish(id);
|
||||
p.list.done(p.list.items);
|
||||
} else {
|
||||
p.list.items.push(msg);
|
||||
}
|
||||
return;
|
||||
}
|
||||
}
|
||||
const name = field(msg, 'Event');
|
||||
if (name) {
|
||||
this.emit('event', msg);
|
||||
this.emit(`event:${name}`, msg);
|
||||
}
|
||||
}
|
||||
|
||||
private finish(id: string): void {
|
||||
const p = this.pending.get(id);
|
||||
if (p) clearTimeout(p.timer);
|
||||
this.pending.delete(id);
|
||||
}
|
||||
|
||||
private send(
|
||||
fields: Record<string, string>,
|
||||
opts: { timeoutMs?: number; list?: (items: AmiMessage[]) => void } = {},
|
||||
): Promise<AmiMessage> {
|
||||
const s = this.socket;
|
||||
if (!s || s.destroyed) return Promise.reject(new AmiDisconnectedError());
|
||||
const id = `sp-${++this.nextId}`;
|
||||
const wire = serialize({ ...fields, ActionID: id });
|
||||
return new Promise((resolve, reject) => {
|
||||
const timer = setTimeout(() => {
|
||||
this.pending.delete(id);
|
||||
reject(new Error(`AMI action ${fields.Action} timed out`));
|
||||
}, opts.timeoutMs ?? 10_000);
|
||||
this.pending.set(id, {
|
||||
resolve,
|
||||
reject,
|
||||
timer,
|
||||
list: opts.list ? { items: [], done: opts.list } : undefined,
|
||||
});
|
||||
s.write(wire);
|
||||
});
|
||||
}
|
||||
|
||||
action(fields: Record<string, string>, opts?: { timeoutMs?: number }): Promise<AmiMessage> {
|
||||
if (!this.connected) return Promise.reject(new AmiDisconnectedError());
|
||||
return this.send(fields, opts);
|
||||
}
|
||||
|
||||
/** Sends a list action and collects its events up to the "...Complete" event. */
|
||||
actionList(fields: Record<string, string>, opts?: { timeoutMs?: number }): Promise<AmiMessage[]> {
|
||||
if (!this.connected) return Promise.reject(new AmiDisconnectedError());
|
||||
return new Promise((resolve, reject) => {
|
||||
this.send(fields, { ...opts, list: resolve }).catch((e: Error) => {
|
||||
// Some list actions answer an empty list with an error, e.g. "No Contacts found".
|
||||
if (e instanceof AmiActionError && /^No .* found/i.test(field(e.response, 'Message') ?? ''))
|
||||
resolve([]);
|
||||
else reject(e);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/** Action: Command. Returns the CLI output lines joined with \n. */
|
||||
async command(cmd: string): Promise<string> {
|
||||
const r = await this.action({ Action: 'Command', Command: cmd });
|
||||
return fields(r, 'Output').join('\n');
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,73 @@
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
import { readFileSync } from 'node:fs';
|
||||
import { describe, expect, it } from 'vitest';
|
||||
import { AmiParser, field, fields, serialize } from './events.ts';
|
||||
|
||||
// Raw bytes Asterisk 24.0.0-rc2 sent for: Login, Command, CoreShowChannels, PJSIPShowContacts (none),
|
||||
// ExtensionStateList, Command (multi-line), DBGet (missing), Ping, an unknown action.
|
||||
const SESSION = readFileSync(
|
||||
new URL('../../test/fixtures/ami/asterisk-24-session.raw', import.meta.url),
|
||||
'latin1',
|
||||
);
|
||||
|
||||
const parseAll = (chunks: string[]) => {
|
||||
const p = new AmiParser();
|
||||
return { msgs: chunks.flatMap((c) => p.push(c)), banner: p.banner };
|
||||
};
|
||||
|
||||
describe('AmiParser', () => {
|
||||
const whole = parseAll([SESSION]);
|
||||
|
||||
it('reads the banner and every message of a real session', () => {
|
||||
expect(whole.banner).toBe('Asterisk Call Manager/13.0.0');
|
||||
const byId = (id: string) => whole.msgs.filter((m) => m.ActionID === id);
|
||||
expect(byId('sp-1')[0]).toMatchObject({
|
||||
Response: 'Success',
|
||||
Message: 'Authentication accepted',
|
||||
});
|
||||
expect(field(byId('sp-2')[0]!, 'Output')).toMatch(/^Asterisk 24\./);
|
||||
expect(byId('sp-3').map((m) => m.EventList)).toEqual(['start', 'Complete']);
|
||||
// Asterisk ends this value with a bare LF.
|
||||
expect(byId('sp-4')[0]).toEqual({
|
||||
Response: 'Error',
|
||||
ActionID: 'sp-4',
|
||||
Message: 'No Contacts found',
|
||||
});
|
||||
expect(byId('sp-5').filter((m) => m.Event === 'ExtensionStatus')).toHaveLength(20);
|
||||
expect(fields(byId('sp-6')[0]!, 'Output')).toHaveLength(3);
|
||||
expect(byId('sp-8')[0]).toMatchObject({ Ping: 'Pong' });
|
||||
expect(byId('sp-9')[0]!.Response).toBe('Error');
|
||||
});
|
||||
|
||||
it('gives the same result for any TCP chunking', () => {
|
||||
const oneByte = parseAll(SESSION.split(''));
|
||||
expect(oneByte).toEqual(whole);
|
||||
for (const size of [2, 3, 7, 50, 333]) {
|
||||
const chunks: string[] = [];
|
||||
for (let i = 0; i < SESSION.length; i += size) chunks.push(SESSION.slice(i, i + size));
|
||||
expect(parseAll(chunks)).toEqual(whole);
|
||||
}
|
||||
// Two messages and part of a third in one chunk.
|
||||
const p = new AmiParser();
|
||||
const msgs = p.push(
|
||||
'Asterisk Call Manager/13.0.0\r\nEvent: A\r\n\r\nEvent: B\r\nX: 1\r\n\r\nEvent: C\r\n',
|
||||
);
|
||||
expect(msgs.map((m) => m.Event)).toEqual(['A', 'B']);
|
||||
expect(p.push('\r\n').map((m) => m.Event)).toEqual(['C']);
|
||||
});
|
||||
|
||||
it('keeps colons inside values and turns repeated keys into arrays', () => {
|
||||
const [m] = new AmiParser().push(
|
||||
'Asterisk Call Manager/13.0.0\r\nEvent: VarSet\r\nValue: a: b: c\r\nChanVariable: x=1\r\nChanVariable: y=2\r\n\r\n',
|
||||
);
|
||||
expect(m).toEqual({ Event: 'VarSet', Value: 'a: b: c', ChanVariable: ['x=1', 'y=2'] });
|
||||
});
|
||||
|
||||
it('serializes actions and refuses header injection', () => {
|
||||
expect(serialize({ Action: 'Ping', ActionID: 'sp-1' })).toBe(
|
||||
'Action: Ping\r\nActionID: sp-1\r\n\r\n',
|
||||
);
|
||||
expect(() => serialize({ Action: 'Command', Command: 'x\r\nAction: Logoff' })).toThrow();
|
||||
expect(() => serialize({ Action: 'Command', Command: 'x\ny' })).toThrow();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,83 @@
|
||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
// AMI wire format (04 §4): one banner line, then messages of "Key: Value\r\n" lines ending with an
|
||||
// empty line. Keys can repeat (Output:, ChanVariable:), so a repeated key becomes an array.
|
||||
|
||||
export type AmiMessage = Record<string, string | string[]>;
|
||||
|
||||
const BANNER = /^Asterisk Call Manager\/[^\r\n]*\r\n/;
|
||||
|
||||
/** Incremental parser: feed it TCP chunks, get whole messages back. */
|
||||
export class AmiParser {
|
||||
private buf = '';
|
||||
private sawBanner = false;
|
||||
banner: string | null = null;
|
||||
|
||||
push(chunk: string): AmiMessage[] {
|
||||
this.buf += chunk;
|
||||
if (!this.sawBanner) {
|
||||
const m = BANNER.exec(this.buf);
|
||||
if (!m) {
|
||||
// Wait for the whole banner line, unless this clearly isn't one.
|
||||
if (this.buf.length < 64 && !this.buf.includes('\n')) return [];
|
||||
} else {
|
||||
this.banner = m[0].trim();
|
||||
this.buf = this.buf.slice(m[0].length);
|
||||
}
|
||||
this.sawBanner = true;
|
||||
}
|
||||
const out: AmiMessage[] = [];
|
||||
for (let end = this.buf.indexOf('\r\n\r\n'); end !== -1; end = this.buf.indexOf('\r\n\r\n')) {
|
||||
const block = this.buf.slice(0, end);
|
||||
this.buf = this.buf.slice(end + 4);
|
||||
const msg = parseBlock(block);
|
||||
if (Object.keys(msg).length) out.push(msg);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
reset(): void {
|
||||
this.buf = '';
|
||||
this.sawBanner = false;
|
||||
this.banner = null;
|
||||
}
|
||||
}
|
||||
|
||||
export function parseBlock(block: string): AmiMessage {
|
||||
const msg: AmiMessage = {};
|
||||
for (const line of block.split('\r\n')) {
|
||||
const i = line.indexOf(':');
|
||||
if (i <= 0) continue;
|
||||
const key = line.slice(0, i);
|
||||
// Asterisk sometimes ends a value with a bare LF ("No Contacts found\n"): drop stray line ends.
|
||||
const value = line
|
||||
.slice(i + 1)
|
||||
.replace(/^ /, '')
|
||||
.replace(/[\r\n]+$/, '');
|
||||
const prev = msg[key];
|
||||
if (prev === undefined) msg[key] = value;
|
||||
else msg[key] = Array.isArray(prev) ? [...prev, value] : [prev, value];
|
||||
}
|
||||
return msg;
|
||||
}
|
||||
|
||||
export function serialize(fields: Record<string, string>): string {
|
||||
let s = '';
|
||||
for (const [k, v] of Object.entries(fields)) {
|
||||
// A CR or LF in a value would start a new header or end the message: AMI injection.
|
||||
if (/[\r\n]/.test(k) || /[\r\n]/.test(v)) throw new Error(`AMI field ${k} contains a newline`);
|
||||
s += `${k}: ${v}\r\n`;
|
||||
}
|
||||
return `${s}\r\n`;
|
||||
}
|
||||
|
||||
/** First value of a possibly repeated key. */
|
||||
export const field = (m: AmiMessage, key: string): string | undefined => {
|
||||
const v = m[key];
|
||||
return Array.isArray(v) ? v[0] : v;
|
||||
};
|
||||
|
||||
/** All values of a possibly repeated key. */
|
||||
export const fields = (m: AmiMessage, key: string): string[] => {
|
||||
const v = m[key];
|
||||
return v === undefined ? [] : Array.isArray(v) ? v : [v];
|
||||
};
|
||||
@@ -0,0 +1 @@
|
||||
* -text
|
||||
+215
@@ -0,0 +1,215 @@
|
||||
Asterisk Call Manager/13.0.0
|
||||
Response: Success
|
||||
ActionID: sp-1
|
||||
Message: Authentication accepted
|
||||
|
||||
Response: Success
|
||||
ActionID: sp-2
|
||||
Message: Command output follows
|
||||
Output: Asterisk 24.0.0-rc2 built by root @ buildkitsandbox on a x86_64 running Linux on 2026-10-17 19:25:59 UTC
|
||||
|
||||
Response: Success
|
||||
ActionID: sp-3
|
||||
EventList: start
|
||||
Message: Channels will follow
|
||||
|
||||
Event: CoreShowChannelsComplete
|
||||
ActionID: sp-3
|
||||
EventList: Complete
|
||||
ListItems: 0
|
||||
|
||||
Response: Error
|
||||
ActionID: sp-4
|
||||
Message: No Contacts found
|
||||
|
||||
|
||||
Response: Success
|
||||
ActionID: sp-5
|
||||
EventList: start
|
||||
Message: Extension Statuses will follow
|
||||
|
||||
Event: ExtensionStatus
|
||||
ActionID: sp-5
|
||||
Exten: 701
|
||||
Context: sp-parkedcalls
|
||||
Hint: park:701@sp-parkedcalls
|
||||
Status: 4
|
||||
StatusText: Unavailable
|
||||
|
||||
Event: ExtensionStatus
|
||||
ActionID: sp-5
|
||||
Exten: 702
|
||||
Context: sp-parkedcalls
|
||||
Hint: park:702@sp-parkedcalls
|
||||
Status: 4
|
||||
StatusText: Unavailable
|
||||
|
||||
Event: ExtensionStatus
|
||||
ActionID: sp-5
|
||||
Exten: 703
|
||||
Context: sp-parkedcalls
|
||||
Hint: park:703@sp-parkedcalls
|
||||
Status: 4
|
||||
StatusText: Unavailable
|
||||
|
||||
Event: ExtensionStatus
|
||||
ActionID: sp-5
|
||||
Exten: 704
|
||||
Context: sp-parkedcalls
|
||||
Hint: park:704@sp-parkedcalls
|
||||
Status: 4
|
||||
StatusText: Unavailable
|
||||
|
||||
Event: ExtensionStatus
|
||||
ActionID: sp-5
|
||||
Exten: 705
|
||||
Context: sp-parkedcalls
|
||||
Hint: park:705@sp-parkedcalls
|
||||
Status: 4
|
||||
StatusText: Unavailable
|
||||
|
||||
Event: ExtensionStatus
|
||||
ActionID: sp-5
|
||||
Exten: 706
|
||||
Context: sp-parkedcalls
|
||||
Hint: park:706@sp-parkedcalls
|
||||
Status: 4
|
||||
StatusText: Unavailable
|
||||
|
||||
Event: ExtensionStatus
|
||||
ActionID: sp-5
|
||||
Exten: 707
|
||||
Context: sp-parkedcalls
|
||||
Hint: park:707@sp-parkedcalls
|
||||
Status: 4
|
||||
StatusText: Unavailable
|
||||
|
||||
Event: ExtensionStatus
|
||||
ActionID: sp-5
|
||||
Exten: 708
|
||||
Context: sp-parkedcalls
|
||||
Hint: park:708@sp-parkedcalls
|
||||
Status: 4
|
||||
StatusText: Unavailable
|
||||
|
||||
Event: ExtensionStatus
|
||||
ActionID: sp-5
|
||||
Exten: 709
|
||||
Context: sp-parkedcalls
|
||||
Hint: park:709@sp-parkedcalls
|
||||
Status: 4
|
||||
StatusText: Unavailable
|
||||
|
||||
Event: ExtensionStatus
|
||||
ActionID: sp-5
|
||||
Exten: 710
|
||||
Context: sp-parkedcalls
|
||||
Hint: park:710@sp-parkedcalls
|
||||
Status: 4
|
||||
StatusText: Unavailable
|
||||
|
||||
Event: ExtensionStatus
|
||||
ActionID: sp-5
|
||||
Exten: 711
|
||||
Context: sp-parkedcalls
|
||||
Hint: park:711@sp-parkedcalls
|
||||
Status: 4
|
||||
StatusText: Unavailable
|
||||
|
||||
Event: ExtensionStatus
|
||||
ActionID: sp-5
|
||||
Exten: 712
|
||||
Context: sp-parkedcalls
|
||||
Hint: park:712@sp-parkedcalls
|
||||
Status: 4
|
||||
StatusText: Unavailable
|
||||
|
||||
Event: ExtensionStatus
|
||||
ActionID: sp-5
|
||||
Exten: 713
|
||||
Context: sp-parkedcalls
|
||||
Hint: park:713@sp-parkedcalls
|
||||
Status: 4
|
||||
StatusText: Unavailable
|
||||
|
||||
Event: ExtensionStatus
|
||||
ActionID: sp-5
|
||||
Exten: 714
|
||||
Context: sp-parkedcalls
|
||||
Hint: park:714@sp-parkedcalls
|
||||
Status: 4
|
||||
StatusText: Unavailable
|
||||
|
||||
Event: ExtensionStatus
|
||||
ActionID: sp-5
|
||||
Exten: 715
|
||||
Context: sp-parkedcalls
|
||||
Hint: park:715@sp-parkedcalls
|
||||
Status: 4
|
||||
StatusText: Unavailable
|
||||
|
||||
Event: ExtensionStatus
|
||||
ActionID: sp-5
|
||||
Exten: 716
|
||||
Context: sp-parkedcalls
|
||||
Hint: park:716@sp-parkedcalls
|
||||
Status: 4
|
||||
StatusText: Unavailable
|
||||
|
||||
Event: ExtensionStatus
|
||||
ActionID: sp-5
|
||||
Exten: 717
|
||||
Context: sp-parkedcalls
|
||||
Hint: park:717@sp-parkedcalls
|
||||
Status: 4
|
||||
StatusText: Unavailable
|
||||
|
||||
Event: ExtensionStatus
|
||||
ActionID: sp-5
|
||||
Exten: 718
|
||||
Context: sp-parkedcalls
|
||||
Hint: park:718@sp-parkedcalls
|
||||
Status: 4
|
||||
StatusText: Unavailable
|
||||
|
||||
Event: ExtensionStatus
|
||||
ActionID: sp-5
|
||||
Exten: 719
|
||||
Context: sp-parkedcalls
|
||||
Hint: park:719@sp-parkedcalls
|
||||
Status: 4
|
||||
StatusText: Unavailable
|
||||
|
||||
Event: ExtensionStatus
|
||||
ActionID: sp-5
|
||||
Exten: 720
|
||||
Context: sp-parkedcalls
|
||||
Hint: park:720@sp-parkedcalls
|
||||
Status: 4
|
||||
StatusText: Unavailable
|
||||
|
||||
Event: ExtensionStateListComplete
|
||||
ActionID: sp-5
|
||||
EventList: Complete
|
||||
ListItems: 20
|
||||
|
||||
Response: Success
|
||||
ActionID: sp-6
|
||||
Message: Command output follows
|
||||
Output: Module Description Use Count Status Support Level
|
||||
Output: res_pjsip.so Basic SIP resource 46 Running core
|
||||
Output: 1 modules loaded
|
||||
|
||||
Response: Error
|
||||
ActionID: sp-7
|
||||
Message: Database entry not found
|
||||
|
||||
Response: Success
|
||||
ActionID: sp-8
|
||||
Ping: Pong
|
||||
Timestamp: 1792271308.421406
|
||||
|
||||
Response: Error
|
||||
ActionID: sp-9
|
||||
Message: Invalid/unknown command: Nope. Use Action: ListCommands to show available commands.
|
||||
|
||||
@@ -85,6 +85,9 @@ Details:
|
||||
- Send `Action: Ping` every 20 s. If no reply in 10 s, destroy the socket and reconnect.
|
||||
- Reconnect backoff: 1, 2, 4, 8, 16, 30, 30, … seconds.
|
||||
- Pending actions reject with `AmiDisconnectedError` when the socket closes.
|
||||
- Some list actions answer an empty list with `Response: Error` instead of an empty EventList
|
||||
(Asterisk 24: `PJSIPShowContacts` → `Message: No Contacts found`, ending in a bare LF). `actionList`
|
||||
turns `No … found` errors into `[]`. The parser strips stray CR/LF from the end of values.
|
||||
- Default action timeout: 10 s.
|
||||
- **Never log the Login action fields.**
|
||||
|
||||
|
||||
Reference in new issue
Block a user