30:T7ae,async function createClient({ socket = { host: "localhost", port: 6380, tls: false }, url = undefined, username = undefined, password = undefined, name = undefined, database = undefined } = {}) { const queue = new RedisCommandsQueue(); var ws = undefined; const newSocket = async () => { ws = new WebSocket( `ws${socket.tls ? "s" : ""}://${socket.host}:${socket.port}` ); ws.binaryType = "arraybuffer"; ws.onmessage = (evt) => { queue.parseResponse(evt.data); }; // wait for socket to open await new Promise((resolve, reject) => { ws.onclose = (evt) => reject(evt.code); ws.onopen = resolve; }); if (username || password) { if (!username) sendCommand(["AUTH", password]); else sendCommand(["AUTH", username, password], { asap: true }); } if (database) { sendCommand(["SELECT", database], { asap: true }); } return ws; }; const tick = async () => { while (true) { const args = queue.getCommandToSend(); if (args === undefined) break; if (ws.readyState > 1 /* CLOSING or CLOSED */) { await newSocket(); } var retries = 5; while (retries-- >= 0) { try { writeCommand(args); break; } catch (err) { console.error(err); await newSocket(); // try a reconnect } } } }; const sendCommand = (args, options) => { const reply = queue.addCommand(args, options); tick(); return reply; }; const writeCommand = (args) => { for (const toWrite of encodeCommand(args)) { ws.send(toWrite); } }; await newSocket(); return { sendCommand, subscribe: (channels, listener, bufferMode) => { const reply = queue.subscribe( C.PubSubSubscribeCommands.SUBSCRIBE, channels, listener, bufferMode ); tick(); return reply; } }; }31:T23dd,// source https://github.com/redis/node-redis/blob/a0de7967f98068790cd09c909bbbb7e7d6deefe8/packages/client/lib/client/commands-queue.ts#L66 class RedisCommandsQueue { static flushQueue(queue, err) { while (queue.length) { queue.shift().reject(err); } } static emitPubSubMessage(listenersMap, message, channel, pattern) { const keyString = decoder.decode(pattern || channel), listeners = listenersMap.get(keyString); if (!listeners) return; for (const listener of listeners.buffers) { listener(message, channel); } if (!listeners.strings.size) return; const messageString = decoder.decode(message), channelString = pattern ? decoder.decode(channel) : keyString; for (const listener of listeners.strings) { listener(messageString, channelString); } } constructor(maxLength = 10000) { this.waitingToBeSent = []; this.waitingForReply = []; this.maxLength = maxLength; this.parser = new Parser({ returnReply: (reply) => { if (this.pubSubState && Array.isArray(reply)) { if (equal(C.PUB_SUB_MESSAGES.message, reply[0])) { return RedisCommandsQueue.emitPubSubMessage( this.pubSubState.listeners.channels, reply[2], reply[1] ); } else if (equal(C.PUB_SUB_MESSAGES.pMessage, reply[0])) { return RedisCommandsQueue.emitPubSubMessage( this.pubSubState.listeners.patterns, reply[3], reply[2], reply[1] ); } else if ( equal(C.PUB_SUB_MESSAGES.subscribe, reply[0]) || equal(C.PUB_SUB_MESSAGES.pSubscribe.equals, reply[0]) || equal(C.PUB_SUB_MESSAGES.unsubscribe.equals, reply[0]) || equal(C.PUB_SUB_MESSAGES.pUnsubscribe.equals, reply[0]) ) { if (--this.waitingForReply[0].channelsCounter === 0) { this.shiftWaitingForReply().resolve(); } return; } } this.shiftWaitingForReply().resolve(reply); }, returnError: (err) => this.shiftWaitingForReply().reject(err) }); } addCommand(args, options) { if (this.pubSubState && !options?.ignorePubSubMode) { return Promise.reject(new Error("Cannot send commands in PubSub mode")); } else if ( this.maxLength && this.waitingToBeSent.length + this.waitingForReply.length >= this.maxLength ) { return Promise.reject(new Error("The queue is full")); } else if (options?.signal?.aborted) { return Promise.reject(new AbortError()); } return new Promise((resolve, reject) => { const node = { args, chainId: options?.chainId, returnBuffers: options?.returnBuffers, resolve, reject }; if (options?.signal) { const listener = () => { this.waitingToBeSent.removeNode(node); node.value.reject(new AbortError()); }; node.value.abort = { signal: options.signal, listener }; // AbortSignal type is incorrent options.signal.addEventListener("abort", listener, { once: true }); } if (options?.asap) { this.waitingToBeSent.unshift(node); } else { this.waitingToBeSent.push(node); } }); } initiatePubSubState() { return (this.pubSubState ??= { subscribed: 0, subscribing: 0, unsubscribing: 0, listeners: { channels: new Map(), patterns: new Map() } }); } subscribe(command, channels, listener, returnBuffers) { const pubSubState = this.initiatePubSubState(), channelsToSubscribe = [], listenersMap = command === C.PubSubSubscribeCommands.SUBSCRIBE ? pubSubState.listeners.channels : pubSubState.listeners.patterns; for (const channel of Array.isArray(channels) ? channels : [channels]) { const channelString = typeof channel === "string" ? channel : channel.toString(); let listeners = listenersMap.get(channelString); if (!listeners) { listeners = { buffers: new Set(), strings: new Set() }; listenersMap.set(channelString, listeners); channelsToSubscribe.push(channel); } // https://github.com/microsoft/TypeScript/issues/23132 (returnBuffers ? listeners.buffers : listeners.strings).add(listener); } if (!channelsToSubscribe.length) { return Promise.resolve(); } return this.pushPubSubCommand(command, channelsToSubscribe); } unsubscribe(command, channels, listener, returnBuffers) { if (!this.pubSubState) { return Promise.resolve(); } const listeners = command === C.PubSubUnsubscribeCommands.UNSUBSCRIBE ? this.pubSubState.listeners.channels : this.pubSubState.listeners.patterns; if (!channels) { const size = listeners.size; listeners.clear(); return this.pushPubSubCommand(command, size); } const channelsToUnsubscribe = []; for (const channel of Array.isArray(channels) ? channels : [channels]) { const sets = listeners.get(channel); if (!sets) continue; let shouldUnsubscribe; if (listener) { // https://github.com/microsoft/TypeScript/issues/23132 (returnBuffers ? sets.buffers : sets.strings).delete(listener); shouldUnsubscribe = !sets.buffers.size && !sets.strings.size; } else { shouldUnsubscribe = true; } if (shouldUnsubscribe) { channelsToUnsubscribe.push(channel); listeners.delete(channel); } } if (!channelsToUnsubscribe.length) { return Promise.resolve(); } return this.pushPubSubCommand(command, channelsToUnsubscribe); } pushPubSubCommand(command, channels) { return new Promise((resolve, reject) => { const pubSubState = this.initiatePubSubState(), isSubscribe = command === C.PubSubSubscribeCommands.SUBSCRIBE || command === C.PubSubSubscribeCommands.PSUBSCRIBE, inProgressKey = isSubscribe ? "subscribing" : "unsubscribing", commandArgs = [command]; let channelsCounter; if (typeof channels === "number") { // unsubscribe only channelsCounter = channels; } else { commandArgs.push(...channels); channelsCounter = channels.length; } pubSubState[inProgressKey] += channelsCounter; this.waitingToBeSent.push({ args: commandArgs, channelsCounter, returnBuffers: true, resolve: () => { pubSubState[inProgressKey] -= channelsCounter; if (isSubscribe) { pubSubState.subscribed += channelsCounter; } else { pubSubState.subscribed -= channelsCounter; if ( !pubSubState.subscribed && !pubSubState.subscribing && !pubSubState.subscribed ) { this.pubSubState = undefined; } } resolve(); }, reject: (err) => { pubSubState[inProgressKey] -= channelsCounter * (isSubscribe ? 1 : -1); reject(err); } }); }); } resubscribe() { if (!this.pubSubState) { return; } this.pubSubState.subscribed = 0; const promises = [], { channels, patterns } = this.pubSubState.listeners; if (channels.size) { promises.push( this.pushPubSubCommand(C.PubSubSubscribeCommands.SUBSCRIBE, [ ...channels.keys() ]) ); } if (patterns.size) { promises.push( this.pushPubSubCommand(C.PubSubSubscribeCommands.PSUBSCRIBE, [ ...patterns.keys() ]) ); } if (promises.length) { return Promise.all(promises); } } getCommandToSend() { const toSend = this.waitingToBeSent.shift(); if (toSend) { this.waitingForReply.push({ resolve: toSend.resolve, reject: toSend.reject, channelsCounter: toSend.channelsCounter, returnBuffers: toSend.returnBuffers }); } this.chainInExecution = toSend?.chainId; return toSend?.args; } parseResponse(data) { this.parser.setReturnBuffers( !!this.waitingForReply[0]?.returnBuffers || !!this.pubSubState?.subscribed ); this.parser.execute(data); } shiftWaitingForReply() { if (!this.waitingForReply.length) { throw new Error("Got an unexpected reply from Redis"); } return this.waitingForReply.shift(); } flushWaitingForReply(err) { RedisCommandsQueue.flushQueue(this.waitingForReply, err); if (!this.chainInExecution) { return; } while (this.waitingToBeSent.head?.value.chainId === this.chainInExecution) { this.waitingToBeSent.shift(); } this.chainInExecution = undefined; } flushAll(err) { RedisCommandsQueue.flushQueue(this.waitingForReply, err); RedisCommandsQueue.flushQueue(this.waitingToBeSent, err); } }32:T3e09,// Original source: https://github.com/NodeRedis/node-redis-parser/blob/4c2d31c8717f05dea1ffd91a0e45d68f452c73bd/lib/parser.js#L1 Parser = { var bufferPool = new ArrayBuffer(32 * 1024); var bufferOffset = 0; var interval = null; var counter = 0; var notDecreased = 0; /** * Used for integer numbers only * @param {JavascriptRedisParser} parser * @returns {undefined|number} */ function parseSimpleNumbers(parser) { const length = parser.buffer.byteLength - 1; var offset = parser.offset; var number = 0; var sign = 1; if (parser.buffer[offset] === 45) { sign = -1; offset++; } while (offset < length) { const c1 = parser.buffer[offset++]; if (c1 === 13) { // \r\n parser.offset = offset + 1; return sign * number; } number = number * 10 + (c1 - 48); } } /** * Used for integer numbers in case of the returnNumbers option * * Reading the string as parts of n SMI is more efficient than * using a string directly. * * @param {JavascriptRedisParser} parser * @returns {undefined|string} */ function parseStringNumbers(parser) { const length = parser.buffer.byteLength - 1; var offset = parser.offset; var number = 0; var res = ""; if (parser.buffer[offset] === 45) { res += "-"; offset++; } while (offset < length) { var c1 = parser.buffer[offset++]; if (c1 === 13) { // \r\n parser.offset = offset + 1; if (number !== 0) { res += number; } return res; } else if (number > 429496728) { res += number * 10 + (c1 - 48); number = 0; } else if (c1 === 48 && number === 0) { res += 0; } else { number = number * 10 + (c1 - 48); } } } /** * Parse a '+' redis simple string response but forward the offsets * onto convertBufferRange to generate a string. * @param {JavascriptRedisParser} parser * @returns {undefined|string|Buffer} */ function parseSimpleString(parser) { const start = parser.offset; const buffer = parser.buffer; const length = buffer.byteLength - 1; var offset = start; while (offset < length) { if (buffer[offset++] === 13) { // \r\n parser.offset = offset + 1; if (parser.optionReturnBuffers === true) { return parser.buffer.slice(start, offset - 1); } else { return decoder.decode(parser.buffer.subarray(start, offset - 1)); } } } } /** * Returns the read length * @param {JavascriptRedisParser} parser * @returns {undefined|number} */ function parseLength(parser) { const length = parser.buffer.byteLength - 1; var offset = parser.offset; var number = 0; while (offset < length) { const c1 = parser.buffer[offset++]; if (c1 === 13) { parser.offset = offset + 1; return number; } number = number * 10 + (c1 - 48); } } /** * Parse a ':' redis integer response * * If stringNumbers is activated the parser always returns numbers as string * This is important for big numbers (number > Math.pow(2, 53)) as js numbers * are 64bit floating point numbers with reduced precision * * @param {JavascriptRedisParser} parser * @returns {undefined|number|string} */ function parseInteger(parser) { if (parser.optionStringNumbers === true) { return parseStringNumbers(parser); } return parseSimpleNumbers(parser); } /** * Parse a '$' redis bulk string response * @param {JavascriptRedisParser} parser * @returns {undefined|null|string} */ function parseBulkString(parser) { const length = parseLength(parser); if (length === undefined) { return; } if (length < 0) { return null; } const offset = parser.offset + length; if (offset + 2 > parser.buffer.byteLength) { parser.bigStrSize = offset + 2; parser.totalChunkSize = parser.buffer.byteLength; parser.bufferCache.push(parser.buffer); return; } const start = parser.offset; parser.offset = offset + 2; if (parser.optionReturnBuffers === true) { return parser.buffer.slice(start, offset); } else { return decoder.decode(parser.buffer.subarray(start, offset)); } } /** * Parse a '-' redis error response * @param {JavascriptRedisParser} parser * @returns {ReplyError} */ function parseError(parser) { var string = parseSimpleString(parser); if (string !== undefined) { if (parser.optionReturnBuffers === true) { string = string.toString(); } return new ReplyError(string); } } /** * Parsing error handler, resets parser buffer * @param {JavascriptRedisParser} parser * @param {number} type * @returns {undefined} */ function handleError(parser, type) { const err = new ParserError( "Protocol error, got " + JSON.stringify(String.fromCharCode(type)) + " as reply type byte", JSON.stringify(parser.buffer), parser.offset ); parser.buffer = null; parser.returnFatalError(err); } /** * Parse a '*' redis array response * @param {JavascriptRedisParser} parser * @returns {undefined|null|any[]} */ function parseArray(parser) { const length = parseLength(parser); if (length === undefined) { return; } if (length < 0) { return null; } const responses = new Array(length); return parseArrayElements(parser, responses, 0); } /** * Push a partly parsed array to the stack * * @param {JavascriptRedisParser} parser * @param {any[]} array * @param {number} pos * @returns {undefined} */ function pushArrayCache(parser, array, pos) { parser.arrayCache.push(array); parser.arrayPos.push(pos); } /** * Parse chunked redis array response * @param {JavascriptRedisParser} parser * @returns {undefined|any[]} */ function parseArrayChunks(parser) { const tmp = parser.arrayCache.pop(); var pos = parser.arrayPos.pop(); if (parser.arrayCache.length) { const res = parseArrayChunks(parser); if (res === undefined) { pushArrayCache(parser, tmp, pos); return; } tmp[pos++] = res; } return parseArrayElements(parser, tmp, pos); } /** * Parse redis array response elements * @param {JavascriptRedisParser} parser * @param {Array} responses * @param {number} i * @returns {undefined|null|any[]} */ function parseArrayElements(parser, responses, i) { const bufferLength = parser.buffer.length; while (i < responses.length) { const offset = parser.offset; if (parser.offset >= bufferLength) { pushArrayCache(parser, responses, i); return; } const response = parseType(parser, parser.buffer[parser.offset++]); if (response === undefined) { if (!(parser.arrayCache.length || parser.bufferCache.length)) { parser.offset = offset; } pushArrayCache(parser, responses, i); return; } responses[i] = response; i++; } return responses; } /** * Called the appropriate parser for the specified type. * * 36: $ * 43: + * 42: * * 58: : * 45: - * * @param {JavascriptRedisParser} parser * @param {number} type * @returns {*} */ function parseType(parser, type) { switch (type) { case 36: return parseBulkString(parser); case 43: return parseSimpleString(parser); case 42: return parseArray(parser); case 58: return parseInteger(parser); case 45: return parseError(parser); default: return handleError(parser, type); } } /** * Decrease the bufferPool size over time * * Balance between increasing and decreasing the bufferPool. * Decrease the bufferPool by 10% by removing the first 10% of the current pool. * @returns {undefined} */ function decreaseBufferPool() { if (bufferPool.byteLength > 50 * 1024) { if (counter === 1 || notDecreased > counter * 2) { const minSliceLen = Math.floor(bufferPool.byteLength / 10); const sliceLength = minSliceLen < bufferOffset ? bufferOffset : minSliceLen; bufferOffset = 0; bufferPool = bufferPool.slice(sliceLength, bufferPool.byteLength); } else { notDecreased++; counter--; } } else { clearInterval(interval); counter = 0; notDecreased = 0; interval = null; } } /** * Check if the requested size fits in the current bufferPool. * If it does not, reset and increase the bufferPool accordingly. * * @param {number} length * @returns {undefined} */ function resizeBuffer(length) { if (bufferPool.byteLength < length + bufferOffset) { const multiplier = length > 1024 * 1024 * 75 ? 2 : 3; if (bufferOffset > 1024 * 1024 * 111) { bufferOffset = 1024 * 1024 * 50; } bufferPool = new Uint8Array(length * multiplier + bufferOffset); bufferOffset = 0; counter++; if (interval === null) { interval = setInterval(decreaseBufferPool, 50); } } } /** * Concat a bulk string containing multiple chunks * * Notes: * 1) The first chunk might contain the whole bulk string including the \r * 2) We are only safe to fully add up elements that are neither the first nor any of the last two elements * * @param {JavascriptRedisParser} parser * @returns {String} */ function concatBulkString(parser) { const list = parser.bufferCache; const oldOffset = parser.offset; var chunks = list.length; var offset = parser.bigStrSize - parser.totalChunkSize; parser.offset = offset; if (offset <= 2) { if (chunks === 2) { return list[0].toString("utf8", oldOffset, list[0].length + offset - 2); } chunks--; offset = list[list.length - 2].length + offset; } var res = decoder.write(list[0].slice(oldOffset)); for (var i = 1; i < chunks - 1; i++) { res += decoder.write(list[i]); } res += decoder.end(list[i].slice(0, offset - 2)); return res; } /** * Concat the collected chunks from parser.bufferCache. * * Increases the bufferPool size beforehand if necessary. * * @param {JavascriptRedisParser} parser * @returns {Buffer} */ function concatBulkBuffer(parser) { const list = parser.bufferCache; const oldOffset = parser.offset; const length = parser.bigStrSize - oldOffset - 2; var chunks = list.length; var offset = parser.bigStrSize - parser.totalChunkSize; parser.offset = offset; if (offset <= 2) { if (chunks === 2) { return list[0].slice(oldOffset, list[0].length + offset - 2); } chunks--; offset = list[list.length - 2].length + offset; } resizeBuffer(length); const start = bufferOffset; list[0].copy(bufferPool, start, oldOffset, list[0].length); bufferOffset += list[0].length - oldOffset; for (var i = 1; i < chunks - 1; i++) { list[i].copy(bufferPool, bufferOffset); bufferOffset += list[i].length; } list[i].copy(bufferPool, bufferOffset, 0, offset - 2); bufferOffset += offset - 2; return bufferPool.slice(start, bufferOffset); } class JavascriptRedisParser { /** * Javascript Redis Parser constructor * @param {{returnError: Function, returnReply: Function, returnFatalError?: Function, returnBuffers: boolean, stringNumbers: boolean }} options * @constructor */ constructor(options) { if (!options) { throw new TypeError("Options are mandatory."); } if ( typeof options.returnError !== "function" || typeof options.returnReply !== "function" ) { throw new TypeError( "The returnReply and returnError options have to be functions." ); } this.setReturnBuffers(!!options.returnBuffers); this.setStringNumbers(!!options.stringNumbers); this.returnError = options.returnError; this.returnFatalError = options.returnFatalError || options.returnError; this.returnReply = options.returnReply; this.reset(); } /** * Reset the parser values to the initial state * * @returns {undefined} */ reset() { this.offset = 0; this.buffer = null; this.bigStrSize = 0; this.totalChunkSize = 0; this.bufferCache = []; this.arrayCache = []; this.arrayPos = []; } /** * Set the returnBuffers option * * @param {boolean} returnBuffers * @returns {undefined} */ setReturnBuffers(returnBuffers) { if (typeof returnBuffers !== "boolean") { throw new TypeError("The returnBuffers argument has to be a boolean"); } this.optionReturnBuffers = returnBuffers; } /** * Set the stringNumbers option * * @param {boolean} stringNumbers * @returns {undefined} */ setStringNumbers(stringNumbers) { if (typeof stringNumbers !== "boolean") { throw new TypeError("The stringNumbers argument has to be a boolean"); } this.optionStringNumbers = stringNumbers; } /** * Parse the redis buffer * @param {Buffer} buffer * @returns {undefined} */ execute(buffer) { buffer = new Uint8Array(buffer); if (this.buffer === null) { this.buffer = buffer; this.offset = 0; } else if (this.bigStrSize === 0) { const oldLength = this.buffer.byteLength; const remainingLength = oldLength - this.offset; const newBuffer = new Uint8Array(remainingLength + buffer.byteLength); // target, targetStart, sourceStart, sourceEnd // this.buffer.copy(newBuffer, 0, this.offset, oldLength); newBuffer.set(this.buffer.subarray(this.offset, oldLength), 0); // target, targetStart, sourceStart, sourceEnd // buffer.copy(newBuffer, remainingLength, 0, buffer.byteLength); newBuffer.set(buffer.subarray(0, buffer.byteLength), remainingLength); this.buffer = newBuffer; this.offset = 0; if (this.arrayCache.length) { const arr = parseArrayChunks(this); if (arr === undefined) { return; } this.returnReply(arr); } } else if (this.totalChunkSize + buffer.byteLength >= this.bigStrSize) { this.bufferCache.push(buffer); var tmp = this.optionReturnBuffers ? concatBulkBuffer(this) : concatBulkString(this); this.bigStrSize = 0; this.bufferCache = []; this.buffer = buffer; if (this.arrayCache.length) { this.arrayCache[0][this.arrayPos[0]++] = tmp; tmp = parseArrayChunks(this); if (tmp === undefined) { return; } } this.returnReply(tmp); } else { this.bufferCache.push(buffer); this.totalChunkSize += buffer.length; return; } while (this.offset < this.buffer.byteLength) { const offset = this.offset; const type = this.buffer[this.offset++]; const response = parseType(this, type); if (response === undefined) { if (!(this.arrayCache.length || this.bufferCache.length)) { this.offset = offset; } return; } if (type === 45) { this.returnError(response); } else { this.returnReply(response); } } this.buffer = null; } } return JavascriptRedisParser; }33:[{"id":"f2e998f5d962cffa4b289f4d7c272aaaf390db875a4e6fb75ecd68f6787b10ca040ac26101042cf8a177cd8e492a75c040fd6b5089811e0c8a0b432bbcf80644","href":"https://static.observableusercontent.com/files/f2e998f5d962cffa4b289f4d7c272aaaf390db875a4e6fb75ecd68f6787b10ca040ac26101042cf8a177cd8e492a75c040fd6b5089811e0c8a0b432bbcf80644","name":"redis.png","type":"image/png","lastModified":1640265007123.35,"size":123339}] 2e:{"title":"Redis Serialization Protocol (RESP) Client for the Web","cells":[{"id":913,"value":"# [Redis Serialization Protocol](https://redis.io/topics/protocol) (RESP) Client for the Web","pinned":false,"hidden":false,"mode":"md"},{"id":905,"value":"FileAttachment(\"redis.png\").image({width: Math.min(640, width)})","pinned":false,"hidden":false,"mode":"ojs"},{"id":964,"value":"```js\nimport {createClient} from '@tomlarkworthy/redis'\n```","pinned":false,"hidden":false,"mode":"md"},{"id":0,"value":"Let's bring a hugely popular database to the web. This is a partial port of the [nodejs redis client](https://github.com/redis/node-redis) adapted for use with websockets. We don't bother creation dedication functions for the [zillions of commands](https://redis.io/commands) available, rather, we just port the generic `sendCommand([...])` catchall that gives you everything in one.\n\n```js\nconst client = await createClient()\nclient.sendCommand([\"SET\", \"k1\", \"foobar\"]);\nconst result = await client.sendCommand([\"GET\", \"k1\"]);\n// result == \"foobar\"\n```\n","pinned":false,"hidden":false,"mode":"md"},{"id":900,"value":"\nTo reach Redis from a browser you need to setup an intermediate websocket to TCP socket bridge (instructions below). Useful in privileged browser environments, such as behind a private network or with [webcode](https://webcode.run). \n\nThe Redis protocol is used by more than just [Redis](https://redis.io/)! See also [KeyDB](https://keydb.dev/) and [SSDB](https://github.com/ideawu/ssdb)","pinned":false,"hidden":false,"mode":"md"},{"id":926,"value":"tweet(\"1474008210650587144\")","pinned":false,"hidden":false,"mode":"ojs"},{"id":819,"value":"### Availible on NPM\n\nThis notebook hosts the source code. The source is published to NPM under [web-redis](https://www.npmjs.com/package/web-redis) using [distiller](https://observablehq.com/@tomlarkworthy/distiller).","pinned":false,"hidden":false,"mode":"md"},{"id":77,"value":"### Start REDIS\n\n```\ndocker run -it --rm --name redis -p 6379:6379 redis \n```","pinned":false,"hidden":false,"mode":"md"},{"id":79,"value":"### Install and Start websockify bridge\n\n```\nnpm install git+https://github.com/endpointservices/websockify-js\n```\n\n```\nnode_modules/websockify/websockify.js 6380 localhost:6379\n```\n\n Warning websockify cannot encrypt the upstream TCP session which may be ok if you run it local to the redis server.\n\nThe command we use for the demo server protected with ssh with LetsEncrypt certbot ends up as \n```\nkillall node;\nnohup node_modules/websockify/websockify.js \\\n--cert /etc/letsencrypt/live/redis.webcode.run/fullchain.pem \\\n--key /etc/letsencrypt/live/redis.webcode.run/privkey.pem \\\n443 localhost:6379 &\n```","pinned":false,"hidden":false,"mode":"md"},{"id":724,"value":"### Alternative: ws-tcp-proxy bridge\n\nAn alternative is a go websocket -> TCP proxy which does support TLS for the upstream, which would make it suitable for TLS encrypted REDIS services such as that on REDIS cloud\n\nhttps://github.com/zquestz/ws-tcp-proxy","pinned":false,"hidden":false,"mode":"md"},{"id":736,"value":"### Managed Redis providers\n\nhttps://geekflare.com/redis-hosting-platform/","pinned":false,"hidden":false,"mode":"md"},{"id":112,"value":"### createClient","pinned":false,"hidden":false,"mode":"md"},{"id":117,"value":"$30","pinned":false,"hidden":false,"mode":"ojs"},{"id":766,"value":"### Works with Redis Cloud\n(if you have a websocket proxy setup up)","pinned":false,"hidden":false,"mode":"md"},{"id":409,"value":"## Ported code\n\nPorting involves:-\n1. exchanging the use of Buffers for UInt8Array + TextEncoder. \n2. Sockets swapped for WebSocket.\n3. TS to JS conversion.\n","pinned":false,"hidden":false,"mode":"md"},{"id":206,"value":"function* encodeCommand(args) {\n let strings = `*${args.length}\\r\\n`,\n stringsLength = 0;\n for (var arg of args) {\n if (typeof arg === \"number\") arg = String(arg);\n const isString = typeof arg === \"string\",\n byteLength = isString ? encoder.encode(arg).length : arg.byteLength;\n strings += `$${byteLength}\\r\\n`;\n\n if (isString) {\n const totalLength = stringsLength + byteLength;\n if (totalLength > 1024) {\n yield strings;\n strings = arg;\n stringsLength = byteLength;\n } else {\n strings += arg;\n stringsLength = totalLength;\n }\n } else {\n yield strings;\n strings = \"\";\n stringsLength = 0;\n yield arg;\n }\n\n strings += \"\\r\\n\";\n }\n\n yield strings;\n}","pinned":false,"hidden":false,"mode":"ojs"},{"id":690,"value":"parseURL = (url) => {\n // https://www.iana.org/assignments/uri-schemes/prov/redis\n const { hostname, port, protocol, username, password, pathname } = new URL(\n url\n ),\n parsed = {\n socket: {\n host: hostname\n }\n };\n\n if (protocol === \"rediss:\") {\n parsed.socket.tls = true;\n } else if (protocol !== \"redis:\") {\n throw new TypeError(\"Invalid protocol\");\n }\n\n if (port) {\n parsed.socket.port = Number(port);\n }\n\n if (username) {\n parsed.username = decodeURIComponent(username);\n }\n\n if (password) {\n parsed.password = decodeURIComponent(password);\n }\n\n if (pathname.length > 1) {\n const database = Number(pathname.substring(1));\n if (isNaN(database)) {\n throw new TypeError(\"Invalid pathname\");\n }\n\n parsed.database = database;\n }\n\n return parsed;\n}","pinned":false,"hidden":false,"mode":"ojs"},{"id":462,"value":"C = ({\n PUB_SUB_MESSAGES: {\n message: encoder.encode(\"message\"),\n pMessage: encoder.encode(\"pmessage\"),\n subscribe: encoder.encode(\"subscribe\"),\n pSubscribe: encoder.encode(\"psubscribe\"),\n unsubscribe: encoder.encode(\"unsunscribe\"),\n pUnsubscribe: encoder.encode(\"punsubscribe\")\n },\n PubSubSubscribeCommands: {\n SUBSCRIBE: \"SUBSCRIBE\",\n PSUBSCRIBE: \"PSUBSCRIBE\"\n },\n PubSubUnsubscribeCommands: {\n UNSUBSCRIBE: \"UNSUBSCRIBE\",\n PUNSUBSCRIBE: \"PUNSUBSCRIBE\"\n }\n})","pinned":false,"hidden":false,"mode":"ojs"},{"id":423,"value":"$31","pinned":false,"hidden":false,"mode":"ojs"},{"id":166,"value":"$32","pinned":false,"hidden":false,"mode":"ojs"},{"id":573,"value":"#### [redis-errors](https://github.com/NodeRedis/redis-errors)","pinned":false,"hidden":false,"mode":"md"},{"id":332,"value":"class RedisError extends Error {\n get name() {\n return this.constructor.name;\n }\n}","pinned":false,"hidden":false,"mode":"ojs"},{"id":567,"value":"class AbortError extends RedisError {\n get name() {\n return this.constructor.name;\n }\n}","pinned":false,"hidden":false,"mode":"ojs"},{"id":569,"value":"class InterruptError extends AbortError {\n get name() {\n return this.constructor.name;\n }\n}","pinned":false,"hidden":false,"mode":"ojs"},{"id":330,"value":"class ReplyError extends RedisError {\n constructor(message) {\n const tmp = Error.stackTraceLimit;\n Error.stackTraceLimit = 2;\n super(message);\n Error.stackTraceLimit = tmp;\n }\n get name() {\n return this.constructor.name;\n }\n}","pinned":false,"hidden":false,"mode":"ojs"},{"id":350,"value":"class ParserError extends RedisError {\n constructor(message, buffer, offset) {\n const tmp = Error.stackTraceLimit;\n Error.stackTraceLimit = 2;\n super(message);\n Error.stackTraceLimit = tmp;\n this.offset = offset;\n this.buffer = buffer;\n }\n\n get name() {\n return this.constructor.name;\n }\n}","pinned":false,"hidden":false,"mode":"ojs"},{"id":414,"value":"## Support","pinned":false,"hidden":false,"mode":"md"},{"id":227,"value":"encoder = new TextEncoder()","pinned":false,"hidden":false,"mode":"ojs"},{"id":635,"value":"decoder = new TextDecoder()","pinned":false,"hidden":false,"mode":"ojs"},{"id":625,"value":"function equal(buf1, buf2) {\n if (buf1.byteLength != buf2.byteLength) return false;\n for (var i = 0; i != buf1.byteLength; i++) {\n if (buf1[i] != buf2[i]) return false;\n }\n return true;\n}","pinned":false,"hidden":false,"mode":"ojs"},{"id":746,"value":"### Testing","pinned":false,"hidden":false,"mode":"md"},{"id":755,"value":"testing = {\n footer;\n const [{ Runtime }, { default: define }] = await Promise.all([\n import(\n \"https://cdn.jsdelivr.net/npm/@observablehq/runtime@4/dist/runtime.js\"\n ),\n import(`https://api.observablehq.com/@tomlarkworthy/testing.js?v=3`)\n ]);\n const module = new Runtime().module(define);\n return Object.fromEntries(\n await Promise.all(\n [\"expect\", \"createSuite\"].map((n) => module.value(n).then((v) => [n, v]))\n )\n );\n}","pinned":false,"hidden":false,"mode":"ojs"},{"id":758,"value":"expect = testing.expect","pinned":false,"hidden":false,"mode":"ojs"},{"id":94,"value":"viewof tests = testing.createSuite({\n name: \"Redis Serialization Protocol Test Suite\",\n timeout_ms: 1000\n})","pinned":false,"hidden":false,"mode":"ojs"},{"id":384,"value":"viewof rerun = Inputs.button(\"rerun tests\")","pinned":false,"hidden":false,"mode":"ojs"},{"id":937,"value":"demoServer = ({\n host: \"redis.webcode.run\",\n port: 443,\n tls: true\n})","pinned":false,"hidden":false,"mode":"ojs"},{"id":130,"value":"client = (rerun,\ncreateClient({\n socket: demoServer\n}))","pinned":false,"hidden":false,"mode":"ojs"},{"id":657,"value":"blockClient = (client,\ncreateClient({\n socket: demoServer\n}))","pinned":false,"hidden":false,"mode":"ojs"},{"id":379,"value":"subClient = (blockClient,\ncreateClient({\n socket: demoServer\n}))","pinned":false,"hidden":false,"mode":"ojs"},{"id":872,"value":"transClient = (subClient,\ncreateClient({\n socket: demoServer\n}))","pinned":false,"hidden":false,"mode":"ojs"},{"id":283,"value":"tests.test(\n `createClient to non-existent server throws code 1006`,\n async (done) => {\n try {\n await createClient({ socket: { host: \"localhost\", port: 19999 } });\n } catch (err) {\n expect(err).toBe(1006);\n done();\n }\n }\n)","pinned":false,"hidden":false,"mode":"ojs"},{"id":830,"value":"### Basic commands","pinned":false,"hidden":false,"mode":"md"},{"id":142,"value":"p = tests.test(`sendCommand([\"PING\"]) returns 'PONG'`, async () => {\n const result = await client.sendCommand([\"PING\"]);\n expect(result).toBe(\"PONG\");\n})","pinned":true,"hidden":false,"mode":"ojs"},{"id":198,"value":"setKey = tests.test(\n 'sendCommand([\"SET\", \"key\", \"value\"]) returns \"OK\"',\n async () => {\n const result = await client.sendCommand([\"SET\", \"key\", \"value\"]);\n expect(result).toBe(\"OK\");\n }\n)","pinned":true,"hidden":false,"mode":"ojs"},{"id":241,"value":"b3 = tests.test('sendCommand([\"GET\", \"key\"]) returns foobar', async () => {\n setKey; // ensure setKey run first\n const result = await client.sendCommand([\"GET\", \"key\"]);\n expect(result).toBe(\"value\");\n})","pinned":true,"hidden":false,"mode":"ojs"},{"id":834,"value":"### Pubsub","pinned":false,"hidden":false,"mode":"md"},{"id":893,"value":"subscribing to a topic switch the client to pub/sub mode (so you can't execute other commands) where messages on topics will be delivered to the client.","pinned":false,"hidden":false,"mode":"md"},{"id":611,"value":"tests.test(\n 'subscribe(\"foo\", listener) receives (message, foo)',\n async (done) => {\n await subClient.subscribe(\"foo\", (message, channel) => {\n expect(message).toBe(\"bar\");\n expect(channel).toBe(\"foo\");\n done();\n });\n client.sendCommand([\"PUBLISH\", \"foo\", \"bar\"]);\n }\n)","pinned":true,"hidden":false,"mode":"ojs"},{"id":841,"value":"### Blocking","pinned":false,"hidden":false,"mode":"md"},{"id":879,"value":"BLPOP is a blocking left pop, it will take an element from a list, blocking until the list fills with an element if necessary. ","pinned":false,"hidden":false,"mode":"md"},{"id":645,"value":"tests.test(\"BLPOP blocks until RPUSH\", async () => {\n const result = blockClient.sendCommand([\"BLPOP\", \"list0\", \"0\"]);\n client.sendCommand([\"LPUSH\", \"list0\", \"v1\"]);\n expect(await result).toEqual([\"list0\", \"v1\"]);\n})","pinned":true,"hidden":false,"mode":"ojs"},{"id":845,"value":"### Transactions","pinned":false,"hidden":false,"mode":"md"},{"id":884,"value":"MULTI/EXEC blocks will execute intermediate commands atomically. This is a very important foundational semantic of Redis. ","pinned":false,"hidden":false,"mode":"md"},{"id":851,"value":"multi1 = tests.test(\"MULTI/EXEC block\", async () => {\n transClient.sendCommand([\"MULTI\"]);\n transClient.sendCommand([\"SET\", \"k0\", \"v0\"]);\n transClient.sendCommand([\"SET\", \"k1\", \"v0\"]);\n const result = await transClient.sendCommand([\"EXEC\"]);\n expect(result).toEqual([\"OK\", \"OK\"]);\n})","pinned":true,"hidden":false,"mode":"ojs"},{"id":979,"value":"multi2 = tests.test(\"MULTI with no EXEC throws\", async (done) => {\n try {\n await transClient.sendCommand([\"EXEC\"]);\n } catch (err) {\n expect(err.message).toContain(\"ERR EXEC without MULTI\");\n done();\n }\n})","pinned":false,"hidden":false,"mode":"ojs"},{"id":888,"value":"WATCH combines with MULTI/EXEC to guaranteed pin key values, allowing read/modify/write transactions.","pinned":false,"hidden":false,"mode":"md"},{"id":858,"value":"tests.test(\"WATCH atomic increment\", async () => {\n multi1;\n transClient.sendCommand([\"WATCH\", \"k1\"]);\n const v = transClient.sendCommand([\"GET\", \"k1\"]);\n transClient.sendCommand([\"MULTI\"]);\n transClient.sendCommand([\"SET\", \"k1\", Number.parseInt(await v) + 1 + \"\"]);\n const result = await transClient.sendCommand([\"EXEC\"]);\n expect(result).toEqual([\"OK\"]);\n})","pinned":true,"hidden":false,"mode":"ojs"},{"id":923,"value":"import { tweet } from \"@mbostock/tweet\"","pinned":false,"hidden":false,"mode":"ojs"},{"id":929,"value":"import { footer } from \"@endpointservices/footer\"","pinned":false,"hidden":false,"mode":"ojs"},{"id":932,"value":"footer","pinned":false,"hidden":false,"mode":"ojs"}],"files":"$W33","stdlib":"1","titling":"auto"}