From 84d6299215fa4f79bb9ddf40848faf5c95d1e6c2 Mon Sep 17 00:00:00 2001 From: Davi de Carvalho Date: Tue, 8 Sep 2026 10:57:58 -0300 Subject: [PATCH 1/3] fix(redis): detect an existing connection by shape instead of instanceof `RedisTransport` decided whether it received a live client or a plain options object with `instanceof Redis || instanceof Cluster`. That check is evaluated against the copy of `ioredis` this package resolved, so a perfectly valid client coming from another copy - a different major, or simply a duplicate in the dependency tree - fails it, and we fall through to `new Redis()`. ioredis ignores the unrecognised properties and connects to 127.0.0.1:6379, silently. `useMessageBuffer` is a second casualty: it only reaches the transport through the third constructor argument, which the options-object branch ignores, so it degrades to `false` and binary payloads are decoded from a lossy string. Both call sites now go through a duck-type check on `duplicate` and `sendCommand`. Both are on the prototype of `Redis` and `Cluster` in every major, and neither name exists in `RedisOptions` or `ClusterOptions`, so an options object can never be misclassified. Values that carry the markers of a client (`status`, `options`, `emit`) without being recognised as one now throw instead of quietly dialing localhost. Co-Authored-By: Claude Opus 5 (1M context) --- src/transports/redis.ts | 69 +++++++- tests/drivers/redis_transport.spec.ts | 223 ++++++++++++++++++++++++++ 2 files changed, 288 insertions(+), 4 deletions(-) diff --git a/src/transports/redis.ts b/src/transports/redis.ts index 0900732..491181a 100644 --- a/src/transports/redis.ts +++ b/src/transports/redis.ts @@ -5,8 +5,10 @@ * @copyright BoringNode */ -import { Redis, Cluster } from 'ioredis' +import { Redis } from 'ioredis' +import type { Cluster } from 'ioredis' import { assert } from '@poppinss/utils/assert' +import { InvalidArgumentsException } from '@poppinss/utils/exception' import debug from '../debug.js' import { JsonEncoder } from '../encoders/json_encoder.js' @@ -24,6 +26,58 @@ export function redis(config: RedisTransportConfig, encoder?: TransportEncoder) return () => new RedisTransport(config, encoder) } +/** + * Detect an existing `ioredis` client by its shape rather than with `instanceof`. + * + * `instanceof` is evaluated against the copy of `ioredis` this package resolved. + * When the host application resolves a different copy - a different major, or + * simply a duplicated one in the dependency tree - the check returns `false` for + * a perfectly valid client and we would fall through to `new Redis(client)`, + * which silently connects to `127.0.0.1:6379`. + * + * `duplicate` and `sendCommand` are on the prototype of both `Redis` and + * `Cluster` in every major, and neither name exists in `RedisOptions` or + * `ClusterOptions`, so an options object can never be mistaken for a client. + */ +function isRedisClient(value: unknown): value is Redis | Cluster { + if (typeof value !== 'object' || value === null) return false + + const candidate = value as Partial + + return typeof candidate.duplicate === 'function' && typeof candidate.sendCommand === 'function' +} + +/** + * `Redis#duplicate` and `Cluster#duplicate` are both parameterless-callable at + * runtime, but TypeScript cannot resolve the call against a `Redis | Cluster` + * union: from ioredis 6 on, both signatures are generic over the reply mapping + * and none of them is compatible with the other. Going through a narrow local + * signature keeps the source compiling against ioredis 5 and 6 alike. + */ +function duplicateClient(client: Redis | Cluster): Redis | Cluster { + return (client.duplicate as unknown as () => Redis | Cluster)() +} + +/** + * Values that look like a client - they carry a `status`, an `options` bag and + * an `emit` method - but that we could not recognize as one. None of those + * names are valid `ioredis` options, so a legitimate configuration object never + * lands here. Rather than quietly building a connection to `127.0.0.1:6379` + * out of it, we fail loudly. + */ +function looksLikeUnsupportedRedisClient(value: unknown): boolean { + if (typeof value !== 'object' || value === null) return false + + const candidate = value as Record + + return ( + typeof candidate.status === 'string' && + typeof candidate.options === 'object' && + candidate.options !== null && + typeof candidate.emit === 'function' + ) +} + export class RedisTransport implements Transport { readonly #publisher: Redis | Cluster readonly #subscriber: Redis | Cluster @@ -51,11 +105,18 @@ export class RedisTransport implements Transport { * If an existing Redis or Cluster instance is passed, we duplicate it * to have separate connections for publisher and subscriber */ - if (options instanceof Redis || options instanceof Cluster) { - this.#publisher = options.duplicate() - this.#subscriber = options.duplicate() + if (isRedisClient(options)) { + this.#publisher = duplicateClient(options) + this.#subscriber = duplicateClient(options) this.#useMessageBuffer = transportOptions?.useMessageBuffer ?? false } else { + if (looksLikeUnsupportedRedisClient(options)) { + throw new InvalidArgumentsException( + 'Cannot use the given Redis connection with "RedisTransport". Expected an "ioredis" ' + + 'client exposing "duplicate()" and "sendCommand()", or a connection options object.' + ) + } + // @ts-expect-error - merged definitions of overloaded constructor is not public this.#publisher = new Redis(options) // @ts-expect-error - merged definitions of overloaded constructor is not public diff --git a/tests/drivers/redis_transport.spec.ts b/tests/drivers/redis_transport.spec.ts index dc69e9a..ddecd26 100644 --- a/tests/drivers/redis_transport.spec.ts +++ b/tests/drivers/redis_transport.spec.ts @@ -13,6 +13,43 @@ import { RedisTransport } from '../../src/transports/redis.js' import { JsonEncoder } from '../../src/encoders/json_encoder.js' import { type TransportEncoder, type TransportMessage } from '../../src/types/main.js' +/** + * A stand-in for a client coming from *another* copy of `ioredis` - a different + * major installed alongside ours, which is the normal outcome of a package + * manager resolving two ranges. + * + * The Proxy sits on a null-prototype target and forwards everything to a real + * client, so it has exactly the shape of a client while failing + * `instanceof Redis` / `instanceof Cluster` against the copy this package + * resolved - which is precisely what a foreign-major client looks like from in + * here. Only one `ioredis` can be installed in this repository, so this is the + * faithful way to express the situation in a test. + */ +function asForeignMajorClient(client: T) { + const duplicates: T[] = [] + + const proxy = new Proxy(Object.create(null), { + get(_target, property) { + const value = (client as any)[property] + + if (property === 'duplicate') { + return (...args: any[]) => { + const duplicated = value.apply(client, args) + duplicates.push(duplicated) + return duplicated + } + } + + return typeof value === 'function' ? value.bind(client) : value + }, + has(_target, property) { + return property in (client as any) + }, + }) as T + + return { client: proxy, duplicates } +} + test.group('Redis Transport', (group) => { let container: StartedRedisContainer @@ -360,4 +397,190 @@ test.group('Redis Transport', (group) => { await setTimeout(200) await transport2.publish('testing-channel', data) }).waitForDone() + + test('should reuse a client coming from another copy of ioredis', async ({ assert, cleanup }) => { + const redisInstance = new Redis({ + host: container.getHost(), + port: container.getMappedPort(6379), + }) + const foreign = asForeignMajorClient(redisInstance) + + cleanup(async () => { + await redisInstance.quit() + }) + + /** + * The premise of the test: same shape, but not an instance of the `ioredis` + * classes this package resolved. + */ + assert.isFalse(foreign.client instanceof Redis) + assert.isFalse(foreign.client instanceof Cluster) + + const transport = new RedisTransport(foreign.client).setId('bus1') + cleanup(() => transport.disconnect()) + + /** + * The connection must have been duplicated (publisher + subscriber) rather + * than handed to `new Redis()` as if it were an options object, which would + * silently dial 127.0.0.1:6379. + */ + assert.lengthOf(foreign.duplicates, 2) + + /** + * And the messages must really land on the configured server, not on the + * default one. + */ + const witness = new Redis({ + host: container.getHost(), + port: container.getMappedPort(6379), + }) + cleanup(async () => { + await witness.quit() + }) + + const received = new Promise((resolve) => { + witness.on('message', (_channel, message) => resolve(message)) + }) + await witness.subscribe('foreign-client-channel') + + await transport.publish('foreign-client-channel', 'test') + + const message = await Promise.race([received, setTimeout(2000).then(() => 'timed-out')]) + + assert.deepEqual(JSON.parse(message), { payload: 'test', busId: 'bus1' }) + }) + + test('should reuse a cluster client coming from another copy of ioredis', async ({ + assert, + cleanup, + }) => { + const cluster = new Cluster([{ host: '127.0.0.1', port: 7000 }], { lazyConnect: true }) + const foreign = asForeignMajorClient(cluster) + + cleanup(async () => { + cluster.disconnect() + }) + + assert.isFalse(foreign.client instanceof Redis) + assert.isFalse(foreign.client instanceof Cluster) + + const transport = new RedisTransport(foreign.client) + cleanup(() => { + for (const duplicated of foreign.duplicates) duplicated.disconnect() + }) + + assert.lengthOf(foreign.duplicates, 2) + assert.instanceOf(foreign.duplicates[0], Cluster) + + transport.onReconnect(() => {}) + assert.equal(foreign.duplicates[1].listenerCount('reconnecting'), 1) + }) + + test('should keep useMessageBuffer when given a client from another copy of ioredis', async ({ + assert, + cleanup, + }, done) => { + assert.plan(4) + + class BinaryEncoder implements TransportEncoder { + encode(message: TransportMessage) { + return Buffer.from(JSON.stringify(message)) + } + + decode(data: string | Buffer) { + const buffer = Buffer.isBuffer(data) ? data : Buffer.from(data, 'binary') + return JSON.parse(buffer.toString()) + } + } + + const redisInstance = new Redis({ + host: container.getHost(), + port: container.getMappedPort(6379), + }) + + cleanup(async () => { + await redisInstance.quit() + }) + + const foreign1 = asForeignMajorClient(redisInstance) + const foreign2 = asForeignMajorClient(redisInstance) + + const transport1 = new RedisTransport(foreign1.client, new BinaryEncoder(), { + useMessageBuffer: true, + }).setId('bus1') + + const transport2 = new RedisTransport(foreign2.client, new BinaryEncoder(), { + useMessageBuffer: true, + }).setId('bus2') + + cleanup(async () => { + await transport1.disconnect() + await transport2.disconnect() + }) + + /** + * `useMessageBuffer` only reaches the transport through the third argument, + * which is ignored on the options-object branch. If the client is not + * recognized, it silently degrades to `false` and the subscriber listens on + * "message" instead of "messageBuffer" - decoding binary payloads through a + * lossy string. + */ + assert.lengthOf(foreign1.duplicates, 2) + assert.equal(foreign1.duplicates[1].listenerCount('messageBuffer'), 1) + assert.equal(foreign1.duplicates[1].listenerCount('message'), 0) + + const data = ['foo', '👍'] + + await transport1.subscribe('testing-channel', (payload) => { + assert.deepEqual(payload, data) + done() + }) + + await setTimeout(200) + await transport2.publish('testing-channel', data) + }).waitForDone() + + test('should throw instead of dialing localhost for an unusable client', async ({ assert }) => { + const unusable = { + status: 'ready', + options: { host: 'redis.internal', port: 6380 }, + emit: () => true, + } + + assert.throws( + // @ts-expect-error - deliberately not a valid argument + () => new RedisTransport(unusable), + /Cannot use the given Redis connection with "RedisTransport"/ + ) + }) + + test('should still build a new connection from a plain options object', async ({ + assert, + cleanup, + }, done) => { + assert.plan(1) + + const transport1 = new RedisTransport({ + host: container.getHost(), + port: container.getMappedPort(6379), + }).setId('bus1') + + const transport2 = new RedisTransport({ + host: container.getHost(), + port: container.getMappedPort(6379), + }).setId('bus2') + + cleanup(async () => { + await transport1.disconnect() + await transport2.disconnect() + }) + + await transport1.subscribe('options-object-channel', (payload) => { + assert.equal(payload, 'test') + done() + }) + + await setTimeout(200) + await transport2.publish('options-object-channel', 'test') + }).waitForDone() }) From 71946390f8a925f69b37189ff91ee7232fbac45c Mon Sep 17 00:00:00 2001 From: Davi de Carvalho Date: Tue, 8 Sep 2026 10:57:58 -0300 Subject: [PATCH 2/3] feat(redis): accept ioredis 6 as a peer dependency The whole `RedisTransport` surface was validated against ioredis 6.0.0 with RESP3 actually negotiated (`options.protocol === 3`, server-side `HELLO` agrees): publish/subscribe/unsubscribe, self-message filtering, the `messageBuffer` binary path (byte-identical round trip), the `duplicate()` path for an existing client, reconnection, and a real three-node Redis cluster. `Redis#duplicate` and `Cluster#duplicate` became generic over the reply mapping in ioredis 6, which made the call against a `Redis | Cluster` union uncompilable (TS2349); it now goes through a narrow local signature so the source typechecks and builds against both majors. Co-Authored-By: Claude Opus 5 (1M context) --- package.json | 2 +- yarn.lock | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/package.json b/package.json index ba67d3f..0d1fa77 100644 --- a/package.json +++ b/package.json @@ -62,7 +62,7 @@ "object-hash": "^3.0.0" }, "peerDependencies": { - "ioredis": "^5.0.0" + "ioredis": "^5.0.0 || ^6.0.0" }, "peerDependenciesMeta": { "ioredis": { diff --git a/yarn.lock b/yarn.lock index deebef1..bfcec41 100644 --- a/yarn.lock +++ b/yarn.lock @@ -130,7 +130,7 @@ __metadata: tsup: "npm:^8.5.1" typescript: "npm:^5.9.3" peerDependencies: - ioredis: ^5.0.0 + ioredis: ^5.0.0 || ^6.0.0 peerDependenciesMeta: ioredis: optional: true From 6c23ca0e7e2d02e2829577584a87f7e4a0b3289c Mon Sep 17 00:00:00 2001 From: Davi de Carvalho Date: Tue, 8 Sep 2026 11:32:24 -0300 Subject: [PATCH 3/3] test(redis): exercise the cross-major path with a real ioredis 6 client The foreign-major client was simulated with a `Proxy` over a null-prototype target, which is faithful in shape but leaves open the fair question of whether it behaves like a real client of another major. A second, genuine copy of `ioredis` is now installed side by side through the npm alias `"ioredis-v6": "npm:ioredis@^6.0.0"`, so the three cross-major tests hand `RedisTransport` an actual ioredis 6 client - the exact shape of an application on `@adonisjs/redis@11` passing its connection to a bus resolved against ioredis 5. Each one asserts the premise at runtime (`instanceof ForeignRedis`, not `instanceof Redis`, and the two constructors are not the same object) before asserting the behaviour, so the tests cannot silently stop covering the bug. Delivery is still proven against the configured server through a witness connection bound to the testcontainers port, and the `useMessageBuffer` case now compares the bytes seen on the subscriber with the bytes the publisher wrote, on top of the `messageBuffer`/`message` listener counts. The `Proxy` helper is dropped: a real foreign client is a strictly harder case than a duplicated copy of the same major, and the "client-shaped object that is no ioredis at all" tripwire is already covered by its own test with a plain object literal. Reverting `src/transports/redis.ts` to the `instanceof` version turns the suite red on exactly these tests: 13 passed / 4 failed against 17 passed on the branch. Co-Authored-By: Claude Opus 5 (1M context) --- package.json | 1 + tests/drivers/redis_transport.spec.ts | 166 ++++++++++++++++---------- yarn.lock | 22 ++++ 3 files changed, 124 insertions(+), 65 deletions(-) diff --git a/package.json b/package.json index 0d1fa77..a638f15 100644 --- a/package.json +++ b/package.json @@ -49,6 +49,7 @@ "del-cli": "^7.0.0", "eslint": "^10.6.0", "ioredis": "^5.11.1", + "ioredis-v6": "npm:ioredis@^6.0.0", "mqtt": "^5.15.2", "prettier": "^3.9.4", "release-it": "^20.2.1", diff --git a/tests/drivers/redis_transport.spec.ts b/tests/drivers/redis_transport.spec.ts index ddecd26..90add29 100644 --- a/tests/drivers/redis_transport.spec.ts +++ b/tests/drivers/redis_transport.spec.ts @@ -8,46 +8,50 @@ import { setImmediate, setTimeout } from 'node:timers/promises' import { test } from '@japa/runner' import { Redis, Cluster } from 'ioredis' +import { Redis as ForeignRedis, Cluster as ForeignCluster } from 'ioredis-v6' import { RedisContainer, type StartedRedisContainer } from '@testcontainers/redis' import { RedisTransport } from '../../src/transports/redis.js' import { JsonEncoder } from '../../src/encoders/json_encoder.js' import { type TransportEncoder, type TransportMessage } from '../../src/types/main.js' /** - * A stand-in for a client coming from *another* copy of `ioredis` - a different - * major installed alongside ours, which is the normal outcome of a package - * manager resolving two ranges. + * `ioredis-v6` is a second, real copy of `ioredis` installed side by side with + * the one this package resolves, through the npm alias + * `"ioredis-v6": "npm:ioredis@^6.0.0"`. A client built from it is exactly what + * an application running `@adonisjs/redis@11` hands to a bus resolved against + * `ioredis` 5: same API, different classes, so `instanceof` says `false`. * - * The Proxy sits on a null-prototype target and forwards everything to a real - * client, so it has exactly the shape of a client while failing - * `instanceof Redis` / `instanceof Cluster` against the copy this package - * resolved - which is precisely what a foreign-major client looks like from in - * here. Only one `ioredis` can be installed in this repository, so this is the - * faithful way to express the situation in a test. + * The cast is part of the scenario too - the two copies ship two unrelated sets + * of types - but only the runtime behaviour is under test here. */ -function asForeignMajorClient(client: T) { - const duplicates: T[] = [] - - const proxy = new Proxy(Object.create(null), { - get(_target, property) { - const value = (client as any)[property] - - if (property === 'duplicate') { - return (...args: any[]) => { - const duplicated = value.apply(client, args) - duplicates.push(duplicated) - return duplicated - } - } +function asForeignClient(client: ForeignRedis): Redis +function asForeignClient(client: ForeignCluster): Cluster +function asForeignClient(client: ForeignRedis | ForeignCluster) { + return client as unknown as Redis | Cluster +} - return typeof value === 'function' ? value.bind(client) : value - }, - has(_target, property) { - return property in (client as any) +/** + * Records what `duplicate()` returns, so a test can tell "the transport reused + * the connection" from "the transport built a brand new one". + * + * The spy is an own property shadowing the prototype method, so the client + * remains an instance of its own `ioredis` copy - which is the very thing these + * tests are about. + */ +function spyOnDuplicate(client: T): T[] { + const duplicates: T[] = [] + const duplicate = (client.duplicate as unknown as () => T).bind(client) + + Object.defineProperty(client, 'duplicate', { + configurable: true, + value: () => { + const duplicated = duplicate() + duplicates.push(duplicated) + return duplicated }, - }) as T + }) - return { client: proxy, duplicates } + return duplicates } test.group('Redis Transport', (group) => { @@ -399,24 +403,27 @@ test.group('Redis Transport', (group) => { }).waitForDone() test('should reuse a client coming from another copy of ioredis', async ({ assert, cleanup }) => { - const redisInstance = new Redis({ + const foreignClient = new ForeignRedis({ host: container.getHost(), port: container.getMappedPort(6379), }) - const foreign = asForeignMajorClient(redisInstance) cleanup(async () => { - await redisInstance.quit() + await foreignClient.quit() }) /** - * The premise of the test: same shape, but not an instance of the `ioredis` + * The premise of the test: a genuine `ioredis` client, built from a genuine + * second copy of the package, which is therefore not an instance of the * classes this package resolved. */ - assert.isFalse(foreign.client instanceof Redis) - assert.isFalse(foreign.client instanceof Cluster) + assert.isFalse((ForeignRedis as unknown) === (Redis as unknown)) + assert.instanceOf(foreignClient, ForeignRedis) + assert.isFalse(foreignClient instanceof Redis) + assert.isFalse(foreignClient instanceof Cluster) - const transport = new RedisTransport(foreign.client).setId('bus1') + const duplicates = spyOnDuplicate(foreignClient) + const transport = new RedisTransport(asForeignClient(foreignClient)).setId('bus1') cleanup(() => transport.disconnect()) /** @@ -424,11 +431,13 @@ test.group('Redis Transport', (group) => { * than handed to `new Redis()` as if it were an options object, which would * silently dial 127.0.0.1:6379. */ - assert.lengthOf(foreign.duplicates, 2) + assert.lengthOf(duplicates, 2) + assert.instanceOf(duplicates[0], ForeignRedis) /** * And the messages must really land on the configured server, not on the - * default one. + * default one. The witness connects to the container explicitly, so it only + * ever sees what was published there. */ const witness = new Redis({ host: container.getHost(), @@ -454,34 +463,37 @@ test.group('Redis Transport', (group) => { assert, cleanup, }) => { - const cluster = new Cluster([{ host: '127.0.0.1', port: 7000 }], { lazyConnect: true }) - const foreign = asForeignMajorClient(cluster) + const foreignCluster = new ForeignCluster([{ host: '127.0.0.1', port: 7000 }], { + lazyConnect: true, + }) - cleanup(async () => { - cluster.disconnect() + cleanup(() => { + foreignCluster.disconnect() }) - assert.isFalse(foreign.client instanceof Redis) - assert.isFalse(foreign.client instanceof Cluster) + assert.isFalse((ForeignCluster as unknown) === (Cluster as unknown)) + assert.instanceOf(foreignCluster, ForeignCluster) + assert.isFalse(foreignCluster instanceof Redis) + assert.isFalse(foreignCluster instanceof Cluster) - const transport = new RedisTransport(foreign.client) + const duplicates = spyOnDuplicate(foreignCluster) + const transport = new RedisTransport(asForeignClient(foreignCluster)) cleanup(() => { - for (const duplicated of foreign.duplicates) duplicated.disconnect() + for (const duplicated of duplicates) duplicated.disconnect() }) - assert.lengthOf(foreign.duplicates, 2) - assert.instanceOf(foreign.duplicates[0], Cluster) + assert.lengthOf(duplicates, 2) + assert.instanceOf(duplicates[0], ForeignCluster) + assert.isFalse(duplicates[0] instanceof Cluster) transport.onReconnect(() => {}) - assert.equal(foreign.duplicates[1].listenerCount('reconnecting'), 1) + assert.equal(duplicates[1].listenerCount('reconnecting'), 1) }) test('should keep useMessageBuffer when given a client from another copy of ioredis', async ({ assert, cleanup, - }, done) => { - assert.plan(4) - + }) => { class BinaryEncoder implements TransportEncoder { encode(message: TransportMessage) { return Buffer.from(JSON.stringify(message)) @@ -493,23 +505,31 @@ test.group('Redis Transport', (group) => { } } - const redisInstance = new Redis({ + const subscriberClient = new ForeignRedis({ + host: container.getHost(), + port: container.getMappedPort(6379), + }) + + const publisherClient = new ForeignRedis({ host: container.getHost(), port: container.getMappedPort(6379), }) cleanup(async () => { - await redisInstance.quit() + await subscriberClient.quit() + await publisherClient.quit() }) - const foreign1 = asForeignMajorClient(redisInstance) - const foreign2 = asForeignMajorClient(redisInstance) + assert.instanceOf(subscriberClient, ForeignRedis) + assert.isFalse(subscriberClient instanceof Redis) - const transport1 = new RedisTransport(foreign1.client, new BinaryEncoder(), { + const duplicates = spyOnDuplicate(subscriberClient) + + const transport1 = new RedisTransport(asForeignClient(subscriberClient), new BinaryEncoder(), { useMessageBuffer: true, }).setId('bus1') - const transport2 = new RedisTransport(foreign2.client, new BinaryEncoder(), { + const transport2 = new RedisTransport(asForeignClient(publisherClient), new BinaryEncoder(), { useMessageBuffer: true, }).setId('bus2') @@ -525,20 +545,36 @@ test.group('Redis Transport', (group) => { * "message" instead of "messageBuffer" - decoding binary payloads through a * lossy string. */ - assert.lengthOf(foreign1.duplicates, 2) - assert.equal(foreign1.duplicates[1].listenerCount('messageBuffer'), 1) - assert.equal(foreign1.duplicates[1].listenerCount('message'), 0) + assert.lengthOf(duplicates, 2) + assert.equal(duplicates[1].listenerCount('messageBuffer'), 1) + assert.equal(duplicates[1].listenerCount('message'), 0) const data = ['foo', '👍'] - await transport1.subscribe('testing-channel', (payload) => { - assert.deepEqual(payload, data) - done() + /** + * The exact bytes the publisher writes. Listening alongside the transport on + * its own subscriber lets us assert the payload survived the trip untouched, + * rather than through a UTF-8 round trip that happens to be lossless. + */ + const expectedBytes = new BinaryEncoder().encode({ payload: data, busId: 'bus2' }) + let receivedBytes: Buffer | undefined + duplicates[1].on('messageBuffer', (_channel: Buffer, message: Buffer) => { + receivedBytes = message }) + let resolvePayload!: (payload: unknown) => void + const payloadReceived = new Promise((resolve) => (resolvePayload = resolve)) + await transport1.subscribe('testing-channel', resolvePayload) + await setTimeout(200) await transport2.publish('testing-channel', data) - }).waitForDone() + + const payload = await Promise.race([payloadReceived, setTimeout(2000).then(() => 'timed-out')]) + + assert.deepEqual(payload, data) + assert.isTrue(Buffer.isBuffer(receivedBytes)) + assert.isTrue(receivedBytes!.equals(expectedBytes as Buffer)) + }) test('should throw instead of dialing localhost for an unusable client', async ({ assert }) => { const unusable = { diff --git a/yarn.lock b/yarn.lock index bfcec41..d3013b8 100644 --- a/yarn.lock +++ b/yarn.lock @@ -122,6 +122,7 @@ __metadata: del-cli: "npm:^7.0.0" eslint: "npm:^10.6.0" ioredis: "npm:^5.11.1" + ioredis-v6: "npm:ioredis@^6.0.0" mqtt: "npm:^5.15.2" object-hash: "npm:^3.0.0" prettier: "npm:^3.9.4" @@ -704,6 +705,13 @@ __metadata: languageName: node linkType: hard +"@ioredis/commands@npm:2.0.0": + version: 2.0.0 + resolution: "@ioredis/commands@npm:2.0.0" + checksum: 10c0/2fb5edb9782790c24a375cc78bd69d45ac6c268d7d7ed8e534f2d4ecc28707bde9882a04eecd68be369ae784801bf93fa267b9a76f86418b252a08f341934e35 + languageName: node + linkType: hard + "@isaacs/cliui@npm:^8.0.2": version: 8.0.2 resolution: "@isaacs/cliui@npm:8.0.2" @@ -3922,6 +3930,20 @@ __metadata: languageName: node linkType: hard +"ioredis-v6@npm:ioredis@^6.0.0": + version: 6.0.0 + resolution: "ioredis@npm:6.0.0" + dependencies: + "@ioredis/commands": "npm:2.0.0" + cluster-key-slot: "npm:1.1.1" + debug: "npm:4.4.3" + denque: "npm:2.1.0" + redis-errors: "npm:1.2.0" + standard-as-callback: "npm:2.1.0" + checksum: 10c0/6f74e1d298440c0d814e6bf9b3acdb0991d4c1862d3fa64c3828668bcc306823f7845710abdc5eab7986a35d86ec8c86c4bb9c9c7d727636d385ecb2cd40abd7 + languageName: node + linkType: hard + "ioredis@npm:^5.11.1": version: 5.11.1 resolution: "ioredis@npm:5.11.1"