You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/08/10 14:47:40 UTC
(iggy) branch master updated: feat(node): drop classic framing and test only against the VSR server (#3843)
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new efcb88071 feat(node): drop classic framing and test only against the VSR server (#3843)
efcb88071 is described below
commit efcb880719aa799cd14ee8025012c6f2ccdf79e6
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Aug 10 16:47:36 2026 +0200
feat(node): drop classic framing and test only against the VSR server (#3843)
---
.github/actions/node-npm/pre-merge/action.yml | 100 ++++--------------
.github/config/components.yml | 4 +-
.github/workflows/_test.yml | 3 +-
.github/workflows/_test_bdd.yml | 14 +--
.github/workflows/_test_examples.yml | 10 +-
.github/workflows/coverage-baseline.yml | 11 +-
codecov.yml | 3 -
examples/node/src/tcp-tls/consumer.ts | 5 +-
examples/node/src/tcp-tls/producer.ts | 20 +++-
foreign/node/CHANGELOG.md | 95 -----------------
foreign/node/README.md | 25 +++--
foreign/node/docker-compose.yml | 9 +-
foreign/node/package-lock.json | 4 +-
foreign/node/package.json | 3 +-
foreign/node/src/bdd/auth.ts | 3 -
foreign/node/src/client/client.config.test.ts | 31 +-----
foreign/node/src/client/client.config.ts | 17 +---
foreign/node/src/client/client.connection.test.ts | 37 +++----
foreign/node/src/client/client.connection.ts | 17 +---
foreign/node/src/client/client.frame.test.ts | 113 ++++++++-------------
foreign/node/src/client/client.frame.ts | 40 ++------
foreign/node/src/client/client.socket.test.ts | 1 -
foreign/node/src/client/client.socket.ts | 29 +-----
foreign/node/src/client/client.type.ts | 7 --
foreign/node/src/client/client.utils.test.ts | 31 ++----
foreign/node/src/client/client.utils.ts | 68 -------------
foreign/node/src/e2e/tcp.cluster.e2e.ts | 21 +---
foreign/node/src/e2e/tcp.consumer-group.e2e.ts | 6 +-
foreign/node/src/e2e/tcp.consumer-stream.e2e.ts | 6 +-
foreign/node/src/e2e/tcp.send-message.e2e.ts | 12 ---
foreign/node/src/e2e/test-client.utils.ts | 2 -
foreign/node/src/e2e/tls.system.e2e.ts | 8 +-
foreign/node/src/wire/command-set.test.ts | 1 -
.../node/src/wire/message/poll-messages.command.ts | 3 +-
foreign/node/src/wire/vsr/namespace.ts | 2 +
scripts/run-examples-from-readme.sh | 5 +-
36 files changed, 198 insertions(+), 568 deletions(-)
diff --git a/.github/actions/node-npm/pre-merge/action.yml b/.github/actions/node-npm/pre-merge/action.yml
index 5dba122ba..d799467c1 100644
--- a/.github/actions/node-npm/pre-merge/action.yml
+++ b/.github/actions/node-npm/pre-merge/action.yml
@@ -20,7 +20,7 @@ description: Node.js pre-merge testing github iggy actions
inputs:
task:
- description: "Task to run (lint, test, build, e2e, e2e-vsr)"
+ description: "Task to run (lint, test, build, e2e)"
required: true
runs:
@@ -39,11 +39,11 @@ runs:
shell: bash
- name: Setup Rust with cache
- if: inputs.task == 'e2e' || inputs.task == 'e2e-vsr'
+ if: inputs.task == 'e2e'
uses: ./.github/actions/utils/setup-rust-with-cache
- name: Install netcat
- if: inputs.task == 'e2e' || inputs.task == 'e2e-vsr'
+ if: inputs.task == 'e2e'
run: sudo apt-get update && sudo apt-get install -y netcat-openbsd
shell: bash
@@ -90,17 +90,16 @@ runs:
npx c8 --reporter=lcov --reports-dir=../../reports/node-coverage/unit npm run test:unit
shell: bash
+ # TODO(hubcio): change to iggy-server once legacy server is removed
+ # (core/server has VSR support)
- name: Start Iggy server
id: iggy
if: inputs.task == 'e2e'
uses: ./.github/actions/utils/server-start
with:
- # Node e2e asserts full cluster metadata (2 nodes from
- # config.toml's [[cluster.nodes]] list), so cluster mode must be
- # on. --replica-id picks the current node out of that list.
- replica-id: "0"
- env:
- IGGY_CLUSTER_ENABLED: "true"
+ cargo-bin: iggy-server-ng
+ cargo-features: vsr
+ wait-timeout-seconds: "90"
- name: E2E tests
if: inputs.task == 'e2e'
@@ -108,91 +107,32 @@ runs:
cd foreign/node
mkdir -p ../../reports/node-coverage/e2e
npx c8 --reporter=lcov --reports-dir=../../reports/node-coverage/e2e npm run test:e2e
- env:
- IGGY_SERVER_HOST: 127.0.0.1
- IGGY_SERVER_TCP_PORT: 8090
shell: bash
- - name: Start Iggy VSR server 0
- id: iggy-vsr-0
- if: inputs.task == 'e2e-vsr'
- uses: ./.github/actions/utils/server-start
- with:
- cargo-bin: iggy-server-ng
- cargo-features: vsr
- replica-id: "0"
- pid-file: ${{ runner.temp }}/iggy-node-vsr-0.pid
- log-file: ${{ runner.temp }}/iggy-node-vsr-0.log
- wait-timeout-seconds: "90"
- env:
- IGGY_CLUSTER_ENABLED: "true"
- IGGY_SYSTEM_PATH: ${{ runner.temp }}/iggy-node-vsr-0-data
-
- - name: Start Iggy VSR server 1
- id: iggy-vsr-1
- if: inputs.task == 'e2e-vsr'
- uses: ./.github/actions/utils/server-start
- with:
- cargo-bin: iggy-server-ng
- cargo-features: vsr
- replica-id: "1"
- tcp_address: 127.0.0.1:8091
- http_address: 127.0.0.1:3001
- pid-file: ${{ runner.temp }}/iggy-node-vsr-1.pid
- log-file: ${{ runner.temp }}/iggy-node-vsr-1.log
- wait-timeout-seconds: "90"
- env:
- IGGY_CLUSTER_ENABLED: "true"
- IGGY_SYSTEM_PATH: ${{ runner.temp }}/iggy-node-vsr-1-data
-
- - name: VSR E2E tests
- if: inputs.task == 'e2e-vsr'
- run: |
- cd foreign/node
- mkdir -p ../../reports/node-coverage/e2e-vsr
- npx c8 --reporter=lcov \
- --reports-dir=../../reports/node-coverage/e2e-vsr \
- npm run test:e2e:vsr
- env:
- IGGY_TEST_PROTOCOL: vsr
- shell: bash
-
- - name: Stop Iggy VSR server 1
- if: always() && inputs.task == 'e2e-vsr'
- uses: ./.github/actions/utils/server-stop
- with:
- pid-file: ${{ steps.iggy-vsr-1.outputs.pid_file }}
- log-file: ${{ steps.iggy-vsr-1.outputs.log_file }}
-
- - name: Stop Iggy VSR server 0
- if: always() && inputs.task == 'e2e-vsr'
+ - name: Stop Iggy server
+ if: always() && inputs.task == 'e2e'
uses: ./.github/actions/utils/server-stop
with:
- pid-file: ${{ steps.iggy-vsr-0.outputs.pid_file }}
- log-file: ${{ steps.iggy-vsr-0.outputs.log_file }}
+ pid-file: ${{ steps.iggy.outputs.pid_file }}
+ log-file: ${{ steps.iggy.outputs.log_file }}
- - name: Upload VSR server logs
- if: always() && inputs.task == 'e2e-vsr'
+ - name: Upload server logs
+ if: always() && inputs.task == 'e2e'
uses: actions/upload-artifact@v7
with:
- name: iggy-node-vsr-server-logs
- path: |
- ${{ steps.iggy-vsr-0.outputs.log_file }}
- ${{ steps.iggy-vsr-1.outputs.log_file }}
+ name: iggy-node-server-logs
+ path: ${{ steps.iggy.outputs.log_file }}
if-no-files-found: ignore
- - name: Stop Iggy server (plain)
- if: always() && inputs.task == 'e2e'
- uses: ./.github/actions/utils/server-stop
- with:
- pid-file: ${{ steps.iggy.outputs.pid_file }}
- log-file: ${{ steps.iggy.outputs.log_file }}
-
+ # TODO(hubcio): change to iggy-server once legacy server is removed
+ # (core/server has VSR support)
- name: Start Iggy server (TLS)
id: iggy-tls
if: inputs.task == 'e2e'
uses: ./.github/actions/utils/server-start
with:
+ cargo-bin: iggy-server-ng
+ cargo-features: vsr
pid-file: ${{ runner.temp }}/iggy-server-tls.pid
log-file: ${{ runner.temp }}/iggy-server-tls.log
env:
diff --git a/.github/config/components.yml b/.github/config/components.yml
index 0f140ad71..9dc7f67ad 100644
--- a/.github/config/components.yml
+++ b/.github/config/components.yml
@@ -250,7 +250,7 @@ components:
- "ci-infrastructure" # CI changes trigger full regression
paths:
- "foreign/node/**"
- tasks: ["lint", "test", "build", "e2e", "e2e-vsr"]
+ tasks: ["lint", "test", "build", "e2e"]
sdk-go:
depends_on:
@@ -370,7 +370,7 @@ components:
paths:
- "bdd/node/**"
- "bdd/scenarios/**"
- tasks: ["bdd-node", "bdd-node-vsr"]
+ tasks: ["bdd-node"]
bdd-csharp:
depends_on:
diff --git a/.github/workflows/_test.yml b/.github/workflows/_test.yml
index 80edc184b..a2b6a61db 100644
--- a/.github/workflows/_test.yml
+++ b/.github/workflows/_test.yml
@@ -127,8 +127,7 @@ jobs:
if: >-
inputs.component == 'sdk-node' &&
(inputs.task == 'test' ||
- inputs.task == 'e2e' ||
- inputs.task == 'e2e-vsr')
+ inputs.task == 'e2e')
uses: codecov/codecov-action@v7.0.0
with:
token: ${{ secrets.CODECOV_TOKEN }}
diff --git a/.github/workflows/_test_bdd.yml b/.github/workflows/_test_bdd.yml
index 1b1fbf732..fbf64d60e 100644
--- a/.github/workflows/_test_bdd.yml
+++ b/.github/workflows/_test_bdd.yml
@@ -53,11 +53,11 @@ jobs:
run: |
# The VSR lanes need the vsr feature on both the server and the CLI,
# otherwise the CLI cannot frame requests for the VSR wire protocol
- # and the healthcheck ping fails. The Go SDK speaks only VSR, while
- # the Python wheels, Java SDK, .NET SDK, and C++ bindings are
- # vsr-built, so those suites are always on this branch.
+ # and the healthcheck ping fails. The Go SDK and the Node SDK speak
+ # only VSR, while the Python wheels, Java SDK, .NET SDK, and C++
+ # bindings are vsr-built, so those suites are always on this branch.
case "${{ inputs.task }}" in
- bdd-rust-vsr|bdd-go|bdd-go-race|bdd-python|bdd-node-vsr|bdd-csharp|bdd-java|bdd-cpp)
+ bdd-rust-vsr|bdd-go|bdd-go-race|bdd-python|bdd-node|bdd-csharp|bdd-java|bdd-cpp)
# TODO: change to iggy-server once legacy server is removed (core/server has VSR support)
SERVER_BIN="iggy-server-ng"
echo "Building the VSR server binary and CLI (--features vsr) for BDD tests..."
@@ -91,12 +91,12 @@ jobs:
if: startsWith(inputs.component, 'bdd-') && startsWith(inputs.task, 'bdd-')
run: |
# Extract SDK name from task (format: bdd-<sdk>, or bdd-<sdk>-vsr
- # for an explicit vsr lane). Python, C#, Java, and C++ have no
- # legacy lane, so their plain task names run vsr.
+ # for an explicit vsr lane). Python, Node, C#, Java, and C++ have
+ # no legacy lane, so their plain task names run vsr.
SDK_NAME=$(echo "${{ inputs.task }}" | sed 's/^bdd-//; s/-vsr$//')
EXTRA_FLAGS=()
case "${{ inputs.task }}" in
- bdd-rust-vsr|bdd-python|bdd-node-vsr|bdd-csharp|bdd-cpp)
+ bdd-rust-vsr|bdd-python|bdd-node|bdd-csharp|bdd-cpp)
EXTRA_FLAGS+=(--vsr)
# TODO: change to iggy-server once legacy server is removed (core/server has VSR support)
export IGGY_SERVER_NG_PATH="target/debug/iggy-server-ng"
diff --git a/.github/workflows/_test_examples.yml b/.github/workflows/_test_examples.yml
index 9d3e3de9d..8a6de9d95 100644
--- a/.github/workflows/_test_examples.yml
+++ b/.github/workflows/_test_examples.yml
@@ -113,11 +113,11 @@ jobs:
echo "Building common binaries for all examples tests..."
echo "Current directory: $(pwd)"
- # The Go SDK speaks only the VSR wire protocol, while the Python
- # wheels, Java SDK, and .NET SDK are vsr-built, so those lanes need
- # the VSR server. Every other language still runs against the legacy
- # one until its own migration lands.
- if [[ "${{ inputs.task }}" == "examples-go" || "${{ inputs.task }}" == "examples-python" || "${{ inputs.task }}" == "examples-csharp" || "${{ inputs.task }}" == "examples-java" ]]; then
+ # The Go SDK and the Node SDK speak only the VSR wire protocol,
+ # while the Python wheels, Java SDK, and .NET SDK are vsr-built, so
+ # those lanes need the VSR server. Every other language still runs
+ # against the legacy one until its own migration lands.
+ if [[ "${{ inputs.task }}" == "examples-go" || "${{ inputs.task }}" == "examples-python" || "${{ inputs.task }}" == "examples-csharp" || "${{ inputs.task }}" == "examples-java" || "${{ inputs.task }}" == "examples-node" ]]; then
# TODO: change to iggy-server once legacy server is removed (core/server has VSR support)
SERVER_BIN="iggy-server-ng"
echo "Building ${SERVER_BIN} (--features vsr)..."
diff --git a/.github/workflows/coverage-baseline.yml b/.github/workflows/coverage-baseline.yml
index 268658f3e..b2d64d122 100644
--- a/.github/workflows/coverage-baseline.yml
+++ b/.github/workflows/coverage-baseline.yml
@@ -447,11 +447,15 @@ jobs:
cd foreign/node
npm ci --ignore-scripts
+ # TODO(hubcio): change to iggy-server once legacy server is removed
+ # (core/server has VSR support)
- name: Start Iggy server
id: iggy
uses: ./.github/actions/utils/server-start
- env:
- IGGY_CLUSTER_ENABLED: true
+ with:
+ cargo-bin: iggy-server-ng
+ cargo-features: vsr
+ wait-timeout-seconds: "90"
- name: Run unit tests with coverage
run: |
@@ -464,9 +468,6 @@ jobs:
cd foreign/node
mkdir -p ../../reports/node-coverage/e2e
npx c8 --reporter=lcov --reports-dir=../../reports/node-coverage/e2e npm run test:e2e
- env:
- IGGY_SERVER_HOST: 127.0.0.1
- IGGY_SERVER_TCP_PORT: 8090
- name: Stop Iggy server
if: always()
diff --git a/codecov.yml b/codecov.yml
index f16b83865..7c71b59c9 100644
--- a/codecov.yml
+++ b/codecov.yml
@@ -68,9 +68,6 @@ flag_management:
- name: node-e2e
paths:
- foreign/node/
- - name: node-e2e-vsr
- paths:
- - foreign/node/
- name: go
paths:
- foreign/go/
diff --git a/examples/node/src/tcp-tls/consumer.ts b/examples/node/src/tcp-tls/consumer.ts
index 9a7ec5229..82a4c7a33 100644
--- a/examples/node/src/tcp-tls/consumer.ts
+++ b/examples/node/src/tcp-tls/consumer.ts
@@ -23,10 +23,13 @@
//
// Prerequisites:
// Start the Iggy server with TLS enabled:
+// TODO(hubcio): change to iggy-server once legacy server is removed
+// (core/server has VSR support)
+// IGGY_ROOT_USERNAME=iggy IGGY_ROOT_PASSWORD=iggy \
// IGGY_TCP_TLS_ENABLED=true \
// IGGY_TCP_TLS_CERT_FILE=core/certs/iggy_cert.pem \
// IGGY_TCP_TLS_KEY_FILE=core/certs/iggy_key.pem \
-// cargo r --bin iggy-server
+// cargo r --bin iggy-server-ng --features vsr
//
// Run this example (from examples/node/):
// DEBUG=iggy:* npx tsx src/tcp-tls/consumer.ts
diff --git a/examples/node/src/tcp-tls/producer.ts b/examples/node/src/tcp-tls/producer.ts
index 7b51cc541..d0184e0fa 100644
--- a/examples/node/src/tcp-tls/producer.ts
+++ b/examples/node/src/tcp-tls/producer.ts
@@ -23,17 +23,28 @@
//
// Prerequisites:
// Start the Iggy server with TLS enabled:
+// TODO(hubcio): change to iggy-server once legacy server is removed
+// (core/server has VSR support)
+// IGGY_ROOT_USERNAME=iggy IGGY_ROOT_PASSWORD=iggy \
// IGGY_TCP_TLS_ENABLED=true \
// IGGY_TCP_TLS_CERT_FILE=core/certs/iggy_cert.pem \
// IGGY_TCP_TLS_KEY_FILE=core/certs/iggy_key.pem \
-// cargo r --bin iggy-server
+// cargo r --bin iggy-server-ng --features vsr
//
// Run this example (from examples/node/):
// DEBUG=iggy:* npx tsx src/tcp-tls/producer.ts
import { readFileSync } from 'node:fs';
import { Client, Partitioning } from 'apache-iggy';
-import { BATCHES_LIMIT, cleanup, initSystem, log, MESSAGES_PER_BATCH, sleep } from '../utils';
+import {
+ BATCHES_LIMIT,
+ cleanup,
+ initSystem,
+ log,
+ MESSAGES_PER_BATCH,
+ PARTITION_ID,
+ sleep
+} from '../utils';
async function produceMessages(
client: Client,
@@ -62,11 +73,14 @@ async function produceMessages(
});
try {
+ // The VSR client routes each send to an explicit partition.
+ // TODO(hubcio): Balanced partitioning to be implemented; not decided
+ // yet whether it'll be on server side or client side.
await client.message.send({
streamId,
topicId,
messages,
- partition: Partitioning.Balanced,
+ partition: Partitioning.PartitionId(PARTITION_ID),
});
} catch (error) {
log('Error sending messages: %o', error);
diff --git a/foreign/node/CHANGELOG.md b/foreign/node/CHANGELOG.md
deleted file mode 100644
index 732a4f587..000000000
--- a/foreign/node/CHANGELOG.md
+++ /dev/null
@@ -1,95 +0,0 @@
-# Changelog
-
-## [1.0.6](https://github.com/iggy-rs/iggy-node-client/compare/v1.0.5...v1.0.6) (2025-01-14)
-
-### Bug Fixes
-
-* add commitlint as husky pre-commit hook and ci check ([#25](https://github.com/iggy-rs/iggy-node-client/issues/25)) ([5dbac10](https://github.com/iggy-rs/iggy-node-client/commit/5dbac1071bea5c26f852a60361ac51217df09d25))
-
-## [1.0.5](https://github.com/iggy-rs/iggy-node-client/compare/v1.0.4...v1.0.5) (2025-01-14)
-
-### Bug Fixes
-
-* set package.json license to Apache-2.0 as well ([#24](https://github.com/iggy-rs/iggy-node-client/issues/24)) ([2523c0f](https://github.com/iggy-rs/iggy-node-client/commit/2523c0fde82958e8a13bb95b2c2d0babe0d1d290))
-
-## [1.0.4](https://github.com/iggy-rs/iggy-node-client/compare/v1.0.3...v1.0.4) (2024-12-16)
-
-### Bug Fixes
-
-* add tests, fix headers.double type, ehance typing ([431063f](https://github.com/iggy-rs/iggy-node-client/commit/431063f253e0fb1739188bbd6614cfc06ccb3fd4))
-
-## [1.0.3](https://github.com/iggy-rs/iggy-node-client/compare/v1.0.2...v1.0.3) (2024-11-24)
-
-### Bug Fixes
-
-* upgrade dependencies ([#19](https://github.com/iggy-rs/iggy-node-client/issues/19)) ([a05d5c2](https://github.com/iggy-rs/iggy-node-client/commit/a05d5c2484f8711f72a23c4ee1222d29e323a5ba))
-
-## [1.0.2](https://github.com/iggy-rs/iggy-node-client/compare/v1.0.1...v1.0.2) (2024-11-24)
-
-### Bug Fixes
-
-* **npm:** configure package as public ([#18](https://github.com/iggy-rs/iggy-node-client/issues/18)) ([7a56455](https://github.com/iggy-rs/iggy-node-client/commit/7a5645512a564322af343bc7ba748c8c58dfab1e))
-
-## [1.0.1](https://github.com/iggy-rs/iggy-node-client/compare/v1.0.0...v1.0.1) (2024-11-24)
-
-### Bug Fixes
-
-* **npm:** publish package ([#17](https://github.com/iggy-rs/iggy-node-client/issues/17)) ([6e2f60e](https://github.com/iggy-rs/iggy-node-client/commit/6e2f60e2f4484b57596285ff79d76ea647962595))
-
-## 1.0.0 (2024-11-24)
-
-### Bug Fixes
-
-* add e2e tests, fix create-group return ([ad52b43](https://github.com/iggy-rs/iggy-node-client/commit/ad52b43a6ee5e8f868eb1178a9b9f02f11cb1204))
-* add package's keywords, cleans up logs ([5287b74](https://github.com/iggy-rs/iggy-node-client/commit/5287b74006733aef4429d889c0cc38a80f375dc2))
-* correctly destroy socket when destroy function is called on TCPClient ([76f7d07](https://github.com/iggy-rs/iggy-node-client/commit/76f7d07a96e44f6701998bb9fb2dab8fa4d75489))
-* enforce check on message id to be either uuid string or 0 ([d05b898](https://github.com/iggy-rs/iggy-node-client/commit/d05b898d62b097ca9770cdab289707546c2fe6aa))
-* fix client auth quirks, add handleResponse & deserializePollMessage as transform stream ([0458d57](https://github.com/iggy-rs/iggy-node-client/commit/0458d579de45605588fff7b7d01119c9625364da))
-* fix consumer group commands ([a33094f](https://github.com/iggy-rs/iggy-node-client/commit/a33094f66eb2e433b2fc8cc6adf7f6a122b0f365))
-* fix createtopic, add new compressionAlgorithm param ([f93a444](https://github.com/iggy-rs/iggy-node-client/commit/f93a4441f4bcd790763d982ead981c4ac86b33c5))
-* fix getStats command (add new totalCpuUsage field) ([#3](https://github.com/iggy-rs/iggy-node-client/issues/3)) ([de4cfda](https://github.com/iggy-rs/iggy-node-client/commit/de4cfdad4046f556a51878fbadb8dde0df9302c5))
-* fix github ci ([78870e3](https://github.com/iggy-rs/iggy-node-client/commit/78870e389333c8b7ca2425748c8806fa5e36a7ae))
-* fix message header typing ([1276374](https://github.com/iggy-rs/iggy-node-client/commit/12763749f95d29a78028f95b5bc33281a62246c9))
-* fix message headers serialization bug ([22ffe16](https://github.com/iggy-rs/iggy-node-client/commit/22ffe1603db9b7a94deadb1fcf6a25f81cfea868))
-* fix module type export ([b0bcdc7](https://github.com/iggy-rs/iggy-node-client/commit/b0bcdc7945ba68d519a129ef33b4353538e9d64e))
-* fix npm test command for ci ([842fe69](https://github.com/iggy-rs/iggy-node-client/commit/842fe697548224a08012574d2e44f14100105b63))
-* fix Partitioning.MessageKey type, fix indent ([05d05b6](https://github.com/iggy-rs/iggy-node-client/commit/05d05b6db68b33bc9045fee16c9fca8c8eb0ae6d))
-* fix tcp client options ([f9bd442](https://github.com/iggy-rs/iggy-node-client/commit/f9bd44204f86f2becfe674c9b38f09657c00ac6c))
-* fix topic deserialisation bug ([3e787be](https://github.com/iggy-rs/iggy-node-client/commit/3e787be2546f29f1f8d31e9e1388e19a62729e50))
-* fix updateUser and changePassword command ([d01e086](https://github.com/iggy-rs/iggy-node-client/commit/d01e08621bbf976c7dc5578273a253a0dcc43e72))
-* fix var naming, add some test ([40d91dd](https://github.com/iggy-rs/iggy-node-client/commit/40d91ddfe41f6114ca3c7da2a30225e81e3226bc))
-* get rid of enums, add type helpers ([6e2613b](https://github.com/iggy-rs/iggy-node-client/commit/6e2613b2f1ab0112d401f87a2f0cfb6e77b8d99d))
-* more e2e tests ([d537aa7](https://github.com/iggy-rs/iggy-node-client/commit/d537aa7454b983edb471fa354e55b5e814fb1524))
-* no ssh pull ([#12](https://github.com/iggy-rs/iggy-node-client/issues/12)) ([f13f0ed](https://github.com/iggy-rs/iggy-node-client/commit/f13f0edd5adf4584ab7f02620a159317654251e5))
-* remove bad symbol in the CI definition ([1878a0c](https://github.com/iggy-rs/iggy-node-client/commit/1878a0c501224d17eefc986ccd5d89e481cc15b8))
-* remove console.error on normal close signal ([0f8c993](https://github.com/iggy-rs/iggy-node-client/commit/0f8c99330b2352220ee23df0b7923185d710cd89))
-* remove initial .gitignore and README.md to avoid rebase conflict ([931b9b8](https://github.com/iggy-rs/iggy-node-client/commit/931b9b8f5d0b8a249de8a30c3be9435c93a8461f))
-* update createUser, createStream & createTopic command return value ([8712b0f](https://github.com/iggy-rs/iggy-node-client/commit/8712b0f5021b7852361117f6fc764eac3851beb7))
-* update readme ([263f271](https://github.com/iggy-rs/iggy-node-client/commit/263f271fed2c529b89329e88d83fc4100b04c639))
-* use debug lib, make poolsize configurable ([2f66e89](https://github.com/iggy-rs/iggy-node-client/commit/2f66e89220b30b3e33aa773b6471b1669658f37f))
-* use pat as token for semantic release ([#10](https://github.com/iggy-rs/iggy-node-client/issues/10)) ([5eccfd2](https://github.com/iggy-rs/iggy-node-client/commit/5eccfd281763773140ce9be39a2c022b2de803b6))
-
-### Features
-
-* add base ci workflow ([#2](https://github.com/iggy-rs/iggy-node-client/issues/2)) ([4cbde14](https://github.com/iggy-rs/iggy-node-client/commit/4cbde140409841bf3d440f01ccaca1a855f37b13))
-* add command client with socket pool management ([e747b29](https://github.com/iggy-rs/iggy-node-client/commit/e747b292374a7a8e1ee8f27d69a9203db3a6b09b))
-* add CommandResponseStream to wrap tcp socket, add parallel call safetiness ([0e6f38f](https://github.com/iggy-rs/iggy-node-client/commit/0e6f38fde621dc1cfe3f12376723f1b43ffee9bf))
-* add consumer stream facility ([8c19d3f](https://github.com/iggy-rs/iggy-node-client/commit/8c19d3fae30a5ab47d0f10b4747158e604aea37f))
-* add create, delete, join & leave consumer-group command ([e7ca376](https://github.com/iggy-rs/iggy-node-client/commit/e7ca3762a9fcb92a3bb5018cb66570079bec60f1))
-* add createPartition & deletePartition command ([2286c10](https://github.com/iggy-rs/iggy-node-client/commit/2286c106534704b02544613944e74f2549ca1a67))
-* add createUser and deleteUser command ([d25d837](https://github.com/iggy-rs/iggy-node-client/commit/d25d837d38abe89a038880651c19bd8925f1affe))
-* add getGroup and getGroups command ([cdfa766](https://github.com/iggy-rs/iggy-node-client/commit/cdfa766c9b800e59c2858599ccf87afa44c142dc))
-* add getOffset and storeOffset command, fix typos ([6f9a425](https://github.com/iggy-rs/iggy-node-client/commit/6f9a4256b713abcfdd6dc146b78268169c9bb198))
-* add pollMessage command ([fe59f82](https://github.com/iggy-rs/iggy-node-client/commit/fe59f825ebb0ba3a4c4c53e90cd8a28f9b21cca2))
-* add purgeTopic & purgeStream command ([86482d1](https://github.com/iggy-rs/iggy-node-client/commit/86482d12bfdf48b23973374e09cb9cc5e852ef18))
-* add SendMessages command ([e1d39d8](https://github.com/iggy-rs/iggy-node-client/commit/e1d39d887e36783e2ca23cd560060df3fc62099a))
-* add updateStream command ([89970c0](https://github.com/iggy-rs/iggy-node-client/commit/89970c0574f0bcade634b338522bcbe6990b4b43))
-* add updateTopic command ([f1e278e](https://github.com/iggy-rs/iggy-node-client/commit/f1e278e993a05c10f10a24761bd2e63f95d9891b))
-* add updateUser and changePassword command, fix permissions deserialization bug ([f3bcda3](https://github.com/iggy-rs/iggy-node-client/commit/f3bcda30b71f5d70145257534848ee4053e714f7))
-* better error, add some test ([f386854](https://github.com/iggy-rs/iggy-node-client/commit/f386854bd388e95a11251803dcb81292b33d7981))
-* publish to npm ([82268f5](https://github.com/iggy-rs/iggy-node-client/commit/82268f5963e72865355e2cfe919307d5bd1500ab))
-* reorganize client declaration ([7cb1bd8](https://github.com/iggy-rs/iggy-node-client/commit/7cb1bd857c5d1050ab6b8d2b0e6b70908b3bc105))
-* start low level command api and base tcp client ([7bb8b32](https://github.com/iggy-rs/iggy-node-client/commit/7bb8b32ec809e53d665295f9e0c069d535b88cae))
-* start unit test on serialization ([2082b71](https://github.com/iggy-rs/iggy-node-client/commit/2082b713a37d59a3d0482669a3a544a6a922ae41))
-* update modified commands for v0.3.0 server release (createTopic, updateTopic, login, createToken) ([cb4f0d1](https://github.com/iggy-rs/iggy-node-client/commit/cb4f0d17c61967e95a6b5502c70d6e8b8569e9c7))
-* wraps command to higher level api, starts client ([60f9466](https://github.com/iggy-rs/iggy-node-client/commit/60f9466cf92965f3bf4ff5918e50876e5bb9aa84))
diff --git a/foreign/node/README.md b/foreign/node/README.md
index ecb045379..994ffee60 100644
--- a/foreign/node/README.md
+++ b/foreign/node/README.md
@@ -32,18 +32,17 @@ npm i --save apache-iggy
### Response frame limit
-**Compatibility note:** response frames larger than `maxResponseFrameSize` (default 64 MiB) are now rejected and close the connection under both framing modes. This is a behavior change for existing classic-framing clients. Raise the limit in the client configuration when polling very large batches.
+**Compatibility note:** response frames larger than `maxResponseFrameSize` (default 64 MiB) are rejected and close the connection. Raise the limit in the client configuration when polling very large batches.
### VSR framing
-Classic framing remains the default. Select VSR explicitly when connecting to
-an Iggy VSR server:
+The SDK speaks the VSR wire protocol exclusively and requires an Iggy VSR
+server:
```typescript
import { SimpleClient, getRawClient } from "apache-iggy";
const config = {
- protocol: "vsr" as const,
transport: "TCP" as const,
options: { host: "127.0.0.1", port: 8090 },
credentials: { username: "iggy", password: "iggy" },
@@ -52,12 +51,18 @@ const client = new SimpleClient(getRawClient(config));
const stats = await client.system.getStats();
```
-VSR is a runtime protocol choice in Node.js, not a build feature. Codes absent
-from the SDK command table use `Operation::NonReplicated` and carry the command
-code in the request header's reserved field. The server remains authoritative
-for classifying or rejecting extension commands.
+Codes absent from the SDK command table use `Operation::NonReplicated` and
+carry the command code in the request header's reserved field. The server
+remains authoritative for classifying or rejecting extension commands.
-The same npm package supports both framing modes over TCP and TLS. VSR restricts `Client` to one pooled connection because authentication, request sequencing, and consumer-group assignments belong to one consensus session. Configurations requesting more than one pooled connection fail before a socket is opened.
+Sends must use explicit `Partitioning.PartitionId` partitioning: the client
+routes each request to a partition-scoped namespace, so broker-side balancing
+(`Partitioning.Balanced`) and key hashing (`Partitioning.MessageKey`) are
+rejected before the request is sent.
+<!-- TODO(hubcio): Balanced and MessageKey partitioning to be implemented;
+not decided yet whether it'll be on server side or client side. -->
+
+VSR works over TCP and TLS. It restricts `Client` to one pooled connection because authentication, request sequencing, and consumer-group assignments belong to one consensus session. Configurations requesting more than one pooled connection fail before a socket is opened.
VSR authentication translates the existing password and personal-access-token
login APIs into the register handshake required by the consensus protocol. A
@@ -68,7 +73,7 @@ new session.
When the server's `[heartbeat]` eviction is enabled, configure the client's `heartbeatInterval` below the server heartbeat interval. Client heartbeats are disabled when `heartbeatInterval` is unset.
-`sendBinaryRequest(code, payload)` has the same signature under classic and VSR framing. Known replicated commands use their registered operation, while unknown codes reach the server as non-replicated requests and are rejected by servers that do not register them. Classic request bytes remain unchanged.
+`sendBinaryRequest(code, payload)` sends an arbitrary command code. Known replicated commands use their registered operation, while unknown codes reach the server as non-replicated requests and are rejected by servers that do not register them.
```typescript
import { ResponseError } from "apache-iggy";
diff --git a/foreign/node/docker-compose.yml b/foreign/node/docker-compose.yml
index b30064648..be5774ead 100644
--- a/foreign/node/docker-compose.yml
+++ b/foreign/node/docker-compose.yml
@@ -17,14 +17,19 @@
services:
iggy-server:
- image: apache/iggy:latest
+ # The SDK is vsr-only, so build the vsr server from the repo instead
+ # of pulling the legacy apache/iggy image.
+ # TODO(hubcio): change to iggy-server once legacy server is removed
+ # (core/server has VSR support)
+ build:
+ context: ../..
+ dockerfile: core/server-ng/Dockerfile
container_name: iggy-server
restart: unless-stopped
networks:
- iggy
ports:
- 3000:3000
- - 8080:8080
- 8090:8090
volumes:
- iggy-server:/local_data
diff --git a/foreign/node/package-lock.json b/foreign/node/package-lock.json
index 11295a84a..b1a49f724 100644
--- a/foreign/node/package-lock.json
+++ b/foreign/node/package-lock.json
@@ -1,12 +1,12 @@
{
"name": "apache-iggy",
- "version": "0.9.0-edge.1",
+ "version": "0.10.0-edge.1",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "apache-iggy",
- "version": "0.9.0-edge.1",
+ "version": "0.10.0-edge.1",
"license": "Apache-2.0",
"dependencies": {
"debug": "4.4.3",
diff --git a/foreign/node/package.json b/foreign/node/package.json
index 7f4bcb6fc..b423ee3da 100644
--- a/foreign/node/package.json
+++ b/foreign/node/package.json
@@ -1,7 +1,7 @@
{
"name": "apache-iggy",
"type": "module",
- "version": "0.9.0-edge.1",
+ "version": "0.10.0-edge.1",
"description": "Official Apache Iggy NodeJS SDK",
"keywords": [
"iggy",
@@ -37,7 +37,6 @@
"scripts": {
"test:unit": "node --import @swc-node/register/esm-register --test --experimental-test-coverage './src/**/*.test.ts'",
"test:e2e": "node --import @swc-node/register/esm-register --test --experimental-test-coverage --test-force-exit './src/e2e/*.e2e.ts'",
- "test:e2e:vsr": "IGGY_TEST_PROTOCOL=vsr node --import @swc-node/register/esm-register --test --experimental-test-coverage --test-force-exit './src/e2e/tcp.*.e2e.ts'",
"test:bdd": "cucumber-js --exit",
"test": "npm run test:unit && npm run test:bdd && npm run test:e2e",
"clean": "rm -Rf dist/",
diff --git a/foreign/node/src/bdd/auth.ts b/foreign/node/src/bdd/auth.ts
index a67e49692..147327ee1 100644
--- a/foreign/node/src/bdd/auth.ts
+++ b/foreign/node/src/bdd/auth.ts
@@ -18,17 +18,14 @@
import assert from 'node:assert/strict';
import { Client } from '../client/index.js';
-import type { Protocol } from '../client/index.js';
import { Given } from "@cucumber/cucumber";
import type { TestWorld } from './world.js';
import { getIggyAddress } from '../tcp.sm.utils.js';
const credentials = { username: 'iggy', password: 'iggy' };
const [host, port] = getIggyAddress();
-const protocol: Protocol = process.env.IGGY_TEST_PROTOCOL === 'vsr' ? 'vsr' : 'classic';
const opt = {
- protocol,
transport: 'TCP' as const,
options: { host, port },
credentials
diff --git a/foreign/node/src/client/client.config.test.ts b/foreign/node/src/client/client.config.test.ts
index bc612da93..e4402e1b9 100644
--- a/foreign/node/src/client/client.config.test.ts
+++ b/foreign/node/src/client/client.config.test.ts
@@ -30,55 +30,34 @@ const config = (): ClientConfig => ({
});
describe('normalizeClientConfig', () => {
- it('defaults to classic without changing classic pool sizing', () => {
- const normalized = normalizeClientConfig({
- ...config(),
- poolSize: { min: 2, max: 4 }
- });
+ it('applies the default response frame limit', () => {
+ const normalized = normalizeClientConfig(config());
- assert.equal(normalized.protocol, 'classic');
- assert.deepEqual(normalized.poolSize, { min: 2, max: 4 });
assert.equal(
normalized.maxResponseFrameSize,
DEFAULT_MAX_RESPONSE_FRAME_SIZE
);
});
- it('restricts VSR to one pooled connection', () => {
- const normalized = normalizeClientConfig({
- ...config(),
- protocol: 'vsr'
- });
+ it('restricts the client to one pooled connection', () => {
+ const normalized = normalizeClientConfig(config());
assert.deepEqual(normalized.poolSize, { min: 1, max: 1 });
assert.throws(
() => normalizeClientConfig({
...config(),
- protocol: 'vsr',
poolSize: { max: 2 }
}),
/exactly one pooled connection/
);
});
- it('rejects invalid protocols before opening a socket', () => {
- assert.throws(
- () => normalizeClientConfig({
- ...config(),
- protocol: 'auto' as 'vsr'
- }),
- /unsupported wire protocol/
- );
- });
-
- it('supports VSR over TLS', () => {
+ it('supports TLS transport', () => {
const normalized = normalizeClientConfig({
...config(),
- protocol: 'vsr',
transport: 'TLS'
});
- assert.equal(normalized.protocol, 'vsr');
assert.equal(normalized.transport, 'TLS');
assert.deepEqual(normalized.poolSize, { min: 1, max: 1 });
});
diff --git a/foreign/node/src/client/client.config.ts b/foreign/node/src/client/client.config.ts
index 7fbb13e25..ed4030819 100644
--- a/foreign/node/src/client/client.config.ts
+++ b/foreign/node/src/client/client.config.ts
@@ -15,20 +15,13 @@
// specific language governing permissions and limitations
// under the License.
-import type { ClientConfig, Protocol } from './client.type.js';
+import type { ClientConfig } from './client.type.js';
export const DEFAULT_MAX_RESPONSE_FRAME_SIZE = 64 * 1024 * 1024;
-const isProtocol = (value: unknown): value is Protocol =>
- value === 'classic' || value === 'vsr';
-
export const normalizeClientConfig = (
config: ClientConfig
-): ClientConfig & { protocol: Protocol } => {
- const protocol = config.protocol ?? 'classic';
- if (!isProtocol(protocol))
- throw new TypeError(`unsupported wire protocol: ${String(protocol)}`);
-
+): ClientConfig => {
const maxResponseFrameSize =
config.maxResponseFrameSize ?? DEFAULT_MAX_RESPONSE_FRAME_SIZE;
if (!Number.isSafeInteger(maxResponseFrameSize) ||
@@ -37,17 +30,15 @@ export const normalizeClientConfig = (
'maxResponseFrameSize must be a safe integer of at least 256 bytes'
);
- if (protocol === 'vsr' &&
- ((config.poolSize?.min ?? 1) > 1 || (config.poolSize?.max ?? 1) > 1))
+ if ((config.poolSize?.min ?? 1) > 1 || (config.poolSize?.max ?? 1) > 1)
throw new TypeError(
'VSR clients currently support exactly one pooled connection'
);
return {
...config,
- protocol,
options: { ...config.options },
maxResponseFrameSize,
- ...(protocol === 'vsr' ? { poolSize: { min: 1, max: 1 } } : {})
+ poolSize: { min: 1, max: 1 }
};
};
diff --git a/foreign/node/src/client/client.connection.test.ts b/foreign/node/src/client/client.connection.test.ts
index d9639f432..5c11c54ba 100644
--- a/foreign/node/src/client/client.connection.test.ts
+++ b/foreign/node/src/client/client.connection.test.ts
@@ -28,6 +28,9 @@ import { describe, it } from 'node:test';
import { ProtocolFrameError } from './client.frame.js';
import { IggyConnection } from './client.connection.js';
import type { ClientConfig } from './client.type.js';
+import { Command2, HEADER_SIZE, REPLY_OFFSET } from '../wire/vsr/header.js';
+
+const FRAME_LIMIT = 2 * HEADER_SIZE;
const startServer = async (): Promise<Server> => {
const server = createServer();
@@ -37,7 +40,6 @@ const startServer = async (): Promise<Server> => {
};
const connectionConfig = (server: Server): ClientConfig => ({
- protocol: 'classic',
transport: 'TCP',
options: {
host: '127.0.0.1',
@@ -45,9 +47,17 @@ const connectionConfig = (server: Server): ClientConfig => ({
},
credentials: { username: 'iggy', password: 'iggy' },
reconnect: { enabled: false, interval: 0, maxRetries: 0 },
- maxResponseFrameSize: 256
+ maxResponseFrameSize: FRAME_LIMIT
});
+const replyFrame = (body: Buffer): Buffer => {
+ const frame = Buffer.alloc(HEADER_SIZE + body.length);
+ frame.writeUInt32LE(frame.length, REPLY_OFFSET.size);
+ frame.writeUInt8(Command2.Reply, REPLY_OFFSET.command);
+ body.copy(frame, HEADER_SIZE);
+ return frame;
+};
+
const closeConnection = async (
connection: IggyConnection,
server: Server
@@ -75,7 +85,7 @@ describe('IggyConnection', () => {
}
);
- it('shares connection attempts, recognizes endpoints, and writes commands',
+ it('shares connection attempts, recognizes endpoints, and writes frames',
async () => {
const server = await startServer();
const received = new Promise<Buffer>((resolve) => {
@@ -109,10 +119,9 @@ describe('IggyConnection', () => {
true
);
- connection.writeCommand(1, Buffer.from('payload'));
- const command = await received;
- assert.equal(command.readUInt32LE(4), 1);
- assert.deepEqual(command.subarray(8), Buffer.from('payload'));
+ const frame = replyFrame(Buffer.from('payload'));
+ connection.writeFrame(frame);
+ assert.deepEqual(await received, frame);
} finally {
await closeConnection(connection, server);
}
@@ -125,17 +134,14 @@ describe('IggyConnection', () => {
const connection = new IggyConnection(connectionConfig(server));
try {
await connection.connect();
- const body = Buffer.from('response');
- const frame = Buffer.alloc(8 + body.length);
- frame.writeUInt32LE(body.length, 4);
- body.copy(frame, 8);
+ const frame = replyFrame(Buffer.from('response'));
const response = once(connection, 'response');
connection._onData(frame.subarray(0, 6));
connection._onData(frame.subarray(6));
assert.deepEqual((await response)[0], frame);
- const malformed = Buffer.alloc(8);
- malformed.writeUInt32LE(256, 4);
+ const malformed = replyFrame(Buffer.alloc(0));
+ malformed.writeUInt32LE(FRAME_LIMIT + 1, REPLY_OFFSET.size);
const error = once(connection, 'error');
connection._onData(malformed);
assert.ok((await error)[0] instanceof ProtocolFrameError);
@@ -225,10 +231,7 @@ describe('IggyConnection', () => {
assert.equal(connection.connected, true);
assert.equal(connections, 2);
- const body = Buffer.from('fresh');
- const frame = Buffer.alloc(8 + body.length);
- frame.writeUInt32LE(body.length, 4);
- body.copy(frame, 8);
+ const frame = replyFrame(Buffer.from('fresh'));
const response = once(connection, 'response');
oldSocket.emit('data', Buffer.alloc(8));
connection._onData(frame);
diff --git a/foreign/node/src/client/client.connection.ts b/foreign/node/src/client/client.connection.ts
index 25e4274b2..99bcda97c 100644
--- a/foreign/node/src/client/client.connection.ts
+++ b/foreign/node/src/client/client.connection.ts
@@ -21,7 +21,6 @@ import type { Socket } from 'node:net';
import { createConnection } from 'node:net';
import { connect as TLSConnect } from 'node:tls';
import type { ClientConfig, TlsOption, TcpOption, ReconnectOption } from "./client.type.js"
-import { serializeCommand } from './client.utils.js';
import { debug } from './client.debug.js';
import { DEFAULT_MAX_RESPONSE_FRAME_SIZE } from './client.config.js';
import {
@@ -141,7 +140,6 @@ export class IggyConnection extends EventEmitter {
this.connectPromise = undefined;
this.reconnectPromise = undefined;
this.responseDecoder = new ResponseFrameDecoder(
- config.protocol ?? 'classic',
config.maxResponseFrameSize ?? DEFAULT_MAX_RESPONSE_FRAME_SIZE
);
this.socket = this._installSocket(getTransport(config));
@@ -427,8 +425,7 @@ export class IggyConnection extends EventEmitter {
try {
for (const response of this.responseDecoder.push(data)) {
- if (this.config.protocol === 'vsr' &&
- peekCommand(response) === Command2.Eviction)
+ if (peekCommand(response) === Command2.Eviction)
this.emit('eviction', evictionError(response));
else
this.emit('response', response);
@@ -443,18 +440,6 @@ export class IggyConnection extends EventEmitter {
}
}
- /**
- * Writes a command to the socket.
- *
- * @param command - Command code
- * @param payload - Command payload
- * @returns True if the write was successful
- */
- writeCommand(command: number, payload: Buffer): void {
- const cmd = serializeCommand(command, payload);
- this.socket.write(cmd);
- }
-
writeFrame(frame: Buffer): void {
this.socket.write(frame);
}
diff --git a/foreign/node/src/client/client.frame.test.ts b/foreign/node/src/client/client.frame.test.ts
index 5b05481d8..bdeba1eda 100644
--- a/foreign/node/src/client/client.frame.test.ts
+++ b/foreign/node/src/client/client.frame.test.ts
@@ -30,13 +30,6 @@ import {
const LIMIT = 1024;
-const classicFrame = (body: Buffer): Buffer => {
- const frame = Buffer.alloc(8 + body.length);
- frame.writeUInt32LE(body.length, 4);
- body.copy(frame, 8);
- return frame;
-};
-
const vsrFrame = (body: Buffer): Buffer => {
const frame = Buffer.alloc(HEADER_SIZE + body.length);
frame.writeUInt32LE(frame.length, REPLY_OFFSET.size);
@@ -46,83 +39,59 @@ const vsrFrame = (body: Buffer): Buffer => {
};
describe('extractResponseFrames', () => {
- for (const protocol of ['classic', 'vsr'] as const) {
- const makeFrame = protocol === 'classic' ? classicFrame : vsrFrame;
-
- it(`buffers ${protocol} headers split at every boundary`, () => {
- const frame = makeFrame(Buffer.from('payload'));
- const headerSize = protocol === 'classic' ? 8 : HEADER_SIZE;
-
- for (let split = 0; split < headerSize; split += 1) {
- const first = extractResponseFrames(
- protocol,
- frame.subarray(0, split),
- LIMIT
- );
- assert.equal(first.frames.length, 0);
- const second = extractResponseFrames(
- protocol,
- Buffer.concat([first.remainder, frame.subarray(split)]),
- LIMIT
- );
- assert.deepEqual(second.frames, [frame]);
- assert.equal(second.remainder.length, 0);
- }
- });
+ it('buffers headers split at every boundary', () => {
+ const frame = vsrFrame(Buffer.from('payload'));
- it(`buffers a fragmented ${protocol} body`, () => {
- const frame = makeFrame(Buffer.from('payload'));
- const split = frame.length - 2;
- const first = extractResponseFrames(
- protocol,
- frame.subarray(0, split),
- LIMIT
- );
+ for (let split = 0; split < HEADER_SIZE; split += 1) {
+ const first = extractResponseFrames(frame.subarray(0, split), LIMIT);
assert.equal(first.frames.length, 0);
const second = extractResponseFrames(
- protocol,
Buffer.concat([first.remainder, frame.subarray(split)]),
LIMIT
);
assert.deepEqual(second.frames, [frame]);
- });
+ assert.equal(second.remainder.length, 0);
+ }
+ });
- it(`extracts coalesced ${protocol} frames and a partial tail`, () => {
- const first = makeFrame(Buffer.from('one'));
- const second = makeFrame(Buffer.from('two'));
- const third = makeFrame(Buffer.from('three'));
- const input = Buffer.concat([first, second, third.subarray(0, 3)]);
- const extracted = extractResponseFrames(protocol, input, LIMIT);
+ it('buffers a fragmented body', () => {
+ const frame = vsrFrame(Buffer.from('payload'));
+ const split = frame.length - 2;
+ const first = extractResponseFrames(frame.subarray(0, split), LIMIT);
+ assert.equal(first.frames.length, 0);
+ const second = extractResponseFrames(
+ Buffer.concat([first.remainder, frame.subarray(split)]),
+ LIMIT
+ );
+ assert.deepEqual(second.frames, [frame]);
+ });
- assert.deepEqual(extracted.frames, [first, second]);
- assert.deepEqual(extracted.remainder, third.subarray(0, 3));
- assert.equal(extracted.remainder.buffer, input.buffer);
- });
- }
+ it('extracts coalesced frames and a partial tail', () => {
+ const first = vsrFrame(Buffer.from('one'));
+ const second = vsrFrame(Buffer.from('two'));
+ const third = vsrFrame(Buffer.from('three'));
+ const input = Buffer.concat([first, second, third.subarray(0, 3)]);
+ const extracted = extractResponseFrames(input, LIMIT);
- it('rejects a VSR size below the header', () => {
+ assert.deepEqual(extracted.frames, [first, second]);
+ assert.deepEqual(extracted.remainder, third.subarray(0, 3));
+ assert.equal(extracted.remainder.buffer, input.buffer);
+ });
+
+ it('rejects a size below the header', () => {
const frame = vsrFrame(Buffer.alloc(0));
frame.writeUInt32LE(0, REPLY_OFFSET.size);
assert.throws(
- () => extractResponseFrames('vsr', frame, LIMIT),
+ () => extractResponseFrames(frame, LIMIT),
ProtocolFrameError
);
});
- it('rejects an oversized VSR frame before buffering its body', () => {
+ it('rejects an oversized frame before buffering its body', () => {
const header = vsrFrame(Buffer.alloc(0));
header.writeUInt32LE(LIMIT + 1, REPLY_OFFSET.size);
assert.throws(
- () => extractResponseFrames('vsr', header, LIMIT),
- ProtocolFrameError
- );
- });
-
- it('rejects an oversized classic frame before buffering its body', () => {
- const header = classicFrame(Buffer.alloc(0));
- header.writeUInt32LE(LIMIT, 4);
- assert.throws(
- () => extractResponseFrames('classic', header, LIMIT),
+ () => extractResponseFrames(header, LIMIT),
ProtocolFrameError
);
});
@@ -130,9 +99,9 @@ describe('extractResponseFrames', () => {
describe('ResponseFrameDecoder', () => {
it('decodes bytewise input without losing coalesced frames', () => {
- const decoder = new ResponseFrameDecoder('classic', LIMIT);
- const first = classicFrame(Buffer.from('first'));
- const second = classicFrame(Buffer.from('second'));
+ const decoder = new ResponseFrameDecoder(LIMIT);
+ const first = vsrFrame(Buffer.from('first'));
+ const second = vsrFrame(Buffer.from('second'));
const input = Buffer.concat([first, second]);
const frames: Buffer[] = [];
@@ -144,17 +113,17 @@ describe('ResponseFrameDecoder', () => {
});
it('clears a partial frame', () => {
- const decoder = new ResponseFrameDecoder('vsr', LIMIT);
- decoder.push(vsrFrame(Buffer.from('body')).subarray(0, 100));
+ const decoder = new ResponseFrameDecoder(LIMIT);
+ decoder.push(vsrFrame(Buffer.from('body')).subarray(0, HEADER_SIZE - 2));
assert.equal(decoder.hasBufferedData, true);
decoder.clear();
assert.equal(decoder.hasBufferedData, false);
});
it('rejects an oversized frame as soon as its header is complete', () => {
- const decoder = new ResponseFrameDecoder('classic', LIMIT);
- const header = Buffer.alloc(8);
- header.writeUInt32LE(LIMIT, 4);
+ const decoder = new ResponseFrameDecoder(LIMIT);
+ const header = vsrFrame(Buffer.alloc(0));
+ header.writeUInt32LE(LIMIT + 1, REPLY_OFFSET.size);
assert.throws(() => decoder.push(header), ProtocolFrameError);
});
});
diff --git a/foreign/node/src/client/client.frame.ts b/foreign/node/src/client/client.frame.ts
index 4bd90ff5b..38cc0566a 100644
--- a/foreign/node/src/client/client.frame.ts
+++ b/foreign/node/src/client/client.frame.ts
@@ -15,14 +15,11 @@
// specific language governing permissions and limitations
// under the License.
-import type { Protocol } from './client.type.js';
import {
- HEADER_SIZE as VSR_HEADER_SIZE,
- readSize as readVsrSize
+ HEADER_SIZE,
+ readSize
} from '../wire/vsr/header.js';
-const CLASSIC_HEADER_SIZE = 8;
-
export class ProtocolFrameError extends Error {
constructor(message: string) {
super(message);
@@ -35,45 +32,35 @@ export type ExtractedFrames = {
remainder: Buffer
};
-const headerSizeFor = (protocol: Protocol): number =>
- protocol === 'vsr' ? VSR_HEADER_SIZE : CLASSIC_HEADER_SIZE;
-
const declaredFrameSize = (
- protocol: Protocol,
header: Buffer,
maximumFrameSize: number
): number => {
- const headerSize = headerSizeFor(protocol);
- const declaredSize = protocol === 'vsr'
- ? readVsrSize(header)
- : CLASSIC_HEADER_SIZE + header.readUInt32LE(4);
+ const declaredSize = readSize(header);
- if (declaredSize < headerSize)
+ if (declaredSize < HEADER_SIZE)
throw new ProtocolFrameError(
- `declared ${protocol} frame size ${declaredSize} is below header size`
+ `declared frame size ${declaredSize} is below header size`
);
if (declaredSize > maximumFrameSize)
throw new ProtocolFrameError(
- `declared ${protocol} frame size ${declaredSize} exceeds ` +
+ `declared frame size ${declaredSize} exceeds ` +
`the ${maximumFrameSize} byte limit`
);
return declaredSize;
};
export const extractResponseFrames = (
- protocol: Protocol,
buffer: Buffer,
maximumFrameSize: number
): ExtractedFrames => {
- const headerSize = headerSizeFor(protocol);
const frames: Buffer[] = [];
let offset = 0;
- while (buffer.length - offset >= headerSize) {
+ while (buffer.length - offset >= HEADER_SIZE) {
const available = buffer.length - offset;
const declaredSize = declaredFrameSize(
- protocol,
- buffer.subarray(offset, offset + headerSize),
+ buffer.subarray(offset, offset + HEADER_SIZE),
maximumFrameSize
);
if (available < declaredSize)
@@ -96,19 +83,15 @@ export const extractResponseFrames = (
* incomplete frame as new socket chunks arrive.
*/
export class ResponseFrameDecoder {
- private readonly protocol: Protocol;
private readonly maximumFrameSize: number;
- private readonly headerSize: number;
private chunks: Buffer[];
private chunkIndex: number;
private chunkOffset: number;
private bufferedLength: number;
private expectedFrameSize?: number;
- constructor(protocol: Protocol, maximumFrameSize: number) {
- this.protocol = protocol;
+ constructor(maximumFrameSize: number) {
this.maximumFrameSize = maximumFrameSize;
- this.headerSize = headerSizeFor(protocol);
this.chunks = [];
this.chunkIndex = 0;
this.chunkOffset = 0;
@@ -137,11 +120,10 @@ export class ResponseFrameDecoder {
const frames: Buffer[] = [];
while (true) {
if (this.expectedFrameSize === undefined) {
- if (this.bufferedLength < this.headerSize)
+ if (this.bufferedLength < HEADER_SIZE)
break;
this.expectedFrameSize = declaredFrameSize(
- this.protocol,
- this.peek(this.headerSize),
+ this.peek(HEADER_SIZE),
this.maximumFrameSize
);
}
diff --git a/foreign/node/src/client/client.socket.test.ts b/foreign/node/src/client/client.socket.test.ts
index 76c8a783c..33e841f9d 100644
--- a/foreign/node/src/client/client.socket.test.ts
+++ b/foreign/node/src/client/client.socket.test.ts
@@ -203,7 +203,6 @@ const singleNodeHandler = (port: number): FrameHandler =>
};
const vsrConfig = (port: number): ClientConfig => ({
- protocol: 'vsr',
transport: 'TCP',
options: { host: '127.0.0.1', port },
credentials: { username: 'iggy', password: 'iggy' },
diff --git a/foreign/node/src/client/client.socket.ts b/foreign/node/src/client/client.socket.ts
index a29e328dd..9bbafa401 100644
--- a/foreign/node/src/client/client.socket.ts
+++ b/foreign/node/src/client/client.socket.ts
@@ -20,10 +20,9 @@ import { EventEmitter } from 'node:events';
import type {
ClientConfig,
ClientCredentials, CommandResponse,
- PasswordCredentials, Protocol, RawClient, SendCommandOptions,
+ PasswordCredentials, RawClient, SendCommandOptions,
TokenCredentials
} from '../client/client.type.js';
-import { handleResponse } from './client.utils.js';
import { ResponseError, responseError } from '../wire/error.utils.js';
import { debug } from './client.debug.js';
import { IggyConnection } from './client.connection.js';
@@ -86,8 +85,6 @@ export class VsrResponseTimeoutError extends Error {
* Implements command queuing, authentication, and heartbeat functionality.
*/
export class CommandResponseStream extends EventEmitter {
- /** Server wire protocol used by this connection */
- readonly protocol: Protocol;
/** Client configuration */
private options: ClientConfig;
/** Underlying connection to the server */
@@ -120,7 +117,6 @@ export class CommandResponseStream extends EventEmitter {
super();
const normalizedConfig = normalizeClientConfig(options);
this.options = normalizedConfig;
- this.protocol = normalizedConfig.protocol;
this.connection = new IggyConnection(normalizedConfig);
this.busy = false;
this.isAuthenticated = false;
@@ -176,7 +172,7 @@ export class CommandResponseStream extends EventEmitter {
if (!this.connection.connected)
await this.connection.connect()
- if (this.options.protocol === 'vsr' && isLoginCommand(command))
+ if (isLoginCommand(command))
await this._ensureVsrLeader();
if (!this.isAuthenticated && !this.isUnloggedCommand(command))
@@ -247,8 +243,6 @@ export class CommandResponseStream extends EventEmitter {
payload: Buffer,
handleResp = true
): Promise<CommandResponse> {
- if (this.options.protocol !== 'vsr')
- return this._processClassic(command, payload, handleResp);
if (isLoginCommand(command) && this.isAuthenticated)
return this._processVsrLogin(command, payload, handleResp);
return this._processVsr(command, payload, handleResp);
@@ -263,22 +257,6 @@ export class CommandResponseStream extends EventEmitter {
return this._processVsr(command, payload, handleResp);
}
- private async _processClassic(
- command: number,
- payload: Buffer,
- handleResp: boolean
- ): Promise<CommandResponse> {
- const response = await this._exchange(
- () => this.connection.writeCommand(command, payload)
- );
- if (!handleResp)
- return response as unknown as CommandResponse;
- const parsed = handleResponse(response);
- if (parsed.status !== 0)
- throw responseError(command, parsed.status);
- return parsed;
- }
-
private async _processVsr(
command: number,
payload: Buffer,
@@ -409,8 +387,7 @@ export class CommandResponseStream extends EventEmitter {
private isUnloggedCommand(command: number): boolean {
return UNLOGGED_COMMAND_CODE.includes(command) ||
- (this.options.protocol === 'vsr' &&
- command === COMMAND_CODE.GetClusterMetadata);
+ command === COMMAND_CODE.GetClusterMetadata;
}
private async _ensureVsrLeader(): Promise<void> {
diff --git a/foreign/node/src/client/client.type.ts b/foreign/node/src/client/client.type.ts
index 080529891..a7e63723a 100644
--- a/foreign/node/src/client/client.type.ts
+++ b/foreign/node/src/client/client.type.ts
@@ -56,8 +56,6 @@ export type SendCommandOptions = {
* Provides direct access to command sending and event handling.
*/
export type RawClient = {
- /** Server wire protocol used by this connection */
- readonly protocol: Protocol,
/** Sends a command to the server and returns the response */
sendCommand: (
code: number,
@@ -114,9 +112,6 @@ export type ReconnectOption = {
*/
export type TransportOption = TcpOption | TlsOption;
-/** Server wire protocol. */
-export type Protocol = 'classic' | 'vsr';
-
/**
* Token-based authentication credentials.
*/
@@ -155,8 +150,6 @@ export type PoolSizeOption = {
* Complete client configuration for connecting to the Iggy server.
*/
export type ClientConfig = {
- /** Server wire protocol (default: classic) */
- protocol?: Protocol,
/** Transport protocol to use (TCP or TLS) */
transport: TransportType,
/** Transport-specific connection options */
diff --git a/foreign/node/src/client/client.utils.test.ts b/foreign/node/src/client/client.utils.test.ts
index 174f09cf5..815058dd3 100644
--- a/foreign/node/src/client/client.utils.test.ts
+++ b/foreign/node/src/client/client.utils.test.ts
@@ -18,32 +18,19 @@
import { describe, it } from 'node:test';
import assert from 'node:assert/strict';
-import { handleResponse, deserializeVoidResponse } from './client.utils.js';
+import { deserializeVoidResponse } from './client.utils.js';
-const SUCCESS = 0;
+describe('deserializeVoidResponse', () => {
-describe('handleResponse', () => {
+ it('returns true only for a success status with an empty payload', () => {
+ const success = { status: 0, length: 0, data: Buffer.alloc(0) };
+ assert.equal(deserializeVoidResponse(success), true);
- it('bounds data to the length field, not the full buffer', () => {
- // Server says: status=0, length=0, no payload.
- // But the raw buffer has 4 trailing bytes (e.g. start of next response).
- const buf = Buffer.alloc(12);
- buf.writeUInt32LE(SUCCESS, 0); // status
- buf.writeUInt32LE(0, 4); // length = 0 (void response)
- buf.writeUInt32LE(42, 8); // trailing bytes — NOT part of this response
+ const failed = { status: 1, length: 0, data: Buffer.alloc(0) };
+ assert.equal(deserializeVoidResponse(failed), false);
- const r = handleResponse(buf);
- assert.equal(r.data.length, 0);
- });
-
- it('deserializeVoidResponse returns true for a valid void response with trailing buffer bytes', () => {
- const buf = Buffer.alloc(12);
- buf.writeUInt32LE(SUCCESS, 0);
- buf.writeUInt32LE(0, 4); // length = 0
- buf.writeUInt32LE(42, 8); // trailing bytes
-
- const r = handleResponse(buf);
- assert.equal(deserializeVoidResponse(r), true);
+ const withData = { status: 0, length: 4, data: Buffer.alloc(4) };
+ assert.equal(deserializeVoidResponse(withData), false);
});
});
diff --git a/foreign/node/src/client/client.utils.ts b/foreign/node/src/client/client.utils.ts
index bb9a92258..dfde49b82 100644
--- a/foreign/node/src/client/client.utils.ts
+++ b/foreign/node/src/client/client.utils.ts
@@ -16,45 +16,7 @@
// under the License.
//
-import { Transform, type TransformCallback } from 'node:stream';
import type { CommandResponse } from './client.type.js';
-import { translateCommandCode } from '../wire/command.code.js';
-import { debug } from './client.debug.js';
-
-
-/**
- * Parses a raw response buffer into a structured CommandResponse.
- * Extracts status code, length, and payload data from the buffer.
- *
- * @param r - Raw response buffer from the server
- * @returns Parsed command response with status, length, and data
- */
-export const handleResponse = (r: Buffer) => {
- const status = r.readUint32LE(0);
- const length = r.readUint32LE(4);
- debug('<== handleResponse', { status, length });
- return {
- status, length, data: r.subarray(8, 8 + length)
- }
-};
-
-/**
- * Creates a Transform stream that parses response buffers.
- * Transforms raw server responses into just the data payload.
- *
- * @returns Transform stream for processing server responses
- */
-export const handleResponseTransform = () => new Transform({
- transform(chunk: Buffer, encoding: BufferEncoding, cb: TransformCallback) {
- try {
- const r = handleResponse(chunk);
- debug('response::', r)
- return cb(null, r.data);
- } catch (err: unknown) {
- return cb(new Error('handleResponseTransform error', { cause: err }), null);
- }
- }
-});
/**
* Deserializes a void response from the server.
@@ -65,33 +27,3 @@ export const handleResponseTransform = () => new Transform({
*/
export const deserializeVoidResponse =
(r: CommandResponse) => r.status === 0 && r.data.length === 0;
-
-/** Length of the command code in bytes */
-const COMMAND_LENGTH = 4;
-
-/**
- * Serializes a command and its payload into a buffer for sending to the server.
- * Creates the wire format: [payload_size (4 bytes)][command (4 bytes)][payload]
- *
- * @param command - Command code to send
- * @param payload - Command payload buffer
- * @returns Buffer ready to be sent to the server
- */
-export const serializeCommand = (command: number, payload: Buffer) => {
- const payloadSize = payload.length + COMMAND_LENGTH;
- const data = Buffer.allocUnsafe(8 + payload.length);
-
- data.writeUint32LE(payloadSize, 0);
- data.writeUint32LE(command, 4);
- data.fill(payload, 8);
-
- debug(
- '==> CMD', command,
- translateCommandCode(command),
- 'LENGTH', payloadSize
- );
-
- debug('FullMessage#Base64', data.toString('base64'));
-
- return data;
-}
diff --git a/foreign/node/src/e2e/tcp.cluster.e2e.ts b/foreign/node/src/e2e/tcp.cluster.e2e.ts
index 8fae9e93f..d3e5c85e4 100644
--- a/foreign/node/src/e2e/tcp.cluster.e2e.ts
+++ b/foreign/node/src/e2e/tcp.cluster.e2e.ts
@@ -20,13 +20,14 @@ import { after, describe, it } from "node:test";
import assert from "node:assert/strict";
import { getTestClient } from "./test-client.utils.js";
-// cluster mode still in dev atm
-// response is mocked from /core/server/config.toml
+// The suite runs against a single-node server (cluster mode off), which
+// reports itself as the sole healthy leader with endpoints derived from
+// its default config.
const expectedMeta = {
- name: "iggy-cluster",
+ name: "single-node",
nodes: [
{
- name: "iggy-node-1",
+ name: "iggy-node",
ip: "127.0.0.1",
endpoints: {
tcp: 8090,
@@ -37,18 +38,6 @@ const expectedMeta = {
role: "Leader",
status: "Healthy",
},
- {
- name: "iggy-node-2",
- ip: "127.0.0.1",
- endpoints: {
- tcp: 8091,
- quic: 8081,
- http: 3001,
- websocket: 8093,
- },
- role: "Follower",
- status: "Healthy",
- },
],
};
diff --git a/foreign/node/src/e2e/tcp.consumer-group.e2e.ts b/foreign/node/src/e2e/tcp.consumer-group.e2e.ts
index d0dd85497..58234fbb1 100644
--- a/foreign/node/src/e2e/tcp.consumer-group.e2e.ts
+++ b/foreign/node/src/e2e/tcp.consumer-group.e2e.ts
@@ -26,10 +26,8 @@ import { getIggyAddress } from '../tcp.sm.utils.js';
describe('e2e -> consumer-group', async () => {
const [host, port] = getIggyAddress();
- const vsr = process.env.IGGY_TEST_PROTOCOL === 'vsr';
const c = new SingleClient({
- protocol: vsr ? 'vsr' : 'classic',
transport: 'TCP',
options: { host, port },
credentials: { username: 'iggy', password: 'iggy' }
@@ -95,9 +93,7 @@ describe('e2e -> consumer-group', async () => {
streamId: streamName,
topicId: topicName,
messages: generateMessages(mn),
- partition: vsr
- ? Partitioning.PartitionId((i / mn) % 3)
- : Partitioning.MessageKey(`key-${ i % 300 }`)
+ partition: Partitioning.PartitionId((i / mn) % 3)
}));
}
payloadLength = ct;
diff --git a/foreign/node/src/e2e/tcp.consumer-stream.e2e.ts b/foreign/node/src/e2e/tcp.consumer-stream.e2e.ts
index e954e7e6a..40d9afe93 100644
--- a/foreign/node/src/e2e/tcp.consumer-stream.e2e.ts
+++ b/foreign/node/src/e2e/tcp.consumer-stream.e2e.ts
@@ -32,10 +32,8 @@ describe('e2e -> consumer-stream', async () => {
const [host, port] = getIggyAddress();
const credentials = { username: 'iggy', password: 'iggy' };
- const vsr = process.env.IGGY_TEST_PROTOCOL === 'vsr';
const opt = {
- protocol: (vsr ? 'vsr' : 'classic') as 'vsr' | 'classic',
transport: 'TCP' as const,
options: { host, port },
credentials
@@ -62,9 +60,7 @@ describe('e2e -> consumer-stream', async () => {
await sendSomeMessages(c.clientProvider)(
streamName,
topicName,
- vsr
- ? Partitioning.PartitionId((i / 100) % 3)
- : Partitioning.MessageKey(`k-${i % 300}`)
+ Partitioning.PartitionId((i / 100) % 3)
);
}
});
diff --git a/foreign/node/src/e2e/tcp.send-message.e2e.ts b/foreign/node/src/e2e/tcp.send-message.e2e.ts
index e085e1612..8b170c7ba 100644
--- a/foreign/node/src/e2e/tcp.send-message.e2e.ts
+++ b/foreign/node/src/e2e/tcp.send-message.e2e.ts
@@ -27,10 +27,6 @@ describe('e2e -> message', async () => {
const c = getTestClient();
- // Only the VSR lane reaches a server that reports offsets. The classic lane
- // runs against the legacy server, which commits without confirming.
- const vsr = process.env.IGGY_TEST_PROTOCOL === 'vsr';
-
const streamName = 'e2e-stream-934';
const topicName = 'e2e-topic-832';
const partitionId = 0;
@@ -53,10 +49,6 @@ describe('e2e -> message', async () => {
it('e2e -> message::send', async () => {
const { confirmations } = await c.message.send(msg);
- if (!vsr) {
- assert.equal(confirmations.length, 0);
- return;
- }
assert.equal(confirmations.length, 1);
assert.equal(confirmations[0].partitionId, partitionId);
});
@@ -177,10 +169,6 @@ describe('e2e -> message', async () => {
...msg,
messages: generateMessages(3)
});
- if (!vsr) {
- assert.equal(confirmations.length, 0);
- return;
- }
// Landing behind the already committed batch is the part no placeholder
// confirmation could reproduce.
assert.equal(confirmations.length, 1);
diff --git a/foreign/node/src/e2e/test-client.utils.ts b/foreign/node/src/e2e/test-client.utils.ts
index b5d2d7713..22090066e 100644
--- a/foreign/node/src/e2e/test-client.utils.ts
+++ b/foreign/node/src/e2e/test-client.utils.ts
@@ -21,10 +21,8 @@ import { getIggyAddress } from '../tcp.sm.utils.js';
const credentials = { username: 'iggy', password: 'iggy' };
const [host, port] = getIggyAddress();
-const protocol = process.env.IGGY_TEST_PROTOCOL === 'vsr' ? 'vsr' : 'classic';
export const getTestClient = () => new Client({
- protocol,
transport: 'TCP',
options: { host, port },
credentials
diff --git a/foreign/node/src/e2e/tls.system.e2e.ts b/foreign/node/src/e2e/tls.system.e2e.ts
index c2f7b2d9b..c3af4878d 100644
--- a/foreign/node/src/e2e/tls.system.e2e.ts
+++ b/foreign/node/src/e2e/tls.system.e2e.ts
@@ -22,11 +22,13 @@
// To run them locally:
//
// 1. Start the server with TLS:
+// TODO(hubcio): change to iggy-server once legacy server is removed
+// (core/server has VSR support)
// IGGY_ROOT_USERNAME=iggy IGGY_ROOT_PASSWORD=iggy \
// IGGY_TCP_TLS_ENABLED=true \
// IGGY_TCP_TLS_CERT_FILE=core/certs/iggy_cert.pem \
// IGGY_TCP_TLS_KEY_FILE=core/certs/iggy_key.pem \
-// cargo r --bin iggy-server
+// cargo r --bin iggy-server-ng --features vsr
//
// 2. Run the tests:
// cd foreign/node
@@ -51,14 +53,10 @@ const caCertPath = process.env.E2E_ROOT_CA_CERT
const getTlsClient = () => {
const [, port] = getIggyAddress();
const caCert = readFileSync(caCertPath);
- const protocol = process.env.IGGY_TEST_PROTOCOL === 'vsr'
- ? 'vsr'
- : 'classic';
// The server certificate SAN is DNS:localhost, so we connect via 'localhost'
// for proper hostname verification (consistent with Python and C# TLS tests).
return new Client({
- protocol,
transport: 'TLS',
options: {
port,
diff --git a/foreign/node/src/wire/command-set.test.ts b/foreign/node/src/wire/command-set.test.ts
index 3c1cf6f75..dbc3a78c9 100644
--- a/foreign/node/src/wire/command-set.test.ts
+++ b/foreign/node/src/wire/command-set.test.ts
@@ -23,7 +23,6 @@ import type { RawClient } from '../client/client.type.js';
import { COMMAND_CODE } from './command.code.js';
const mockRawClient = (): RawClient => ({
- protocol: 'classic',
sendCommand: async () => {
throw new Error('sendCommand should not be called by the session-control guard');
},
diff --git a/foreign/node/src/wire/message/poll-messages.command.ts b/foreign/node/src/wire/message/poll-messages.command.ts
index 335e3718e..4023f387e 100644
--- a/foreign/node/src/wire/message/poll-messages.command.ts
+++ b/foreign/node/src/wire/message/poll-messages.command.ts
@@ -247,8 +247,7 @@ export const pollMessages = (getClient: ClientProvider) =>
const client = await getClient();
const release = client.hold?.();
try {
- if (client.protocol === 'vsr' &&
- request.consumer.kind === ConsumerKind.Group &&
+ if (request.consumer.kind === ConsumerKind.Group &&
request.partitionId === null) {
const state = getGroupState(client);
while (true) {
diff --git a/foreign/node/src/wire/vsr/namespace.ts b/foreign/node/src/wire/vsr/namespace.ts
index 1a6ec20e9..1aa8cbcb1 100644
--- a/foreign/node/src/wire/vsr/namespace.ts
+++ b/foreign/node/src/wire/vsr/namespace.ts
@@ -142,6 +142,8 @@ const namespaceFromIds = (
* `SendMessages`: `[metadata_len u32][stream ident][topic ident]
* [partitioning kind u8, len u8, value]...`. Only explicit `PartitionId`
* partitioning is routable under VSR; the broker never picks a partition.
+ * TODO(hubcio): Balanced and MessageKey partitioning to be implemented;
+ * not decided yet whether it'll be on server side or client side.
*/
const namespaceFromSendMessages = (payload: Buffer): bigint => {
if (payload.length < 4)
diff --git a/scripts/run-examples-from-readme.sh b/scripts/run-examples-from-readme.sh
index 639e912e7..09933d408 100755
--- a/scripts/run-examples-from-readme.sh
+++ b/scripts/run-examples-from-readme.sh
@@ -217,7 +217,10 @@ run_rust_examples() {
# shellcheck disable=SC2329
run_node_examples() {
- resolve_server_binary "${TARGET}"
+ # The Node SDK is vsr-only, so examples run against the vsr server.
+ # TODO(hubcio): change to iggy-server once legacy server is removed
+ # (core/server has VSR support)
+ resolve_server_binary "${TARGET}" iggy-server-ng
export DEBUG=iggy:examples
unset -f TRANSFORM_COMMAND 2>/dev/null || true