Redis client improvements

package.json

@@ -43,12 +43,16 @@
     "tslib": "2"
   },
   "dependencies": {
-    "json-joy": "^11.25.0",
-    "thingies": "^1.16.0"
+    "@jsonjoy.com/jit-router": "^1.0.1",
+    "@jsonjoy.com/util": "^1.1.3",
+    "json-joy": "^16.8.0",
+    "sonic-forest": "^1.0.3",
+    "thingies": "^2.1.0",
+    "tree-dump": "^1.0.1"
   },
   "devDependencies": {
     "@types/benchmark": "^2.1.5",
-    "@types/jest": "^29.5.11",
+    "@types/jest": "^29.5.12",
     "benchmark": "^2.1.4",
     "cluster-key-slot": "^1.1.2",
     "commands": "https://github.com/streamich/commands#4321d5d40473c48fadf49fd99662032eac9b855b",
@@ -57,13 +61,13 @@
     "prettier": "^3.1.1",
     "pretty-quick": "^3.1.3",
     "redis": "^4.6.11",
-    "rimraf": "^5.0.5",
-    "ts-jest": "^29.1.1",
+    "rimraf": "^5.0.7",
+    "ts-jest": "^29.1.4",
     "ts-node": "^10.9.2",
-    "tslib": "^2.6.2",
+    "tslib": "^2.6.3",
     "tslint": "^6.1.3",
     "tslint-config-common": "^1.6.2",
-    "typescript": "^5.3.3"
+    "typescript": "^5.4.5"
   },
   "jest": {
     "verbose": true,

src/__tests__/setupStandalone.ts

@@ -1,8 +1,8 @@
 import * as net from 'net';
 import {StandaloneClient} from '../standalone/StandaloneClient';
 import {StandaloneTestSetup} from './types';
-import {RespEncoder} from 'json-joy/es2020/json-pack/resp/RespEncoder';
-import {RespStreamingDecoder} from 'json-joy/es2020/json-pack/resp/RespStreamingDecoder';
+import {RespEncoder} from '@jsonjoy.com/json-pack/lib/resp/RespEncoder';
+import {RespStreamingDecoder} from '@jsonjoy.com/json-pack/lib/resp/RespStreamingDecoder';
 import {ReconnectingSocket} from '../util/ReconnectingSocket';
 import * as config from './config';
 

src/__tests__/standalone-singleton.spec.ts

@@ -3,8 +3,8 @@ import {StandaloneClient} from '../standalone/StandaloneClient';
 import {ClusterTestSetup} from './types';
 import * as commands from './commands';
 import {ReconnectingSocket} from '../util/ReconnectingSocket';
-import {RespEncoder} from 'json-joy/es2020/json-pack/resp/RespEncoder';
-import {RespStreamingDecoder} from 'json-joy/es2020/json-pack/resp/RespStreamingDecoder';
+import {RespEncoder} from '@jsonjoy.com/json-pack/lib/resp/RespEncoder';
+import {RespStreamingDecoder} from '@jsonjoy.com/json-pack/lib/resp/RespStreamingDecoder';
 import * as config from './config';
 
 const host = config.standalone.host;

src/cluster/RedisCluster.ts

@@ -1,6 +1,6 @@
-import {FanOut} from 'thingies/es2020/fanout';
-import {RespEncoder} from 'json-joy/es2020/json-pack/resp';
-import {RespStreamingDecoder} from 'json-joy/es2020/json-pack/resp/RespStreamingDecoder';
+import {FanOut} from 'thingies/lib/fanout';
+import {RespEncoder} from '@jsonjoy.com/json-pack/lib/resp';
+import {RespStreamingDecoder} from '@jsonjoy.com/json-pack/lib/resp/RespStreamingDecoder';
 import * as commands from '../generated/commands';
 import {Cmd, MultiCmd, PartialExcept, RedisClientCodecOpts} from '../types';
 import {RedisClusterRouter} from './RedisClusterRouter';
@@ -10,18 +10,23 @@ import {isAskError, isMovedError, parseMovedError} from './errors';
 import {RedisClusterNodeClient, RedisClusterNodeClientOpts} from './RedisClusterNodeClient';
 import {RedirectType} from './constants';
 import {withTimeout} from '../util/timeout';
-import {printTree} from 'json-joy/es2020/util/print/printTree';
+import {printTree} from 'tree-dump/lib/printTree';
 import {getSlotAny} from '../util/slots';
 import {isMultiCmd} from '../util/commands';
 import {ScriptRegistry} from '../ScriptRegistry';
-import {ascii} from '../util/buf';
-import type {Printable} from 'json-joy/es2020/util/print/types';
-import type {CmdOpts} from '../standalone';
+import {cmpUint8Array, ascii} from '../util/buf';
 import {isNoscriptError} from '../standalone/errors';
+import {AvlMap} from 'sonic-forest/lib/avl/AvlMap';
+import {bufferToUint8Array} from '@jsonjoy.com/util/lib/buffers/bufferToUint8Array';
+import type {Printable} from 'tree-dump/lib/types';
+import type {CmdOpts} from '../standalone';
 
 const EVALSHA = ascii`EVALSHA`;
 const SCRIPT = ascii`SCRIPT`;
 const LOAD = ascii`LOAD`;
+const SSUBSCRIBE = ascii`SSUBSCRIBE`;
+const SPUBLISH = ascii`SPUBLISH`;
+const SUNSUBSCRIBE = ascii`SUNSUBSCRIBE`;
 
 export interface RedisClusterOpts extends RedisClientCodecOpts {
   /**
@@ -268,7 +273,7 @@ export class RedisCluster implements Printable {
     return client;
   }
 
-  public async getClientForKey(key: string, write: boolean): Promise<RedisClusterNodeClient> {
+  public async getClientForKey(key: string | Uint8Array, write: boolean): Promise<RedisClusterNodeClient> {
     if (!key) return await this.getAnyClientOrSeedClient();
     const slot = getSlotAny(key);
     await this.whenRouterReady();
@@ -278,7 +283,7 @@ export class RedisCluster implements Printable {
     return await this.getAnyClientOrSeedClient();
   }
 
-  public async getMasterClientForKey(key: string): Promise<RedisClusterNodeClient> {
+  public async getMasterClientForKey(key: string | Uint8Array): Promise<RedisClusterNodeClient> {
     if (!this._routerReady) await this.whenRouterReady();
     const router = this.router;
     const slot = getSlotAny(key);
@@ -337,6 +342,7 @@ export class RedisCluster implements Printable {
       if (typeof cmd !== 'string' || !cmd) throw new Error('INVALID_COMMAND');
       cmd = cmd.toUpperCase();
       isWrite = commands.write.has(cmd);
+      // FIXME: handle case when `args[1]` is Uint8Array or Buffer.
       if (!key) key = (args.length > 1 ? args[1] + '' : '') || '';
     }
     if (!call.client) {
@@ -397,6 +403,39 @@ export class RedisCluster implements Printable {
     }
   }
 
+  // ------------------------------------------------------------------ Pub/sub
+
+  public readonly ssubs = new AvlMap<Uint8Array, FanOut<Uint8Array>>(cmpUint8Array);
+
+  public async ssubscribe(channel: Uint8Array | string, listener: (message: Uint8Array) => void): Promise<() => void> {
+    // TODO: Remove bufferToUint8Array?
+    const channelBuf = typeof channel === 'string' ? bufferToUint8Array(Buffer.from(channel)) : channel;
+    let fanout = this.ssubs.get(channelBuf);
+    // TODO: Need to keep a client reference here ...
+    if (!fanout) {
+      await this.cmd([SSUBSCRIBE, channelBuf]);
+      fanout = new FanOut<Uint8Array>();
+      this.ssubs.set(channelBuf, fanout);
+    }
+    const unsubscribe = fanout.listen(listener);
+    return () => {
+      unsubscribe();
+      if (fanout!.listeners.size === 0) {
+        this.ssubs.del(channelBuf);
+        this.cmd([SUNSUBSCRIBE, channelBuf]);
+        // this.cmdFnF([SUNSUBSCRIBE, channelBuf]);
+      }
+    };
+  }
+
+  public async spublish(channel: Uint8Array | string, message: Uint8Array | string): Promise<number> {
+    return (await this.cmd([SPUBLISH, channel, message])) as number;
+  }
+
+  public async sunsubscribe(channel: Uint8Array | string): Promise<void> {
+
+  }
+
   // ---------------------------------------------------------------- Printable
 
   public toString(tab?: string): string {

src/cluster/RedisClusterNode.ts

@@ -1,6 +1,6 @@
 import {NodeHealth, NodeRole} from './constants';
-import {printTree} from 'json-joy/es2020/util/print/printTree';
-import type {Printable} from 'json-joy/es2020/util/print/types';
+import {printTree} from 'tree-dump/lib/printTree';
+import type {Printable} from 'tree-dump/lib/types';
 import type {RedisClusterShardsResponseNode} from './types';
 import type {RedisCluster} from './RedisCluster';
 

src/cluster/RedisClusterNodeClient.ts

@@ -1,9 +1,9 @@
 import * as tls from 'tls';
 import * as net from 'net';
 import {StandaloneClient} from '../standalone';
-import {printTree} from 'json-joy/es2020/util/print/printTree';
+import {printTree} from 'tree-dump/lib/printTree';
 import {ReconnectingSocket} from '../util/ReconnectingSocket';
-import type {Printable} from 'json-joy/es2020/util/print/types';
+import type {Printable} from 'tree-dump/lib/types';
 import type {RedisClientCodecOpts} from '../types';
 import type {RedisClusterShardsResponse} from './types';
 import type {ScriptRegistry} from '../ScriptRegistry';
@@ -84,6 +84,6 @@ export class RedisClusterNodeClient extends StandaloneClient implements Printabl
   // ---------------------------------------------------------------- Printable
 
   public toString(tab?: string): string {
-    return 'client' + printTree(tab, [(tab) => `host: ${this.host}`, (tab) => `port: ${this.port}`]);
+    return 'client' + printTree(tab, [(tab: string) => `host: ${this.host}`, (tab: string) => `port: ${this.port}`]);
   }
 }

src/cluster/RedisClusterRouter.ts

@@ -1,10 +1,10 @@
-import {AvlMap} from 'json-joy/es2020/util/trees/avl/AvlMap';
+import {AvlMap} from 'sonic-forest/lib/avl/AvlMap';
 import {RedisClusterSlotRange} from './RedisClusterSlotRange';
 import {RedisClusterNode} from './RedisClusterNode';
 import {NodeHealth, NodeRole} from './constants';
-import {xorShift32} from 'thingies/es2020/xorshift';
-import {printTree} from 'json-joy/es2020/util/print/printTree';
-import type {Printable} from 'json-joy/es2020/util/print/types';
+import {xorShift32} from 'thingies/lib/xorshift';
+import {printTree} from 'tree-dump/lib/printTree';
+import type {Printable} from 'tree-dump/lib/types';
 import type {RedisCluster} from './RedisCluster';
 import type {RedisClusterNodeClient} from './RedisClusterNodeClient';
 

src/cluster/RedisClusterSlotRange.ts

@@ -1,6 +1,6 @@
 import {RedisClusterNode} from './RedisClusterNode';
-import {printTree} from 'json-joy/es2020/util/print/printTree';
-import type {Printable} from 'json-joy/es2020/util/print/types';
+import {printTree} from 'tree-dump/lib/printTree';
+import type {Printable} from 'tree-dump/lib/types';
 
 export class RedisClusterSlotRange implements Printable {
   constructor(

src/cluster/__tests__/refs.spec.ts

@@ -1,5 +1,5 @@
 import {RedisCluster} from '../RedisCluster';
-import {tick} from 'thingies/es2020/tick';
+import {tick} from 'thingies/lib/tick';
 import {ScriptRegistry} from '../../ScriptRegistry';
 
 const setup = () => {

src/server/RedisCore.ts

@@ -1,6 +1,6 @@
-import {AvlNode} from 'json-joy/es2020/util/trees/avl/AvlMap';
-import {IAvlTreeNode} from 'json-joy/es2020/util/trees/avl/types';
-import {insert} from 'json-joy/es2020/util/trees/avl/util';
+import {AvlNode} from 'sonic-forest/lib/avl/AvlMap';
+import {IAvlTreeNode} from 'sonic-forest/lib/avl/types';
+import {insert} from 'sonic-forest/lib/avl/util';
 import {cmpUint8Array} from '../util/buf';
 
 export class KeyNode<V = unknown> implements IAvlTreeNode<Uint8Array, V> {

src/server/RedisMemoryServer.ts

@@ -1,8 +1,8 @@
-import {RespEncoder} from 'json-joy/es2020/json-pack/resp';
+import {FanOut} from 'thingies/lib/fanout';
+import {RespEncoder} from '@jsonjoy.com/json-pack/lib/resp';
+import {RespStreamingDecoder} from '@jsonjoy.com/json-pack/lib/resp/RespStreamingDecoder';
 import {RedisServer} from './RedisServer';
 import {StandaloneClient} from '../standalone';
-import {FanOut} from 'thingies/es2020/fanout';
-import {RespStreamingDecoder} from 'json-joy/es2020/json-pack/resp/RespStreamingDecoder';
 import {ParsedCmd, PublicKeys} from '../types';
 import {RedisServerConnection} from './connection/types';
 import type {ReconnectingSocket} from '../util/ReconnectingSocket';

src/server/RedisRuntime.ts

@@ -1,7 +1,6 @@
+import {Router, RouteMatcher} from '@jsonjoy.com/jit-router';
 import {ParsedCmd} from '../types';
 import {commands} from './commands';
-import {Router} from 'json-joy/es2020/util/router';
-import {RouteMatcher} from 'json-joy/es2020/util/router/codegen';
 import {RedisCore} from './RedisCore';
 import type {Command} from './Command';
 import type {RedisServerConnection} from './connection/types';

src/server/RedisTcpServer.ts

@@ -1,5 +1,5 @@
 import * as net from 'net';
-import {RespEncoder} from 'json-joy/es2020/json-pack/resp';
+import {RespEncoder} from '@jsonjoy.com/json-pack/lib/resp';
 import {RedisServerTcpConnection} from './connection/RedisServerTcpConnection';
 import {RedisServer} from './RedisServer';
 

src/server/connection/RedisServerTcpConnection.ts

@@ -1,7 +1,7 @@
 import * as net from 'net';
 import {RedisServerConnection} from './types';
-import {RespEncoder} from 'json-joy/es2020/json-pack/resp';
-import {RespStreamingDecoder} from 'json-joy/es2020/json-pack/resp/RespStreamingDecoder';
+import {RespEncoder} from '@jsonjoy.com/json-pack/lib/resp';
+import {RespStreamingDecoder} from '@jsonjoy.com/json-pack/lib/resp/RespStreamingDecoder';
 
 const noop = () => {};
 

src/standalone/StandaloneCall.ts

@@ -1,4 +1,4 @@
-import {Defer} from 'thingies/es2020/Defer';
+import {Defer} from 'thingies/lib/Defer';
 import type {Cmd, MultiCmd} from '../types';
 
 export const callNoRes = (args: Cmd | MultiCmd) => {

src/standalone/StandaloneClient.ts

@@ -1,12 +1,12 @@
-import {RespEncoder} from 'json-joy/es2020/json-pack/resp/RespEncoder';
-import {RespStreamingDecoder} from 'json-joy/es2020/json-pack/resp/RespStreamingDecoder';
-import {RespPush} from 'json-joy/es2020/json-pack/resp/extensions';
-import {FanOut} from 'thingies/es2020/fanout';
-import {Defer} from 'thingies/es2020/Defer';
+import {RespEncoder} from '@jsonjoy.com/json-pack/lib/resp/RespEncoder';
+import {RespStreamingDecoder} from '@jsonjoy.com/json-pack/lib/resp/RespStreamingDecoder';
+import {RespPush} from '@jsonjoy.com/json-pack/lib/resp/extensions';
+import {FanOut} from 'thingies/lib/fanout';
+import {Defer} from 'thingies/lib/Defer';
 import {StandaloneCall, callNoRes} from './StandaloneCall';
 import {isPushMessage, isMultiCmd, isPushPmessage, isPushSmessage} from '../util/commands';
-import {AvlMap} from 'json-joy/es2020/util/trees/avl/AvlMap';
-import {bufferToUint8Array} from 'json-joy/es2020/util/buffers/bufferToUint8Array';
+import {AvlMap} from 'sonic-forest/lib/avl/AvlMap';
+import {bufferToUint8Array} from '@jsonjoy.com/util/lib/buffers/bufferToUint8Array';
 import {cmpUint8Array, ascii} from '../util/buf';
 import {ReconnectingSocket} from '../util/ReconnectingSocket';
 import {ScriptRegistry} from '../ScriptRegistry';
@@ -431,6 +431,7 @@ export class StandaloneClient {
     if (!fanout) {
       fanout = new FanOut<Uint8Array>();
       this.ssubs.set(channelBuf, fanout);
+      // TODO: create FanOut only after SSUBSCRIBE succeeds
       this.cmdFnF([SSUBSCRIBE, channelBuf]);
     }
     const unsubscribe = fanout.listen(listener);

src/types.ts

@@ -1,5 +1,5 @@
-import type {RespEncoder} from 'json-joy/es2020/json-pack/resp';
-import type {RespStreamingDecoder} from 'json-joy/es2020/json-pack/resp/RespStreamingDecoder';
+import type {RespEncoder} from '@jsonjoy.com/json-pack/lib/resp';
+import type {RespStreamingDecoder} from '@jsonjoy.com/json-pack/lib/resp/RespStreamingDecoder';
 
 export type PartialExcept<T, K extends keyof T> = Partial<T> & Pick<T, K>;
 

src/util/ReconnectingSocket.ts

@@ -1,5 +1,5 @@
 import * as net from 'net';
-import {FanOut} from 'thingies/es2020/fanout';
+import {FanOut} from 'thingies/lib/fanout';
 import type {PartialExcept} from '../types';
 
 export interface ReconnectingSocketOpts {

src/util/buf.ts

@@ -1,4 +1,4 @@
-import {bufferToUint8Array} from 'json-joy/es2020/util/buffers/bufferToUint8Array';
+import {bufferToUint8Array} from '@jsonjoy.com/util/lib/buffers/bufferToUint8Array';
 
 export const cmpUint8Array = (a: Uint8Array, b: Uint8Array): 1 | 0 | -1 => {
   const len1 = a.length;