93 lines
2.9 KiB
TypeScript
93 lines
2.9 KiB
TypeScript
/**
|
|
* Minimal RESP (REdis Serialization Protocol) codec — just enough to drive
|
|
* Redis pub/sub over a raw TCP socket, with no external client dependency.
|
|
*
|
|
* Encodes commands as arrays of bulk strings, and streaming-parses the replies
|
|
* we care about: simple strings (+), errors (-), integers (:), bulk strings ($),
|
|
* and arrays (*) — which is what SUBSCRIBE/PSUBSCRIBE confirmations and
|
|
* message/pmessage pushes are made of.
|
|
*/
|
|
|
|
export type RespValue = string | number | null | RespValue[];
|
|
|
|
const encoder = new TextEncoder();
|
|
const decoder = new TextDecoder();
|
|
|
|
/** Encode a command (`["PUBLISH", topic, payload]`) as a RESP array of bulk strings. */
|
|
export function encodeCommand(args: string[]): Uint8Array {
|
|
let head = `*${args.length}\r\n`;
|
|
const parts: Uint8Array[] = [];
|
|
for (const arg of args) {
|
|
const bytes = encoder.encode(arg);
|
|
head += `$${bytes.length}\r\n`;
|
|
parts.push(encoder.encode(head), bytes, encoder.encode("\r\n"));
|
|
head = "";
|
|
}
|
|
return concat(parts);
|
|
}
|
|
|
|
/** Concatenate byte arrays. */
|
|
export function concat(parts: Uint8Array[]): Uint8Array {
|
|
let len = 0;
|
|
for (const p of parts) len += p.length;
|
|
const out = new Uint8Array(len);
|
|
let off = 0;
|
|
for (const p of parts) {
|
|
out.set(p, off);
|
|
off += p.length;
|
|
}
|
|
return out;
|
|
}
|
|
|
|
function indexOfCRLF(buf: Uint8Array, from: number): number {
|
|
for (let i = from; i + 1 < buf.length; i++) {
|
|
if (buf[i] === 13 && buf[i + 1] === 10) return i;
|
|
}
|
|
return -1;
|
|
}
|
|
|
|
/**
|
|
* Parse one RESP reply from `buf` at `off`. Returns the value and the offset just
|
|
* past it, or `null` if the buffer doesn't yet hold a complete reply (caller
|
|
* should wait for more bytes).
|
|
*/
|
|
export function parseReply(buf: Uint8Array, off = 0): { value: RespValue; next: number } | null {
|
|
if (off >= buf.length) return null;
|
|
const type = buf[off];
|
|
const lineEnd = indexOfCRLF(buf, off + 1);
|
|
if (lineEnd === -1) return null;
|
|
const line = decoder.decode(buf.subarray(off + 1, lineEnd));
|
|
const after = lineEnd + 2;
|
|
|
|
switch (type) {
|
|
case 43: // '+' simple string
|
|
case 45: // '-' error
|
|
return { value: line, next: after };
|
|
case 58: // ':' integer
|
|
return { value: Number(line), next: after };
|
|
case 36: {
|
|
// '$' bulk string
|
|
const len = Number(line);
|
|
if (len === -1) return { value: null, next: after };
|
|
if (after + len + 2 > buf.length) return null;
|
|
return { value: decoder.decode(buf.subarray(after, after + len)), next: after + len + 2 };
|
|
}
|
|
case 42: {
|
|
// '*' array
|
|
const count = Number(line);
|
|
if (count === -1) return { value: null, next: after };
|
|
const arr: RespValue[] = [];
|
|
let cur = after;
|
|
for (let i = 0; i < count; i++) {
|
|
const r = parseReply(buf, cur);
|
|
if (!r) return null;
|
|
arr.push(r.value);
|
|
cur = r.next;
|
|
}
|
|
return { value: arr, next: cur };
|
|
}
|
|
default:
|
|
return null;
|
|
}
|
|
}
|