This is an automated email from the ASF dual-hosted git repository. spetz pushed a commit to branch sdk_raw_command in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 9eb2c524d9cd414dfd66412c252f16e61e1ed74d Author: spetz <[email protected]> AuthorDate: Fri Jul 17 15:19:50 2026 +0200 feat(sdk): expose raw command API across foreign clients --- .../step_definitions/raw_command_steps.cpp | 65 +++ bdd/cpp/features/step_definitions/world.hpp | 2 + bdd/cpp/scripts/entrypoint.sh | 8 +- bdd/docker-compose.yml | 56 ++- bdd/go/tests/raw_command.go | 126 ++++++ bdd/go/tests/suite_test.go | 10 + .../org/apache/iggy/bdd/BasicMessagingSteps.java | 36 ++ .../test/java/org/apache/iggy/bdd/TestContext.java | 2 + bdd/node/Dockerfile | 1 + bdd/php/tests/BasicMessagingFeatureTest.php | 88 +--- bdd/php/tests/RawCommandFeatureTest.php | 102 +++++ bdd/php/tests/SharedFeatureParser.php | 94 +++++ bdd/python/tests/conftest.py | 2 + bdd/python/tests/test_raw_command.py | 82 ++++ bdd/rust/Cargo.toml | 5 + bdd/rust/tests/common/global_context.rs | 4 +- .../version.go => bdd/rust/tests/raw_command.rs | 12 +- bdd/rust/tests/steps/mod.rs | 1 + bdd/rust/tests/steps/raw_command.rs | 60 +++ bdd/scenarios/raw_command.feature | 41 ++ foreign/cpp/Cargo.toml | 2 +- foreign/cpp/MODULE.bazel | 2 +- foreign/cpp/src/client.rs | 14 + foreign/cpp/src/lib.rs | 1 + foreign/cpp/tests/client/low_level_e2e.cpp | 42 ++ .../Iggy_SDK.Tests.BDD/Context/TestContext.cs | 2 + .../csharp/Iggy_SDK.Tests.BDD/Context/TestHooks.cs | 2 + .../Iggy_SDK.Tests.BDD/Iggy_SDK.Tests.BDD.csproj | 3 + .../StepDefinitions/RawCommandSteps.cs | 71 ++++ .../Iggy_SDK.Tests.Integration/RawCommandTests.cs | 83 ++++ foreign/csharp/Iggy_SDK/IggyClient/IIggyClient.cs | 13 + .../Implementations/HttpMessageStream.cs | 13 + .../IggyClient/Implementations/TcpMessageStream.cs | 28 ++ foreign/csharp/Iggy_SDK/Iggy_SDK.csproj | 2 +- foreign/csharp/Iggy_SDK/Utils/CommandCodes.cs | 2 + foreign/go/client/tcp/tcp_core.go | 32 ++ foreign/go/client/tcp/tcp_core_test.go | 104 +++++ foreign/go/contracts/client.go | 4 + foreign/go/contracts/version.go | 2 +- foreign/go/internal/command/code.go | 2 + foreign/java/gradle.properties | 2 +- .../iggy/client/async/tcp/AsyncIggyTcpClient.java | 45 ++ .../iggy/client/blocking/IggyBaseClient.java | 12 + .../iggy/client/blocking/http/IggyHttpClient.java | 7 + .../iggy/client/blocking/tcp/IggyTcpClient.java | 6 + .../java/org/apache/iggy/serde/CommandCode.java | 6 +- .../blocking/http/RawCommandHttpClientTest.java} | 27 +- .../blocking/tcp/RawCommandTcpClientTest.java | 67 +++ foreign/node/package-lock.json | 4 +- foreign/node/package.json | 2 +- foreign/node/src/bdd/raw.ts | 48 +++ foreign/node/src/bdd/world.ts | 4 +- foreign/node/src/e2e/tcp.raw.e2e.ts | 53 +++ foreign/node/src/wire/command-set.test.ts | 98 +++++ foreign/node/src/wire/command-set.ts | 30 +- foreign/node/src/wire/command.code.ts | 2 + foreign/php/Cargo.toml | 2 +- foreign/php/README.md | 2 + foreign/php/iggy-php.stubs.php | 10 + foreign/php/src/client.rs | 17 +- foreign/php/tests/RawCommandTest.php | 77 ++++ foreign/python/Cargo.toml | 2 +- foreign/python/apache_iggy.pyi | 19 + foreign/python/pyproject.toml | 2 +- foreign/python/src/client.rs | 34 +- foreign/python/tests/test_raw_command.py | 47 +++ foreign/python/uv.lock | 2 +- raw-command-iggy-sdk.md | 461 +++++++++++++++++++++ scripts/run-bdd-tests.sh | 6 +- 69 files changed, 2186 insertions(+), 129 deletions(-) diff --git a/bdd/cpp/features/step_definitions/raw_command_steps.cpp b/bdd/cpp/features/step_definitions/raw_command_steps.cpp new file mode 100644 index 000000000..d960d0f11 --- /dev/null +++ b/bdd/cpp/features/step_definitions/raw_command_steps.cpp @@ -0,0 +1,65 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#define CUKE_OBJECT_PREFIX IggyBddRawCommand + +#include <gtest/gtest.h> + +#include <cucumber-cpp/autodetect.hpp> + +#include <cstdint> +#include <exception> +#include <string> + +#include "world.hpp" + +WHEN("^I send a raw command with code ([0-9]+) and an empty payload$") { + REGEX_PARAM(std::uint32_t, code); + cucumber::ScenarioScope<bdd::GlobalContext> context; + + context->raw_response.clear(); + context->raw_error.clear(); + try { + const auto response = context->client->send_raw_with_response(code, rust::Vec<std::uint8_t>()); + context->raw_response.assign(response.begin(), response.end()); + } catch (const std::exception &error) { + context->raw_error = error.what(); + } +} + +THEN("^the raw command should succeed with an empty response$") { + cucumber::ScenarioScope<bdd::GlobalContext> context; + + EXPECT_TRUE(context->raw_error.empty()); + EXPECT_TRUE(context->raw_response.empty()); +} + +THEN("^the raw command should succeed with a non-empty response$") { + cucumber::ScenarioScope<bdd::GlobalContext> context; + + EXPECT_TRUE(context->raw_error.empty()); + EXPECT_FALSE(context->raw_response.empty()); +} + +THEN("^the raw command should fail with an invalid command error$") { + cucumber::ScenarioScope<bdd::GlobalContext> context; + + EXPECT_TRUE(context->raw_response.empty()); + EXPECT_NE(context->raw_error.find("Invalid command"), std::string::npos); +} diff --git a/bdd/cpp/features/step_definitions/world.hpp b/bdd/cpp/features/step_definitions/world.hpp index 470c0661e..7b8fdc487 100644 --- a/bdd/cpp/features/step_definitions/world.hpp +++ b/bdd/cpp/features/step_definitions/world.hpp @@ -45,6 +45,8 @@ struct GlobalContext { std::uint64_t last_sent_id_lo = 0; std::string last_sent_payload; PolledData polled; + std::vector<std::uint8_t> raw_response; + std::string raw_error; GlobalContext() = default; GlobalContext(const GlobalContext &) = delete; diff --git a/bdd/cpp/scripts/entrypoint.sh b/bdd/cpp/scripts/entrypoint.sh index b41b1b592..b9cf1b794 100755 --- a/bdd/cpp/scripts/entrypoint.sh +++ b/bdd/cpp/scripts/entrypoint.sh @@ -54,10 +54,16 @@ fi cd /workspace/bdd/cpp || exit 1 +tags=() +case "${BDD_FEATURE:-all}" in + basic_messaging) tags=(--tags @basic-messaging) ;; + raw_command) tags=(--tags @raw-command) ;; +esac + # --strict fails the run on undefined steps; without it a feature step with no matching C++ # definition would be reported and still exit 0. set +e -bundle exec cucumber --strict +bundle exec cucumber --strict "${tags[@]}" status=$? set -e diff --git a/bdd/docker-compose.yml b/bdd/docker-compose.yml index 0c6076634..2b9c7b23f 100644 --- a/bdd/docker-compose.yml +++ b/bdd/docker-compose.yml @@ -30,6 +30,7 @@ services: volumes: - ./scenarios/basic_messaging.feature:/app/features/basic_messaging.feature - ./scenarios/leader_redirection.feature:/app/features/leader_redirection.feature + - ./scenarios/raw_command.feature:/app/features/raw_command.feature command: - sh - -c @@ -37,6 +38,7 @@ services: case "$$BDD_FEATURE" in basic_messaging) cargo test -p bdd --features bdd --test basic_messaging ;; leader_redirection) cargo test -p bdd --features bdd --test leader_redirection ;; + raw_command) cargo test -p bdd --features bdd --test raw_command ;; *) cargo test -p bdd --features bdd ;; esac networks: @@ -50,10 +52,21 @@ services: environment: - IGGY_ROOT_USERNAME=iggy - IGGY_ROOT_PASSWORD=iggy + - BDD_FEATURE=${BDD_FEATURE:-all} volumes: - ./scenarios/basic_messaging.feature:/app/features/basic_messaging.feature + - ./scenarios/raw_command.feature:/app/features/raw_command.feature working_dir: /app - command: [ "uv", "run", "--project", "/workspace/bdd/python", "pytest", "tests/", "-v" ] + command: + - sh + - -c + - | + case "$$BDD_FEATURE" in + basic_messaging) TEST_PATH='tests/test_basic_messaging.py' ;; + raw_command) TEST_PATH='tests/test_raw_command.py' ;; + *) TEST_PATH='tests/' ;; + esac + uv run --project /workspace/bdd/python pytest $$TEST_PATH -v networks: - iggy-bdd-network @@ -69,10 +82,20 @@ services: - IGGY_PORT=8090 - IGGY_USERNAME=iggy - IGGY_PASSWORD=iggy - - BDD_FEATURE_FILE=/app/features/basic_messaging.feature + - BDD_FEATURE=${BDD_FEATURE:-all} volumes: - ./scenarios/basic_messaging.feature:/app/features/basic_messaging.feature - command: [ "./scripts/test.sh", "--configuration", "/workspace/bdd/php/phpunit.xml.dist" ] + - ./scenarios/raw_command.feature:/app/features/raw_command.feature + command: + - sh + - -c + - | + case "$$BDD_FEATURE" in + basic_messaging) GROUP='--group basic-messaging' ;; + raw_command) GROUP='--group raw-command' ;; + *) GROUP='' ;; + esac + ./scripts/test.sh --configuration /workspace/bdd/php/phpunit.xml.dist $$GROUP networks: - iggy-bdd-network @@ -98,7 +121,17 @@ services: environment: - IGGY_ROOT_USERNAME=iggy - IGGY_ROOT_PASSWORD=iggy - command: [ "npm", "run", "test:bdd" ] + - BDD_FEATURE=${BDD_FEATURE:-all} + command: + - sh + - -c + - | + case "$$BDD_FEATURE" in + basic_messaging) TAGS='--tags @basic-messaging' ;; + raw_command) TAGS='--tags @raw-command' ;; + *) TAGS='' ;; + esac + npm run test:bdd -- $$TAGS networks: - iggy-bdd-network @@ -119,6 +152,7 @@ services: case "$$BDD_FEATURE" in basic_messaging) FILTER='--filter-trait Category=basic-messaging' ;; leader_redirection) FILTER='--filter-trait Category=requires-leader-awareness' ;; + raw_command) FILTER='--filter-trait Category=raw-command' ;; esac dotnet test $$FILTER networks: @@ -132,9 +166,19 @@ services: environment: - IGGY_ROOT_USERNAME=iggy - IGGY_ROOT_PASSWORD=iggy + - BDD_FEATURE=${BDD_FEATURE:-all} volumes: - ./scenarios/basic_messaging.feature:/app/features/basic_messaging.feature - command: [ "gradle", "--no-daemon", "test" ] + - ./scenarios/raw_command.feature:/app/features/raw_command.feature + command: + - sh + - -c + - | + case "$$BDD_FEATURE" in + basic_messaging) export CUCUMBER_FILTER_TAGS='@basic-messaging' ;; + raw_command) export CUCUMBER_FILTER_TAGS='@raw-command' ;; + esac + gradle --no-daemon test networks: - iggy-bdd-network @@ -146,8 +190,10 @@ services: environment: - IGGY_ROOT_USERNAME=iggy - IGGY_ROOT_PASSWORD=iggy + - BDD_FEATURE=${BDD_FEATURE:-all} volumes: - ./scenarios/basic_messaging.feature:/app/features/basic_messaging.feature + - ./scenarios/raw_command.feature:/app/features/raw_command.feature networks: - iggy-bdd-network diff --git a/bdd/go/tests/raw_command.go b/bdd/go/tests/raw_command.go new file mode 100644 index 000000000..78e251be8 --- /dev/null +++ b/bdd/go/tests/raw_command.go @@ -0,0 +1,126 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package tests + +import ( + "context" + "errors" + "fmt" + "os" + + "github.com/apache/iggy/foreign/go/client" + "github.com/apache/iggy/foreign/go/client/tcp" + iggcon "github.com/apache/iggy/foreign/go/contracts" + ierror "github.com/apache/iggy/foreign/go/errors" + "github.com/cucumber/godog" +) + +type rawCommandCtxKey struct{} + +type rawCommandCtx struct { + client iggcon.Client + serverAddr string + lastResponse []byte + lastError error +} + +func getRawCommandCtx(ctx context.Context) *rawCommandCtx { + return ctx.Value(rawCommandCtxKey{}).(*rawCommandCtx) +} + +type rawCommandSteps struct{} + +func (rawCommandSteps) givenRunningServer(ctx context.Context) error { + address := os.Getenv("IGGY_TCP_ADDRESS") + if address == "" { + address = "127.0.0.1:8090" + } + getRawCommandCtx(ctx).serverAddr = address + return nil +} + +func (rawCommandSteps) givenAuthenticationAsRoot(ctx context.Context) error { + state := getRawCommandCtx(ctx) + iggyClient, err := client.NewIggyClient(client.WithTcp(tcp.WithServerAddress(state.serverAddr))) + if err != nil { + return fmt.Errorf("create client: %w", err) + } + if err = iggyClient.Connect(ctx); err != nil { + return fmt.Errorf("connect client: %w", err) + } + if _, err = iggyClient.LoginUser(ctx, "iggy", "iggy"); err != nil { + return fmt.Errorf("authenticate client: %w", err) + } + state.client = iggyClient + return nil +} + +func (rawCommandSteps) whenSendRawCommand(ctx context.Context, code uint32) error { + state := getRawCommandCtx(ctx) + state.lastResponse, state.lastError = state.client.SendRawWithResponse(ctx, code, nil) + return nil +} + +func (rawCommandSteps) thenEmptyResponse(ctx context.Context) error { + state := getRawCommandCtx(ctx) + if state.lastError != nil { + return fmt.Errorf("raw command failed: %w", state.lastError) + } + if len(state.lastResponse) != 0 { + return fmt.Errorf("expected empty response, got %d bytes", len(state.lastResponse)) + } + return nil +} + +func (rawCommandSteps) thenNonEmptyResponse(ctx context.Context) error { + state := getRawCommandCtx(ctx) + if state.lastError != nil { + return fmt.Errorf("raw command failed: %w", state.lastError) + } + if len(state.lastResponse) == 0 { + return errors.New("expected non-empty response") + } + return nil +} + +func (rawCommandSteps) thenInvalidCommand(ctx context.Context) error { + if !errors.Is(getRawCommandCtx(ctx).lastError, ierror.ErrInvalidCommand) { + return fmt.Errorf("expected invalid command error, got %v", getRawCommandCtx(ctx).lastError) + } + return nil +} + +func initRawCommandScenario(sc *godog.ScenarioContext) { + sc.Before(func(context.Context, *godog.Scenario) (context.Context, error) { + return context.WithValue(context.Background(), rawCommandCtxKey{}, &rawCommandCtx{}), nil + }) + steps := rawCommandSteps{} + sc.Step(`I have a running Iggy server`, steps.givenRunningServer) + sc.Step(`I am authenticated as the root user`, steps.givenAuthenticationAsRoot) + sc.Step(`^I send a raw command with code (\d+) and an empty payload$`, steps.whenSendRawCommand) + sc.Step(`the raw command should succeed with an empty response`, steps.thenEmptyResponse) + sc.Step(`the raw command should succeed with a non-empty response`, steps.thenNonEmptyResponse) + sc.Step(`the raw command should fail with an invalid command error`, steps.thenInvalidCommand) + sc.After(func(ctx context.Context, _ *godog.Scenario, scenarioError error) (context.Context, error) { + state := getRawCommandCtx(ctx) + if state.client != nil { + scenarioError = errors.Join(scenarioError, state.client.Close()) + } + return ctx, scenarioError + }) +} diff --git a/bdd/go/tests/suite_test.go b/bdd/go/tests/suite_test.go index 15d951b44..3aca7b078 100644 --- a/bdd/go/tests/suite_test.go +++ b/bdd/go/tests/suite_test.go @@ -51,6 +51,16 @@ func TestFeatures(t *testing.T) { }, }) } + if feature == "all" || feature == "raw_command" { + suites = append(suites, godog.TestSuite{ + ScenarioInitializer: initRawCommandScenario, + Options: &godog.Options{ + Format: "pretty", + Paths: []string{"../../scenarios/raw_command.feature"}, + TestingT: t, + }, + }) + } if len(suites) == 0 { t.Fatalf("unknown BDD_FEATURE=%q", feature) diff --git a/bdd/java/src/test/java/org/apache/iggy/bdd/BasicMessagingSteps.java b/bdd/java/src/test/java/org/apache/iggy/bdd/BasicMessagingSteps.java index 0ad53dad0..fde21676b 100644 --- a/bdd/java/src/test/java/org/apache/iggy/bdd/BasicMessagingSteps.java +++ b/bdd/java/src/test/java/org/apache/iggy/bdd/BasicMessagingSteps.java @@ -24,6 +24,8 @@ import io.cucumber.java.en.Then; import io.cucumber.java.en.When; import org.apache.iggy.client.blocking.IggyBaseClient; import org.apache.iggy.client.blocking.tcp.IggyTcpClient; +import org.apache.iggy.exception.IggyErrorCode; +import org.apache.iggy.exception.IggyServerException; import org.apache.iggy.message.Message; import org.apache.iggy.message.Partitioning; import org.apache.iggy.message.PollingStrategy; @@ -43,6 +45,7 @@ import java.util.Optional; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; public class BasicMessagingSteps { @@ -209,6 +212,39 @@ public class BasicMessagingSteps { assertEquals(context.lastSentMessage, lastPayload, "Last message should match sent message"); } + @When("I send a raw command with code {int} and an empty payload") + public void sendRawCommand(int code) { + try { + context.lastRawResponse = getClient().sendRawWithResponse(code, new byte[0]); + context.lastRawError = null; + } catch (RuntimeException error) { + context.lastRawResponse = null; + context.lastRawError = error; + } + } + + @Then("the raw command should succeed with an empty response") + public void rawCommandSucceedsWithEmptyResponse() { + assertNull(context.lastRawError, "Raw command should succeed"); + assertNotNull(context.lastRawResponse, "Raw response should be present"); + assertEquals(0, context.lastRawResponse.length, "Raw response should be empty"); + } + + @Then("the raw command should succeed with a non-empty response") + public void rawCommandSucceedsWithNonEmptyResponse() { + assertNull(context.lastRawError, "Raw command should succeed"); + assertNotNull(context.lastRawResponse, "Raw response should be present"); + assertTrue(context.lastRawResponse.length > 0, "Raw response should not be empty"); + } + + @Then("the raw command should fail with an invalid command error") + public void rawCommandFailsWithInvalidCommand() { + assertNull(context.lastRawResponse, "Raw response should be absent"); + assertTrue(context.lastRawError instanceof IggyServerException, "Expected an Iggy server error"); + IggyServerException error = (IggyServerException) context.lastRawError; + assertEquals(IggyErrorCode.INVALID_COMMAND, error.getErrorCode()); + } + private IggyBaseClient getClient() { if (context.client == null) { throw new IllegalStateException("Iggy client not initialized"); diff --git a/bdd/java/src/test/java/org/apache/iggy/bdd/TestContext.java b/bdd/java/src/test/java/org/apache/iggy/bdd/TestContext.java index e87983250..13ea24b94 100644 --- a/bdd/java/src/test/java/org/apache/iggy/bdd/TestContext.java +++ b/bdd/java/src/test/java/org/apache/iggy/bdd/TestContext.java @@ -32,4 +32,6 @@ class TestContext { Long lastTopicPartitions; PolledMessages lastPolledMessages; String lastSentMessage; + byte[] lastRawResponse; + RuntimeException lastRawError; } diff --git a/bdd/node/Dockerfile b/bdd/node/Dockerfile index 4c83e8c02..26eb5043e 100644 --- a/bdd/node/Dockerfile +++ b/bdd/node/Dockerfile @@ -31,6 +31,7 @@ FROM node:26-slim COPY ./foreign/node . COPY ./bdd/scenarios/basic_messaging.feature ./bdd/ +COPY ./bdd/scenarios/raw_command.feature ./bdd/ RUN npm ci diff --git a/bdd/php/tests/BasicMessagingFeatureTest.php b/bdd/php/tests/BasicMessagingFeatureTest.php index b4d4c84bc..d47b09f60 100644 --- a/bdd/php/tests/BasicMessagingFeatureTest.php +++ b/bdd/php/tests/BasicMessagingFeatureTest.php @@ -18,10 +18,13 @@ declare(strict_types=1); +require_once __DIR__ . '/SharedFeatureParser.php'; + use Iggy\Client as IggyClient; use Iggy\PollingStrategy; use Iggy\SendMessage; use PHPUnit\Framework\Attributes\DataProvider; +use PHPUnit\Framework\Attributes\Group; use PHPUnit\Framework\Attributes\TestDox; use PHPUnit\Framework\TestCase; @@ -36,6 +39,7 @@ final class BasicMessagingFeatureTest extends TestCase private ?int $lastSentMessageCount = null; #[DataProvider('scenarioCases')] + #[Group('basic-messaging')] #[TestDox('Basic messaging shared BDD scenario passes for the PHP SDK')] public function testBasicMessagingScenario(string $scenarioName, array $steps): void { @@ -61,89 +65,7 @@ final class BasicMessagingFeatureTest extends TestCase public static function scenarioCases(): array { - $featureFile = getenv('BDD_FEATURE_FILE') ?: __DIR__ . '/../../scenarios/basic_messaging.feature'; - if (!is_file($featureFile)) { - throw new RuntimeException("feature file not found at {$featureFile}"); - } - - $lines = file($featureFile, FILE_IGNORE_NEW_LINES); - if ($lines === false) { - throw new RuntimeException("failed to read feature file at {$featureFile}"); - } - - $backgroundSteps = []; - $scenarios = []; - $currentScenario = null; - $currentSteps = []; - $section = null; - $stepPattern = '/^(Given|When|Then|And|But|\*) (.+)$/'; - $unsupportedStructurePattern = '/^(Scenario Outline|Rule|Examples):|^\|/'; - foreach ($lines as $line) { - $line = trim($line); - if ($line === '' || str_starts_with($line, '#') || str_starts_with($line, '@')) { - continue; - } - - if ($line === 'Background:') { - $section = 'background'; - continue; - } - - if (str_starts_with($line, 'Scenario:')) { - if ($currentScenario !== null) { - $scenarios[$currentScenario] = $currentSteps; - } - - $currentScenario = trim(substr($line, strlen('Scenario:'))); - $currentSteps = []; - $section = 'scenario'; - continue; - } - - if (preg_match($unsupportedStructurePattern, $line) === 1) { - throw new RuntimeException("Unsupported BDD structure: {$line}"); - } - - if (preg_match($stepPattern, $line, $matches) !== 1) { - continue; - } - - if ($section === null) { - throw new RuntimeException("BDD step appears before Background or Scenario: {$line}"); - } - - if ($section === 'background') { - $backgroundSteps[] = $matches[2]; - continue; - } - - if ($section === 'scenario' && $currentScenario !== null) { - $currentSteps[] = $matches[2]; - } - } - - if ($currentScenario !== null) { - $scenarios[$currentScenario] = $currentSteps; - } - - if ($backgroundSteps === []) { - throw new RuntimeException('no BDD background steps were loaded from the feature'); - } - - if ($scenarios === []) { - throw new RuntimeException('no BDD scenarios were loaded from the feature'); - } - - $cases = []; - foreach ($scenarios as $scenarioName => $scenarioSteps) { - if ($scenarioSteps === []) { - throw new RuntimeException("scenario has no steps: {$scenarioName}"); - } - - $cases[$scenarioName] = [$scenarioName, [...$backgroundSteps, ...$scenarioSteps]]; - } - - return $cases; + return SharedFeatureParser::load(__DIR__ . '/../../scenarios/basic_messaging.feature'); } private function runStep(string $step): void diff --git a/bdd/php/tests/RawCommandFeatureTest.php b/bdd/php/tests/RawCommandFeatureTest.php new file mode 100644 index 000000000..a6dd1f3eb --- /dev/null +++ b/bdd/php/tests/RawCommandFeatureTest.php @@ -0,0 +1,102 @@ +<?php +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +declare(strict_types=1); + +require_once __DIR__ . '/SharedFeatureParser.php'; + +use Iggy\Client as IggyClient; +use Iggy\Exception\IggyException; +use PHPUnit\Framework\Attributes\DataProvider; +use PHPUnit\Framework\Attributes\Group; +use PHPUnit\Framework\Attributes\TestDox; +use PHPUnit\Framework\TestCase; + +final class RawCommandFeatureTest extends TestCase +{ + private ?IggyClient $client = null; + private ?string $lastResponse = null; + private ?Throwable $lastError = null; + + #[DataProvider('scenarioCases')] + #[Group('raw-command')] + #[TestDox('Raw command shared BDD scenario passes for the PHP SDK')] + public function testRawCommandScenario(string $scenarioName, array $steps): void + { + assert_true($scenarioName !== '', 'scenario name must not be empty'); + foreach ($steps as $step) { + $this->runStep($step); + } + } + + public static function scenarioCases(): array + { + return SharedFeatureParser::load(__DIR__ . '/../../scenarios/raw_command.feature'); + } + + private function runStep(string $step): void + { + if ($step === 'I have a running Iggy server') { + $this->client = new IggyClient(server_host() . ':' . server_port()); + $this->client->connect(); + $this->client->ping(); + return; + } + if ($step === 'I am authenticated as the root user') { + $this->requireClient()->loginUser( + env_or_default('IGGY_USERNAME', 'iggy'), + env_or_default('IGGY_PASSWORD', 'iggy'), + ); + return; + } + if (preg_match('/^I send a raw command with code (\d+) and an empty payload$/', $step, $matches) === 1) { + try { + $this->lastResponse = $this->requireClient()->sendRawWithResponse((int) $matches[1], ''); + $this->lastError = null; + } catch (Throwable $error) { + $this->lastResponse = null; + $this->lastError = $error; + } + return; + } + if ($step === 'the raw command should succeed with an empty response') { + assert_same(null, $this->lastError); + assert_same('', $this->lastResponse); + return; + } + if ($step === 'the raw command should succeed with a non-empty response') { + assert_same(null, $this->lastError); + assert_true($this->lastResponse !== null && $this->lastResponse !== ''); + return; + } + if ($step === 'the raw command should fail with an invalid command error') { + assert_same(null, $this->lastResponse); + assert_instance_of(IggyException::class, $this->lastError); + assert_true(str_contains(strtolower((string) $this->lastError?->getMessage()), 'invalid command')); + return; + } + + self::fail("Unsupported BDD step: {$step}"); + } + + private function requireClient(): IggyClient + { + assert_not_null($this->client, 'BDD client has not been initialized'); + return $this->client; + } +} diff --git a/bdd/php/tests/SharedFeatureParser.php b/bdd/php/tests/SharedFeatureParser.php new file mode 100644 index 000000000..38a8d8c5e --- /dev/null +++ b/bdd/php/tests/SharedFeatureParser.php @@ -0,0 +1,94 @@ +<?php +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +declare(strict_types=1); + +final class SharedFeatureParser +{ + public static function load(string $featureFile): array + { + if (!is_file($featureFile)) { + throw new RuntimeException("feature file not found at {$featureFile}"); + } + + $lines = file($featureFile, FILE_IGNORE_NEW_LINES); + if ($lines === false) { + throw new RuntimeException("failed to read feature file at {$featureFile}"); + } + + $backgroundSteps = []; + $scenarios = []; + $currentScenario = null; + $currentSteps = []; + $section = null; + foreach ($lines as $line) { + $line = trim($line); + if ($line === '' || str_starts_with($line, '#') || str_starts_with($line, '@')) { + continue; + } + + if ($line === 'Background:') { + $section = 'background'; + continue; + } + + if (str_starts_with($line, 'Scenario:')) { + if ($currentScenario !== null) { + $scenarios[$currentScenario] = $currentSteps; + } + $currentScenario = trim(substr($line, strlen('Scenario:'))); + $currentSteps = []; + $section = 'scenario'; + continue; + } + + if (preg_match('/^(Scenario Outline|Rule|Examples):|^\|/', $line) === 1) { + throw new RuntimeException("Unsupported BDD structure: {$line}"); + } + + if (preg_match('/^(Given|When|Then|And|But|\*) (.+)$/', $line, $matches) !== 1) { + continue; + } + if ($section === null) { + throw new RuntimeException("BDD step appears before Background or Scenario: {$line}"); + } + if ($section === 'background') { + $backgroundSteps[] = $matches[2]; + } elseif ($currentScenario !== null) { + $currentSteps[] = $matches[2]; + } + } + + if ($currentScenario !== null) { + $scenarios[$currentScenario] = $currentSteps; + } + if ($backgroundSteps === [] || $scenarios === []) { + throw new RuntimeException("feature must contain a background and at least one scenario: {$featureFile}"); + } + + $cases = []; + foreach ($scenarios as $scenarioName => $scenarioSteps) { + if ($scenarioSteps === []) { + throw new RuntimeException("scenario has no steps: {$scenarioName}"); + } + $cases[$scenarioName] = [$scenarioName, [...$backgroundSteps, ...$scenarioSteps]]; + } + + return $cases; + } +} diff --git a/bdd/python/tests/conftest.py b/bdd/python/tests/conftest.py index 390ada907..4c968eea1 100644 --- a/bdd/python/tests/conftest.py +++ b/bdd/python/tests/conftest.py @@ -40,6 +40,8 @@ class GlobalContext: last_topic_partitions: int | None = None last_polled_messages: list[ReceiveMessage] | None = None last_sent_message: str | None = None # Store message payload as string + last_raw_response: bytes | None = None + last_raw_error: RuntimeError | None = None @pytest.fixture(scope="session") diff --git a/bdd/python/tests/test_raw_command.py b/bdd/python/tests/test_raw_command.py new file mode 100644 index 000000000..37e2939d5 --- /dev/null +++ b/bdd/python/tests/test_raw_command.py @@ -0,0 +1,82 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +import asyncio +import socket + +from apache_iggy import IggyClient +from pytest_bdd import given, parsers, scenarios, then, when + +scenarios("/app/features/raw_command.feature") + + +@given("I have a running Iggy server") +def running_server(context): + async def connect(): + host, port = context.server_addr.split(":") + try: + address = f"{socket.gethostbyname(host)}:{port}" + except socket.gaierror: + address = context.server_addr + + context.client = IggyClient(address) + await context.client.connect() + await context.client.ping() + + asyncio.run(connect()) + + +@given("I am authenticated as the root user") +def authenticated_root_user(context): + async def login(): + await context.client.login_user("iggy", "iggy") + + asyncio.run(login()) + + +@when(parsers.parse("I send a raw command with code {code:d} and an empty payload")) +def send_raw_command(context, code): + async def send(): + try: + context.last_raw_response = await context.client.send_raw_with_response( + code, b"" + ) + context.last_raw_error = None + except RuntimeError as error: + context.last_raw_response = None + context.last_raw_error = error + + asyncio.run(send()) + + +@then("the raw command should succeed with an empty response") +def raw_command_succeeds_with_empty_response(context): + assert context.last_raw_error is None + assert context.last_raw_response == b"" + + +@then("the raw command should succeed with a non-empty response") +def raw_command_succeeds_with_non_empty_response(context): + assert context.last_raw_error is None + assert context.last_raw_response + + +@then("the raw command should fail with an invalid command error") +def raw_command_fails_with_invalid_command(context): + assert context.last_raw_response is None + assert context.last_raw_error is not None + assert "invalid command" in str(context.last_raw_error).lower() diff --git a/bdd/rust/Cargo.toml b/bdd/rust/Cargo.toml index ebedb19d2..0b761fb42 100644 --- a/bdd/rust/Cargo.toml +++ b/bdd/rust/Cargo.toml @@ -44,3 +44,8 @@ required-features = ["bdd"] name = "leader_redirection" harness = false required-features = ["bdd"] + +[[test]] +name = "raw_command" +harness = false +required-features = ["bdd"] diff --git a/bdd/rust/tests/common/global_context.rs b/bdd/rust/tests/common/global_context.rs index b18d56388..977c715e2 100644 --- a/bdd/rust/tests/common/global_context.rs +++ b/bdd/rust/tests/common/global_context.rs @@ -15,9 +15,10 @@ // specific language governing permissions and limitations // under the License. +use bytes::Bytes; use cucumber::World; use iggy::clients::client::IggyClient; -use iggy::prelude::{IggyMessage, PolledMessages}; +use iggy::prelude::{IggyError, IggyMessage, PolledMessages}; #[derive(Debug, World, Default)] pub struct GlobalContext { @@ -30,4 +31,5 @@ pub struct GlobalContext { pub last_topic_partitions: Option<u32>, pub last_polled_messages: Option<PolledMessages>, pub last_sent_message: Option<IggyMessage>, + pub last_raw_result: Option<Result<Bytes, IggyError>>, } diff --git a/foreign/go/contracts/version.go b/bdd/rust/tests/raw_command.rs similarity index 76% copy from foreign/go/contracts/version.go copy to bdd/rust/tests/raw_command.rs index 5733ff7a5..d5eae32de 100644 --- a/foreign/go/contracts/version.go +++ b/bdd/rust/tests/raw_command.rs @@ -15,6 +15,14 @@ // specific language governing permissions and limitations // under the License. -package iggcon +pub(crate) mod common; +pub(crate) mod helpers; +pub(crate) mod steps; -const Version = "0.8.1-edge.1" +use crate::common::global_context::GlobalContext; +use cucumber::World; + +#[tokio::main] +async fn main() { + GlobalContext::run("../../bdd/scenarios/raw_command.feature").await; +} diff --git a/bdd/rust/tests/steps/mod.rs b/bdd/rust/tests/steps/mod.rs index 907084ec3..51b40fe99 100644 --- a/bdd/rust/tests/steps/mod.rs +++ b/bdd/rust/tests/steps/mod.rs @@ -18,6 +18,7 @@ pub mod auth; pub mod leader_redirection; pub mod messages; +pub mod raw_command; pub mod server; pub mod streams; pub mod topics; diff --git a/bdd/rust/tests/steps/raw_command.rs b/bdd/rust/tests/steps/raw_command.rs new file mode 100644 index 000000000..72676cdec --- /dev/null +++ b/bdd/rust/tests/steps/raw_command.rs @@ -0,0 +1,60 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use crate::common::global_context::GlobalContext; +use bytes::Bytes; +use cucumber::{then, when}; +use iggy::prelude::IggyError; + +#[when(regex = r"^I send a raw command with code (\d+) and an empty payload$")] +pub async fn when_send_raw_command(world: &mut GlobalContext, code: u32) { + let client = world.client.as_ref().expect("Client should be available"); + world.last_raw_result = Some(client.send_binary_request(code, Bytes::new()).await); +} + +#[then("the raw command should succeed with an empty response")] +pub async fn then_raw_command_succeeds_with_empty_response(world: &mut GlobalContext) { + let response = world + .last_raw_result + .as_ref() + .expect("Should have sent a raw command") + .as_ref() + .expect("Raw command should have succeeded"); + assert!(response.is_empty(), "Response should be empty"); +} + +#[then("the raw command should succeed with a non-empty response")] +pub async fn then_raw_command_succeeds_with_non_empty_response(world: &mut GlobalContext) { + let response = world + .last_raw_result + .as_ref() + .expect("Should have sent a raw command") + .as_ref() + .expect("Raw command should have succeeded"); + assert!(!response.is_empty(), "Response should not be empty"); +} + +#[then("the raw command should fail with an invalid command error")] +pub async fn then_raw_command_fails_with_invalid_command_error(world: &mut GlobalContext) { + let error = world + .last_raw_result + .as_ref() + .expect("Should have sent a raw command") + .as_ref() + .expect_err("Raw command should have failed"); + assert_eq!(*error, IggyError::InvalidCommand); +} diff --git a/bdd/scenarios/raw_command.feature b/bdd/scenarios/raw_command.feature new file mode 100644 index 000000000..158c35d97 --- /dev/null +++ b/bdd/scenarios/raw_command.feature @@ -0,0 +1,41 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +@raw-command +Feature: Raw Command API + As a developer using Apache Iggy + I want to send a command code with a payload directly + So that I can call commands that have no typed SDK method + + Background: + Given I have a running Iggy server + And I am authenticated as the root user + + Scenario: Known command codes round-trip + When I send a raw command with code 1 and an empty payload + Then the raw command should succeed with an empty response + + When I send a raw command with code 10 and an empty payload + Then the raw command should succeed with a non-empty response + + Scenario: Unknown command code is rejected + When I send a raw command with code 60000 and an empty payload + Then the raw command should fail with an invalid command error + + Scenario: Session control command code is rejected + When I send a raw command with code 38 and an empty payload + Then the raw command should fail with an invalid command error diff --git a/foreign/cpp/Cargo.toml b/foreign/cpp/Cargo.toml index ca0e8ad6e..96add1213 100644 --- a/foreign/cpp/Cargo.toml +++ b/foreign/cpp/Cargo.toml @@ -17,7 +17,7 @@ [package] name = "iggy-cpp" -version = "0.1.0" +version = "0.1.1" edition = "2024" [package.metadata.cargo-machete] diff --git a/foreign/cpp/MODULE.bazel b/foreign/cpp/MODULE.bazel index 4e28a2851..985713492 100644 --- a/foreign/cpp/MODULE.bazel +++ b/foreign/cpp/MODULE.bazel @@ -17,7 +17,7 @@ module( name = "iggy_cpp", - version = "0.1.0", + version = "0.1.1", ) bazel_dep(name = "rules_cc", version = "0.2.20") diff --git a/foreign/cpp/src/client.rs b/foreign/cpp/src/client.rs index 4051d40fb..898f2fab9 100644 --- a/foreign/cpp/src/client.rs +++ b/foreign/cpp/src/client.rs @@ -16,6 +16,7 @@ // under the License. use crate::{RUNTIME, ffi}; +use bytes::Bytes; use iggy::prelude::{ Client as IggyConnectionClient, ClusterClient, CompressionAlgorithm as RustCompressionAlgorithm, Consumer, ConsumerGroupClient, @@ -1129,6 +1130,19 @@ impl Client { Ok(()) }) } + + /// Sends a command code and payload and returns the raw response bytes. + /// Session-control codes return an invalid-command error. + pub fn send_raw_with_response(&self, code: u32, payload: Vec<u8>) -> Result<Vec<u8>, String> { + RUNTIME.block_on(async { + let response = self + .inner + .send_binary_request(code, Bytes::from(payload)) + .await + .map_err(|error| format!("Could not send raw command '{code}': {error}"))?; + Ok(response.to_vec()) + }) + } } pub unsafe fn delete_connection(client: *mut Client) -> Result<(), String> { diff --git a/foreign/cpp/src/lib.rs b/foreign/cpp/src/lib.rs index 458e280b6..bc3afa29d 100644 --- a/foreign/cpp/src/lib.rs +++ b/foreign/cpp/src/lib.rs @@ -480,6 +480,7 @@ mod ffi { snapshot_compression: String, snapshot_types: Vec<String>, ) -> Result<Vec<u8>>; + fn send_raw_with_response(self: &Client, code: u32, payload: Vec<u8>) -> Result<Vec<u8>>; // Future functions fn disconnect(self: &Client) -> Result<()>; diff --git a/foreign/cpp/tests/client/low_level_e2e.cpp b/foreign/cpp/tests/client/low_level_e2e.cpp index 590c8f597..4ae17d354 100644 --- a/foreign/cpp/tests/client/low_level_e2e.cpp +++ b/foreign/cpp/tests/client/low_level_e2e.cpp @@ -1703,3 +1703,45 @@ TEST_F(LowLevelE2E_Client, SnapshotWithInvalidSnapshotTypeThrows) { ASSERT_THROW(client->snapshot("deflated", make_snapshot_types({"not-a-real-type"})), std::exception); } + +TEST_F(LowLevelE2E_Client, SendRawWithResponsePingReturnsEmptyBytes) { + RecordProperty("description", "Returns an empty response body for a raw ping command with an empty payload."); + constexpr std::uint32_t ping_command_code = 1; + iggy::ffi::Client *client = GetLoggedInClient(); + + rust::Vec<std::uint8_t> empty_payload; + rust::Vec<std::uint8_t> response; + ASSERT_NO_THROW({ response = client->send_raw_with_response(ping_command_code, empty_payload); }); + EXPECT_TRUE(response.empty()); +} + +TEST_F(LowLevelE2E_Client, SendRawWithResponseGetStatsReturnsNonEmptyBytes) { + RecordProperty("description", + "Returns a non-empty response body for a raw get-stats command with an empty payload."); + constexpr std::uint32_t get_stats_command_code = 10; + iggy::ffi::Client *client = GetLoggedInClient(); + + rust::Vec<std::uint8_t> empty_payload; + rust::Vec<std::uint8_t> response; + ASSERT_NO_THROW({ response = client->send_raw_with_response(get_stats_command_code, empty_payload); }); + EXPECT_FALSE(response.empty()); +} + +TEST_F(LowLevelE2E_Client, SendRawWithResponseLoginUserCodeThrows) { + RecordProperty("description", + "Rejects the login-user session-control code client-side before it reaches the server."); + constexpr std::uint32_t login_user_command_code = 38; + iggy::ffi::Client *client = GetLoggedInClient(); + + rust::Vec<std::uint8_t> empty_payload; + ASSERT_THROW(client->send_raw_with_response(login_user_command_code, empty_payload), std::exception); +} + +TEST_F(LowLevelE2E_Client, SendRawWithResponseUnknownCommandCodeThrows) { + RecordProperty("description", "Rejects an unknown command code with an invalid-command error from the server."); + constexpr std::uint32_t unknown_command_code = 60000; + iggy::ffi::Client *client = GetLoggedInClient(); + + rust::Vec<std::uint8_t> empty_payload; + ASSERT_THROW(client->send_raw_with_response(unknown_command_code, empty_payload), std::exception); +} diff --git a/foreign/csharp/Iggy_SDK.Tests.BDD/Context/TestContext.cs b/foreign/csharp/Iggy_SDK.Tests.BDD/Context/TestContext.cs index 8edd5df70..81c12f2e8 100644 --- a/foreign/csharp/Iggy_SDK.Tests.BDD/Context/TestContext.cs +++ b/foreign/csharp/Iggy_SDK.Tests.BDD/Context/TestContext.cs @@ -34,4 +34,6 @@ public class TestContext public Message? LastSendMessage { get; set; } public bool RedirectionOccurred { get; set; } public uint? LastStreamId { get; set; } + public byte[]? LastRawResponse { get; set; } + public Exception? LastRawError { get; set; } } diff --git a/foreign/csharp/Iggy_SDK.Tests.BDD/Context/TestHooks.cs b/foreign/csharp/Iggy_SDK.Tests.BDD/Context/TestHooks.cs index 3e7c55095..58c5debdb 100644 --- a/foreign/csharp/Iggy_SDK.Tests.BDD/Context/TestHooks.cs +++ b/foreign/csharp/Iggy_SDK.Tests.BDD/Context/TestHooks.cs @@ -39,6 +39,8 @@ public class TestHooks _context.CreatedStream = null; _context.RedirectionOccurred = false; _context.LastStreamId = null; + _context.LastRawResponse = null; + _context.LastRawError = null; } [AfterScenario] diff --git a/foreign/csharp/Iggy_SDK.Tests.BDD/Iggy_SDK.Tests.BDD.csproj b/foreign/csharp/Iggy_SDK.Tests.BDD/Iggy_SDK.Tests.BDD.csproj index c97aac998..489a43fc6 100644 --- a/foreign/csharp/Iggy_SDK.Tests.BDD/Iggy_SDK.Tests.BDD.csproj +++ b/foreign/csharp/Iggy_SDK.Tests.BDD/Iggy_SDK.Tests.BDD.csproj @@ -61,6 +61,9 @@ <ReqnrollFeatureFile Include="..\..\..\bdd\scenarios\leader_redirection.feature"> <Link>Features\leader_redirection.feature</Link> </ReqnrollFeatureFile> + <ReqnrollFeatureFile Include="..\..\..\bdd\scenarios\raw_command.feature"> + <Link>Features\raw_command.feature</Link> + </ReqnrollFeatureFile> </ItemGroup> <ItemGroup> diff --git a/foreign/csharp/Iggy_SDK.Tests.BDD/StepDefinitions/RawCommandSteps.cs b/foreign/csharp/Iggy_SDK.Tests.BDD/StepDefinitions/RawCommandSteps.cs new file mode 100644 index 000000000..9515147e2 --- /dev/null +++ b/foreign/csharp/Iggy_SDK.Tests.BDD/StepDefinitions/RawCommandSteps.cs @@ -0,0 +1,71 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +using Apache.Iggy.Exceptions; +using Reqnroll; +using Shouldly; +using TestContext = Apache.Iggy.Tests.BDD.Context.TestContext; + +namespace Apache.Iggy.Tests.BDD.StepDefinitions; + +[Binding] +public class RawCommandSteps +{ + private readonly TestContext _context; + + public RawCommandSteps(TestContext context) + { + _context = context; + } + + [When(@"I send a raw command with code (\d+) and an empty payload")] + public async Task WhenISendARawCommand(uint code) + { + try + { + _context.LastRawResponse = await _context.IggyClient.SendRawWithResponseAsync(code, []); + _context.LastRawError = null; + } + catch (Exception error) + { + _context.LastRawResponse = null; + _context.LastRawError = error; + } + } + + [Then(@"the raw command should succeed with an empty response")] + public void ThenTheRawCommandShouldSucceedWithAnEmptyResponse() + { + _context.LastRawError.ShouldBeNull(); + _context.LastRawResponse.ShouldBeEmpty(); + } + + [Then(@"the raw command should succeed with a non-empty response")] + public void ThenTheRawCommandShouldSucceedWithANonEmptyResponse() + { + _context.LastRawError.ShouldBeNull(); + _context.LastRawResponse.ShouldNotBeEmpty(); + } + + [Then(@"the raw command should fail with an invalid command error")] + public void ThenTheRawCommandShouldFailWithAnInvalidCommandError() + { + _context.LastRawResponse.ShouldBeNull(); + var error = _context.LastRawError.ShouldBeOfType<IggyInvalidStatusCodeException>(); + error.StatusCode.ShouldBe(3); + } +} diff --git a/foreign/csharp/Iggy_SDK.Tests.Integration/RawCommandTests.cs b/foreign/csharp/Iggy_SDK.Tests.Integration/RawCommandTests.cs new file mode 100644 index 000000000..18add9f5e --- /dev/null +++ b/foreign/csharp/Iggy_SDK.Tests.Integration/RawCommandTests.cs @@ -0,0 +1,83 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +using Apache.Iggy.Enums; +using Apache.Iggy.Exceptions; +using Apache.Iggy.Tests.Integrations.Attributes; +using Apache.Iggy.Tests.Integrations.Fixtures; +using Shouldly; + +namespace Apache.Iggy.Tests.Integrations; + +public class RawCommandTests +{ + [ClassDataSource<IggyServerFixture>(Shared = SharedType.PerAssembly)] + public required IggyServerFixture Fixture { get; init; } + + [Test] + [SkipHttp] + [MethodDataSource<IggyServerFixture>(nameof(IggyServerFixture.ProtocolData))] + public async Task SendRawWithResponse_Tcp_ShouldReturnRawPayloads(Protocol protocol) + { + var client = await Fixture.CreateAuthenticatedClient(protocol); + + var pingResponse = await client.SendRawWithResponseAsync(1, []); + var statsResponse = await client.SendRawWithResponseAsync(10, []); + + pingResponse.ShouldBeEmpty(); + statsResponse.ShouldNotBeEmpty(); + } + + [Test] + [SkipHttp] + [MethodDataSource<IggyServerFixture>(nameof(IggyServerFixture.ProtocolData))] + public async Task SendRawWithResponse_Tcp_ShouldRejectSessionControlCodes(Protocol protocol) + { + var client = await Fixture.CreateAuthenticatedClient(protocol); + + foreach (var code in new uint[] { 38, 39, 40, 44, 45 }) + { + var exception = await Should.ThrowAsync<IggyInvalidStatusCodeException>( + () => client.SendRawWithResponseAsync(code, [])); + exception.StatusCode.ShouldBe(3); + } + } + + [Test] + [SkipHttp] + [MethodDataSource<IggyServerFixture>(nameof(IggyServerFixture.ProtocolData))] + public async Task SendRawWithResponse_Tcp_ShouldPropagateServerError(Protocol protocol) + { + var client = await Fixture.CreateAuthenticatedClient(protocol); + + var exception = await Should.ThrowAsync<IggyInvalidStatusCodeException>( + () => client.SendRawWithResponseAsync(60_000, [])); + + exception.StatusCode.ShouldBe(3); + } + + [Test] + [SkipTcp] + [MethodDataSource<IggyServerFixture>(nameof(IggyServerFixture.ProtocolData))] + public async Task SendRawWithResponse_Http_ShouldThrowFeatureUnavailable(Protocol protocol) + { + var client = await Fixture.CreateAuthenticatedClient(protocol); + + await Should.ThrowAsync<FeatureUnavailableException>( + () => client.SendRawWithResponseAsync(1, [])); + } +} diff --git a/foreign/csharp/Iggy_SDK/IggyClient/IIggyClient.cs b/foreign/csharp/Iggy_SDK/IggyClient/IIggyClient.cs index bcc7a0a26..0cf6cf276 100644 --- a/foreign/csharp/Iggy_SDK/IggyClient/IIggyClient.cs +++ b/foreign/csharp/Iggy_SDK/IggyClient/IIggyClient.cs @@ -58,4 +58,17 @@ public interface IIggyClient : IIggyPublisher, IIggyStream, IIggyTopic, IIggyCon /// </summary> /// <returns>The current address of the client.</returns> string GetCurrentAddress(); + + /// <summary> + /// Sends a command code with a payload and returns the raw response bytes. + /// </summary> + /// <remarks> + /// Session-control codes are rejected with an invalid-command error. + /// HTTP clients report that this operation is unavailable. + /// </remarks> + /// <param name="code">The numeric command code to send.</param> + /// <param name="payload">The request payload.</param> + /// <param name="token">The cancellation token to cancel the operation.</param> + /// <returns>A task that represents the asynchronous operation and returns the raw response payload bytes.</returns> + Task<byte[]> SendRawWithResponseAsync(uint code, byte[] payload, CancellationToken token = default); } diff --git a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/HttpMessageStream.cs b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/HttpMessageStream.cs index 96bd38520..75a806e90 100644 --- a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/HttpMessageStream.cs +++ b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/HttpMessageStream.cs @@ -525,6 +525,19 @@ public class HttpMessageStream : IIggyClient return []; } + /// <summary> + /// This method is only supported in TCP protocol + /// </summary> + /// <param name="code">The numeric command code to send.</param> + /// <param name="payload">The opaque request payload.</param> + /// <param name="token">The cancellation token to cancel the operation.</param> + /// <returns>A task representing the asynchronous operation.</returns> + /// <exception cref="FeatureUnavailableException"></exception> + public Task<byte[]> SendRawWithResponseAsync(uint code, byte[] payload, CancellationToken token = default) + { + throw new FeatureUnavailableException(); + } + /// <inheritdoc /> public Task ConnectAsync(CancellationToken token = default) { diff --git a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs index 373488ec1..227532d2e 100644 --- a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs +++ b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs @@ -45,6 +45,17 @@ namespace Apache.Iggy.IggyClient.Implementations; /// </summary> public sealed class TcpMessageStream : IIggyClient { + private const int InvalidCommandStatus = 3; + + private static readonly HashSet<uint> SessionControlCodes = + [ + CommandCodes.LOGIN_USER_CODE, + CommandCodes.LOGOUT_USER_CODE, + CommandCodes.LOGIN_REGISTER_CODE, + CommandCodes.LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE, + CommandCodes.LOGIN_REGISTER_WITH_PAT_CODE + ]; + private readonly IggyClientConfigurator _configuration; private readonly EventAggregator<ConnectionStateChangedEventArgs> _connectionEvents; private readonly SemaphoreSlim _connectionSemaphore; @@ -582,6 +593,23 @@ public sealed class TcpMessageStream : IIggyClient return result.Memory.Span.ToArray(); } + /// <inheritdoc /> + public async Task<byte[]> SendRawWithResponseAsync(uint code, byte[] payload, CancellationToken token = default) + { + if (SessionControlCodes.Contains(code)) + { + throw new IggyInvalidStatusCodeException(InvalidCommandStatus, + $"Invalid response status code: {InvalidCommandStatus}"); + } + + var buffer = new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + payload.Length]; + TcpMessageStreamHelpers.CreatePayload(buffer, payload, (int)code); + + using IMemoryOwner<byte> result = await SendWithResponseAsync(buffer, token); + + return result.Memory.Length <= 1 ? [] : result.Memory.Span.ToArray(); + } + /// <inheritdoc /> public async Task ConnectAsync(CancellationToken token = default) { diff --git a/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj b/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj index e4b821d99..776883aac 100644 --- a/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj +++ b/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj @@ -27,7 +27,7 @@ <TargetFrameworks>net8.0;net10.0</TargetFrameworks> <AssemblyName>Apache.Iggy</AssemblyName> <RootNamespace>Apache.Iggy</RootNamespace> - <Version>0.8.1-edge.3</Version> + <Version>0.8.1-edge.4</Version> <GenerateDocumentationFile>true</GenerateDocumentationFile> </PropertyGroup> diff --git a/foreign/csharp/Iggy_SDK/Utils/CommandCodes.cs b/foreign/csharp/Iggy_SDK/Utils/CommandCodes.cs index 2b2beeb0e..8922d37d6 100644 --- a/foreign/csharp/Iggy_SDK/Utils/CommandCodes.cs +++ b/foreign/csharp/Iggy_SDK/Utils/CommandCodes.cs @@ -35,10 +35,12 @@ internal static class CommandCodes internal const int CHANGE_PASSWORD_CODE = 37; internal const int LOGIN_USER_CODE = 38; internal const int LOGOUT_USER_CODE = 39; + internal const int LOGIN_REGISTER_CODE = 40; internal const int GET_PERSONAL_ACCESS_TOKENS_CODE = 41; internal const int CREATE_PERSONAL_ACCESS_TOKEN_CODE = 42; internal const int DELETE_PERSONAL_ACCESS_TOKEN_CODE = 43; internal const int LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE = 44; + internal const int LOGIN_REGISTER_WITH_PAT_CODE = 45; internal const int POLL_MESSAGES_CODE = 100; internal const int SEND_MESSAGES_CODE = 101; internal const int FLUSH_UNSAVED_BUFFER_CODE = 102; diff --git a/foreign/go/client/tcp/tcp_core.go b/foreign/go/client/tcp/tcp_core.go index 8952fce1a..425976cd0 100644 --- a/foreign/go/client/tcp/tcp_core.go +++ b/foreign/go/client/tcp/tcp_core.go @@ -305,6 +305,38 @@ func (c *IggyTcpClient) do(ctx context.Context, cmd command.Command) ([]byte, er return resp, err } +// SendRawWithResponse sends a command code and payload and returns the raw response body. +// Session-control codes return ierror.ErrInvalidCommand without writing to the connection. +func (c *IggyTcpClient) SendRawWithResponse(ctx context.Context, code uint32, payload []byte) ([]byte, error) { + if isSessionControlCode(code) { + return nil, ierror.ErrInvalidCommand + } + + bp := acquireRequestBuf() + defer releaseRequestBuf(bp) + + buf := append((*bp)[:0], 0, 0, 0, 0, 0, 0, 0, 0) + binary.LittleEndian.PutUint32(buf[4:8], code) + buf = append(buf, payload...) + binary.LittleEndian.PutUint32(buf[0:4], uint32(len(buf)-4)) + *bp = buf + + return c.sendWireAndFetchResponse(ctx, buf) +} + +func isSessionControlCode(code uint32) bool { + switch code { + case uint32(command.LoginUserCode), + uint32(command.LogoutUserCode), + uint32(command.LoginRegisterCode), + uint32(command.LoginWithAccessTokenCode), + uint32(command.LoginRegisterWithPATCode): + return true + default: + return false + } +} + // encodeWireRequest writes the wire-format request (4-byte length, 4-byte // code, then body) into buf, growing it as needed. The length prefix is // written from the realized body length, so a buggy or unimplemented diff --git a/foreign/go/client/tcp/tcp_core_test.go b/foreign/go/client/tcp/tcp_core_test.go index f996f0298..0264fb1ee 100644 --- a/foreign/go/client/tcp/tcp_core_test.go +++ b/foreign/go/client/tcp/tcp_core_test.go @@ -30,6 +30,7 @@ import ( iggcon "github.com/apache/iggy/foreign/go/contracts" ierror "github.com/apache/iggy/foreign/go/errors" + "github.com/apache/iggy/foreign/go/internal/command" ) // emptyWireReq is an 8-byte wire payload for a zero-code request with empty body: @@ -383,3 +384,106 @@ func TestDisconnect_ShutdownClientIsNotResurrected(t *testing.T) { t.Errorf("got %v, want %v", err, ierror.ErrClientShutdown) } } + +func TestSendRawWithResponse_FrameLayoutAndSuccess(t *testing.T) { + c, serverConn := newTestClient(t) + const code = uint32(60_000) + payload := []byte{0xAA, 0xBB, 0xCC} + body := []byte("opaque response") + + type capturedFrame struct { + length uint32 + code uint32 + body []byte + } + captured := make(chan capturedFrame, 1) + go func() { + var lengthBuf [RequestInitialBytesLength]byte + if _, err := serverConn.Read(lengthBuf[:]); err != nil { + t.Errorf("server: read request length: %v", err) + return + } + length := binary.LittleEndian.Uint32(lengthBuf[:]) + req := make([]byte, length) + if _, err := serverConn.Read(req); err != nil { + t.Errorf("server: read request body: %v", err) + return + } + captured <- capturedFrame{ + length: length, + code: binary.LittleEndian.Uint32(req[:4]), + body: req[4:], + } + + resp := make([]byte, 8+len(body)) + binary.LittleEndian.PutUint32(resp[0:4], 0) + binary.LittleEndian.PutUint32(resp[4:8], uint32(len(body))) + copy(resp[8:], body) + if _, err := serverConn.Write(resp); err != nil { + t.Errorf("server: write response: %v", err) + } + }() + + result, err := c.SendRawWithResponse(context.Background(), code, payload) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if string(result) != string(body) { + t.Errorf("got response %q, want %q", result, body) + } + + frame := <-captured + wantLength := uint32(4 + len(payload)) + if frame.length != wantLength { + t.Errorf("got length prefix=%d, want %d", frame.length, wantLength) + } + if frame.code != code { + t.Errorf("got code=%d, want %d", frame.code, code) + } + if !bytes.Equal(frame.body, payload) { + t.Errorf("got payload=%v, want %v", frame.body, payload) + } +} + +func TestSendRawWithResponse_ErrorStatus(t *testing.T) { + c, serverConn := newTestClient(t) + + go serverRespond(t, serverConn, uint32(ierror.InvalidCommandCode), nil) + + _, err := c.SendRawWithResponse(context.Background(), 60_000, nil) + if !errors.Is(err, ierror.ErrInvalidCommand) { + t.Errorf("got %v, want %v", err, ierror.ErrInvalidCommand) + } +} + +func TestSendRawWithResponse_SessionControlGuard(t *testing.T) { + c, serverConn := newTestClient(t) + + wrote := make(chan struct{}) + go func() { + buf := make([]byte, 1) + if _, err := serverConn.Read(buf); err == nil { + close(wrote) + } + }() + + guardedCodes := []uint32{ + uint32(command.LoginUserCode), + uint32(command.LogoutUserCode), + uint32(command.LoginRegisterCode), + uint32(command.LoginWithAccessTokenCode), + uint32(command.LoginRegisterWithPATCode), + } + for _, guardedCode := range guardedCodes { + _, err := c.SendRawWithResponse(context.Background(), guardedCode, nil) + if !errors.Is(err, ierror.ErrInvalidCommand) { + t.Errorf("code %d: got %v, want %v", guardedCode, err, ierror.ErrInvalidCommand) + } + } + + select { + case <-wrote: + t.Fatal("expected no bytes to be written to the wire for session-control codes") + case <-time.After(50 * time.Millisecond): + } +} diff --git a/foreign/go/contracts/client.go b/foreign/go/contracts/client.go index 144cc17e3..68efe8d1a 100644 --- a/foreign/go/contracts/client.go +++ b/foreign/go/contracts/client.go @@ -300,4 +300,8 @@ type Client interface { // GetClient get the info about a specific client by unique ID (not to be confused with the user). // Authentication is required, and the permission to read the server info. GetClient(ctx context.Context, clientId uint32) (*ClientInfoDetails, error) + + // SendRawWithResponse sends a command code and payload and returns the raw response body. + // Session-control codes return ierror.ErrInvalidCommand without writing to the connection. + SendRawWithResponse(ctx context.Context, code uint32, payload []byte) ([]byte, error) } diff --git a/foreign/go/contracts/version.go b/foreign/go/contracts/version.go index 5733ff7a5..f6d9eb82a 100644 --- a/foreign/go/contracts/version.go +++ b/foreign/go/contracts/version.go @@ -17,4 +17,4 @@ package iggcon -const Version = "0.8.1-edge.1" +const Version = "0.8.1-edge.2" diff --git a/foreign/go/internal/command/code.go b/foreign/go/internal/command/code.go index 644abf2dd..a2648af5a 100644 --- a/foreign/go/internal/command/code.go +++ b/foreign/go/internal/command/code.go @@ -36,10 +36,12 @@ const ( ChangePasswordCode Code = 37 LoginUserCode Code = 38 LogoutUserCode Code = 39 + LoginRegisterCode Code = 40 GetAccessTokensCode Code = 41 CreateAccessTokenCode Code = 42 DeleteAccessTokenCode Code = 43 LoginWithAccessTokenCode Code = 44 + LoginRegisterWithPATCode Code = 45 PollMessagesCode Code = 100 SendMessagesCode Code = 101 GetOffsetCode Code = 120 diff --git a/foreign/java/gradle.properties b/foreign/java/gradle.properties index a6847759f..5339adf06 100644 --- a/foreign/java/gradle.properties +++ b/foreign/java/gradle.properties @@ -15,5 +15,5 @@ # specific language governing permissions and limitations # under the License. -version=0.8.1-SNAPSHOT +version=0.8.2-SNAPSHOT group=org.apache.iggy diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java index 9a9738db5..1e6625280 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java @@ -19,6 +19,7 @@ package org.apache.iggy.client.async.tcp; +import io.netty.buffer.Unpooled; import org.apache.iggy.IggyVersion; import org.apache.iggy.client.async.ConsumerGroupsClient; import org.apache.iggy.client.async.ConsumerOffsetsClient; @@ -33,6 +34,8 @@ import org.apache.iggy.client.async.tcp.AsyncTcpConnection.TCPConnectionPoolConf import org.apache.iggy.config.RetryPolicy; import org.apache.iggy.exception.IggyMissingCredentialsException; import org.apache.iggy.exception.IggyNotConnectedException; +import org.apache.iggy.exception.IggyServerException; +import org.apache.iggy.serde.CommandCode; import org.apache.iggy.user.IdentityInfo; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -94,6 +97,7 @@ import java.util.concurrent.CompletableFuture; */ public class AsyncIggyTcpClient { + private static final int INVALID_COMMAND_ERROR_CODE = 3; private static final Logger log = LoggerFactory.getLogger(AsyncIggyTcpClient.class); private final String host; @@ -217,6 +221,39 @@ public class AsyncIggyTcpClient { return usersClient.login(username.get(), password.get()); } + /** + * Sends a command code and payload and returns the raw response payload. + * + * <p>Session-control codes complete the returned future with an invalid-command error. + * + * @param code the command code + * @param payload the command payload + * @return a future containing the raw response payload + * @throws IggyNotConnectedException if {@link #connect()} has not been called + */ + public CompletableFuture<byte[]> sendRawWithResponse(int code, byte[] payload) { + if (isSessionControlCode(code)) { + return CompletableFuture.failedFuture( + IggyServerException.fromTcpResponse(INVALID_COMMAND_ERROR_CODE, new byte[0])); + } + if (connection == null) { + throw new IggyNotConnectedException(); + } + + return connection.send(code, Unpooled.wrappedBuffer(payload)).thenApply(response -> { + try { + if (response.readableBytes() <= 1) { + return new byte[0]; + } + byte[] responsePayload = new byte[response.readableBytes()]; + response.readBytes(responsePayload); + return responsePayload; + } finally { + response.release(); + } + }); + } + /** * Returns the async users client for authentication operations. * @@ -348,4 +385,12 @@ public class AsyncIggyTcpClient { } return CompletableFuture.completedFuture(null); } + + private static boolean isSessionControlCode(int code) { + return code == CommandCode.User.LOGIN.getValue() + || code == CommandCode.User.LOGOUT.getValue() + || code == CommandCode.User.LOGIN_REGISTER.getValue() + || code == CommandCode.PersonalAccessToken.LOGIN.getValue() + || code == CommandCode.PersonalAccessToken.LOGIN_REGISTER.getValue(); + } } diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/IggyBaseClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/IggyBaseClient.java index b0cf1cd65..9aaa971d2 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/IggyBaseClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/IggyBaseClient.java @@ -21,6 +21,18 @@ package org.apache.iggy.client.blocking; public interface IggyBaseClient { + /** + * Sends a command code and payload and returns the raw response payload. + * + * <p>Session-control codes are rejected with an invalid-command error. HTTP clients report + * that this operation is unsupported. + * + * @param code the command code + * @param payload the command payload + * @return the raw response payload + */ + byte[] sendRawWithResponse(int code, byte[] payload); + SystemClient system(); StreamsClient streams(); diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/IggyHttpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/IggyHttpClient.java index bb532394f..bdb22f210 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/IggyHttpClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/IggyHttpClient.java @@ -31,6 +31,7 @@ import org.apache.iggy.client.blocking.SystemClient; import org.apache.iggy.client.blocking.TopicsClient; import org.apache.iggy.client.blocking.UsersClient; import org.apache.iggy.exception.IggyMissingCredentialsException; +import org.apache.iggy.exception.IggyOperationNotSupportedException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -110,6 +111,12 @@ public class IggyHttpClient implements IggyBaseClient, Closeable { usersClient.login(username.get(), password.get()); } + /** {@inheritDoc} */ + @Override + public byte[] sendRawWithResponse(int code, byte[] payload) { + throw new IggyOperationNotSupportedException("sendRawWithResponse", "HTTP"); + } + @Override public SystemClient system() { return systemClient; diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/IggyTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/IggyTcpClient.java index e3b8c03d8..6816fe42e 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/IggyTcpClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/IggyTcpClient.java @@ -93,6 +93,12 @@ public class IggyTcpClient implements IggyBaseClient, Closeable { return FutureUtil.resolve(asyncClient.login()); } + /** {@inheritDoc} */ + @Override + public byte[] sendRawWithResponse(int code, byte[] payload) { + return FutureUtil.resolve(asyncClient.sendRawWithResponse(code, payload)); + } + @Override public void close() { FutureUtil.resolve(asyncClient.close()); diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/CommandCode.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/CommandCode.java index e9f1f7905..236394707 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/CommandCode.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/CommandCode.java @@ -55,7 +55,8 @@ public interface CommandCode { UPDATE_PERMISSIONS(36), CHANGE_PASSWORD(37), LOGIN(38), - LOGOUT(39); + LOGOUT(39), + LOGIN_REGISTER(40); private final int value; @@ -73,7 +74,8 @@ public interface CommandCode { GET_ALL(41), CREATE(42), DELETE(43), - LOGIN(44); + LOGIN(44), + LOGIN_REGISTER(45); private final int value; diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/IggyBaseClient.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/http/RawCommandHttpClientTest.java similarity index 61% copy from foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/IggyBaseClient.java copy to foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/http/RawCommandHttpClientTest.java index b0cf1cd65..06551ad16 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/IggyBaseClient.java +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/http/RawCommandHttpClientTest.java @@ -17,25 +17,20 @@ * under the License. */ -package org.apache.iggy.client.blocking; +package org.apache.iggy.client.blocking.http; -public interface IggyBaseClient { +import org.apache.iggy.exception.IggyOperationNotSupportedException; +import org.junit.jupiter.api.Test; - SystemClient system(); +import static org.assertj.core.api.Assertions.assertThatThrownBy; - StreamsClient streams(); +class RawCommandHttpClientTest { - UsersClient users(); + @Test + void shouldRejectRawCommands() { + var client = new IggyHttpClient("http://127.0.0.1:3000";); - TopicsClient topics(); - - PartitionsClient partitions(); - - ConsumerGroupsClient consumerGroups(); - - ConsumerOffsetsClient consumerOffsets(); - - MessagesClient messages(); - - PersonalAccessTokensClient personalAccessTokens(); + assertThatThrownBy(() -> client.sendRawWithResponse(1, new byte[0])) + .isInstanceOf(IggyOperationNotSupportedException.class); + } } diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/RawCommandTcpClientTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/RawCommandTcpClientTest.java new file mode 100644 index 000000000..eceb57e78 --- /dev/null +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/RawCommandTcpClientTest.java @@ -0,0 +1,67 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iggy.client.blocking.tcp; + +import org.apache.iggy.client.blocking.IggyBaseClient; +import org.apache.iggy.client.blocking.IntegrationTest; +import org.apache.iggy.exception.IggyErrorCode; +import org.apache.iggy.exception.IggyServerException; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class RawCommandTcpClientTest extends IntegrationTest { + + @Override + protected IggyBaseClient getClient() { + return TcpClientFactory.create(serverHost(), serverTcpPort()); + } + + @BeforeEach + void authenticate() { + login(); + } + + @Test + void shouldReturnRawResponsePayloads() { + assertThat(client.sendRawWithResponse(1, new byte[0])).isEmpty(); + assertThat(client.sendRawWithResponse(10, new byte[0])).isNotEmpty(); + } + + @Test + void shouldRejectAllSessionControlCodes() { + for (int code : new int[] {38, 39, 40, 44, 45}) { + assertThatThrownBy(() -> client.sendRawWithResponse(code, new byte[0])) + .isInstanceOf(IggyServerException.class) + .extracting(exception -> ((IggyServerException) exception).getErrorCode()) + .isEqualTo(IggyErrorCode.INVALID_COMMAND); + } + } + + @Test + void shouldPropagateUnknownCommandError() { + assertThatThrownBy(() -> client.sendRawWithResponse(60_000, new byte[0])) + .isInstanceOf(IggyServerException.class) + .extracting(exception -> ((IggyServerException) exception).getErrorCode()) + .isEqualTo(IggyErrorCode.INVALID_COMMAND); + } +} diff --git a/foreign/node/package-lock.json b/foreign/node/package-lock.json index 67fb8a3e2..afe1087ed 100644 --- a/foreign/node/package-lock.json +++ b/foreign/node/package-lock.json @@ -1,12 +1,12 @@ { "name": "apache-iggy", - "version": "0.8.1-edge.1", + "version": "0.8.1-edge.2", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "apache-iggy", - "version": "0.8.1-edge.1", + "version": "0.8.1-edge.2", "license": "Apache-2.0", "dependencies": { "debug": "4.4.3", diff --git a/foreign/node/package.json b/foreign/node/package.json index 8cc59ca73..339cec683 100644 --- a/foreign/node/package.json +++ b/foreign/node/package.json @@ -1,7 +1,7 @@ { "name": "apache-iggy", "type": "module", - "version": "0.8.1-edge.1", + "version": "0.8.1-edge.2", "description": "Official Apache Iggy NodeJS SDK", "keywords": [ "iggy", diff --git a/foreign/node/src/bdd/raw.ts b/foreign/node/src/bdd/raw.ts new file mode 100644 index 000000000..9eaa96e14 --- /dev/null +++ b/foreign/node/src/bdd/raw.ts @@ -0,0 +1,48 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +import assert from 'node:assert/strict'; +import { Then, When } from '@cucumber/cucumber'; +import type { TestWorld } from './world.js'; + +When( + 'I send a raw command with code {int} and an empty payload', + async function (this: TestWorld, code: number) { + try { + this.rawResponse = await this.client.sendRawWithResponse(code, Buffer.alloc(0)); + this.rawError = undefined; + } catch (error) { + this.rawResponse = undefined; + this.rawError = error instanceof Error ? error : new Error(String(error)); + } + } +); + +Then('the raw command should succeed with an empty response', function (this: TestWorld) { + assert.equal(this.rawError, undefined); + assert.deepEqual(this.rawResponse, Buffer.alloc(0)); +}); + +Then('the raw command should succeed with a non-empty response', function (this: TestWorld) { + assert.equal(this.rawError, undefined); + assert.ok(this.rawResponse && this.rawResponse.length > 0); +}); + +Then('the raw command should fail with an invalid command error', function (this: TestWorld) { + assert.equal(this.rawResponse, undefined); + assert.match(this.rawError?.message ?? '', /code: 3, message: Invalid command/); +}); diff --git a/foreign/node/src/bdd/world.ts b/foreign/node/src/bdd/world.ts index 8d84f773e..aeade3c04 100644 --- a/foreign/node/src/bdd/world.ts +++ b/foreign/node/src/bdd/world.ts @@ -26,5 +26,7 @@ export interface TestWorld { stream: Stream, topic: Topic, sendMessages: CreateMessage[], - polledMessages: Message[] + polledMessages: Message[], + rawResponse?: Buffer, + rawError?: Error }; diff --git a/foreign/node/src/e2e/tcp.raw.e2e.ts b/foreign/node/src/e2e/tcp.raw.e2e.ts new file mode 100644 index 000000000..4f140c489 --- /dev/null +++ b/foreign/node/src/e2e/tcp.raw.e2e.ts @@ -0,0 +1,53 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. +// + +import { after, describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { getTestClient } from './test-client.utils.js'; +import { COMMAND_CODE } from '../wire/command.code.js'; + +describe('e2e -> raw', async () => { + + const c = getTestClient(); + + it('e2e -> raw::ping', async () => { + const response = await c.sendRawWithResponse(COMMAND_CODE.Ping, Buffer.alloc(0)); + assert.deepEqual(response, Buffer.alloc(0)); + }); + + it('e2e -> raw::getStats', async () => { + const response = await c.sendRawWithResponse(COMMAND_CODE.GetStats, Buffer.alloc(0)); + assert.ok(response.length > 0); + }); + + it('e2e -> raw::sessionControlCodeRejectedClientSide', async () => { + await assert.rejects( + () => c.sendRawWithResponse(COMMAND_CODE.LoginUser, Buffer.alloc(0)) + ); + }); + + it('e2e -> raw::unknownCodeRejectedByServer', async () => { + await assert.rejects( + () => c.sendRawWithResponse(60000, Buffer.alloc(0)) + ); + }); + + after(() => { + c.destroy(); + }); +}); diff --git a/foreign/node/src/wire/command-set.test.ts b/foreign/node/src/wire/command-set.test.ts new file mode 100644 index 000000000..7cca4ea84 --- /dev/null +++ b/foreign/node/src/wire/command-set.test.ts @@ -0,0 +1,98 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. +// + +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { SimpleClient } from '../client/client.js'; +import type { RawClient } from '../client/client.type.js'; +import { COMMAND_CODE } from './command.code.js'; + +const mockRawClient = (): RawClient => ({ + sendCommand: async () => { + throw new Error('sendCommand should not be called by the session-control guard'); + }, + isAuthenticated: true, + authenticate: async () => { + throw new Error('authenticate should not be called by the session-control guard'); + }, + destroy: () => {}, + on: () => {}, + once: () => {}, + getReadStream: () => { + throw new Error('getReadStream should not be called by the session-control guard'); + }, +}); + +describe('CommandAPI.sendRawWithResponse', () => { + + describe('session-control guard', () => { + + [ + COMMAND_CODE.LoginUser, + COMMAND_CODE.LogoutUser, + COMMAND_CODE.LoginRegister, + COMMAND_CODE.LoginWithAccessToken, + COMMAND_CODE.LoginRegisterWithAccessToken, + ].forEach((code) => { + it(`rejects code ${code} before reaching the client provider`, async () => { + const client = new SimpleClient(mockRawClient()); + await assert.rejects( + () => client.sendRawWithResponse(code, Buffer.alloc(0)), + /code: 3, message: Invalid command/ + ); + }); + }); + + }); + + it('forwards a custom code and opaque payload to sendCommand', async () => { + const customCode = 60_000; + const payload = Buffer.from([0xAA, 0xBB, 0xCC]); + const expectedResponse = Buffer.from('opaque response'); + const raw = mockRawClient(); + raw.sendCommand = async (code, sentPayload) => { + assert.equal(code, customCode); + assert.deepEqual(sentPayload, payload); + return { + status: 0, + length: expectedResponse.length, + data: expectedResponse, + }; + }; + const client = new SimpleClient(raw); + const response = await client.sendRawWithResponse(customCode, payload); + assert.deepEqual(response, expectedResponse); + }); + + it('normalizes a one-byte response to an empty buffer', async () => { + const raw = mockRawClient(); + raw.sendCommand = async () => ({ + status: 0, + length: 1, + data: Buffer.from([1]), + }); + + const response = await new SimpleClient(raw).sendRawWithResponse( + COMMAND_CODE.Ping, + Buffer.alloc(0) + ); + + assert.deepEqual(response, Buffer.alloc(0)); + }); + +}); diff --git a/foreign/node/src/wire/command-set.ts b/foreign/node/src/wire/command-set.ts index 0388ecfc4..b3c3cc8fd 100644 --- a/foreign/node/src/wire/command-set.ts +++ b/foreign/node/src/wire/command-set.ts @@ -18,6 +18,9 @@ import type { ClientProvider } from '../client/client.type.js'; +import { COMMAND_CODE } from './command.code.js'; +import { responseError } from './error.utils.js'; + import { login } from './session/login.command.js'; import { logout } from './session/logout.command.js'; import { loginWithToken } from './session/login-with-token.command.js'; @@ -198,6 +201,15 @@ const clusterAPI = (c: ClientProvider) => ({ type ClusterAPI = ReturnType<typeof clusterAPI>; +const SESSION_CONTROL_CODES = new Set([ + COMMAND_CODE.LoginUser, + COMMAND_CODE.LogoutUser, + COMMAND_CODE.LoginRegister, + COMMAND_CODE.LoginWithAccessToken, + COMMAND_CODE.LoginRegisterWithAccessToken, +]); + +const INVALID_COMMAND_ERROR_CODE = 3; export abstract class AbstractAPI { clientProvider: ClientProvider; @@ -225,4 +237,20 @@ export abstract class CommandAPI extends AbstractAPI { constructor(c: ClientProvider) { super(c); } -}; + + /** + * Sends a command code with a payload and returns the raw response payload. + * Session-control codes are rejected with an invalid-command error. + * + * @param code - Command code to send + * @param payload - Raw command payload + * @returns Raw response payload + */ + async sendRawWithResponse(code: number, payload: Buffer): Promise<Buffer> { + if (SESSION_CONTROL_CODES.has(code)) + throw responseError(code, INVALID_COMMAND_ERROR_CODE); + + const response = await (await this.clientProvider()).sendCommand(code, payload); + return response.length <= 1 ? Buffer.alloc(0) : response.data; + } +} diff --git a/foreign/node/src/wire/command.code.ts b/foreign/node/src/wire/command.code.ts index e61d4a1d1..bfdcafcba 100644 --- a/foreign/node/src/wire/command.code.ts +++ b/foreign/node/src/wire/command.code.ts @@ -35,10 +35,12 @@ export const COMMAND_CODE = { ChangePassword: 37, LoginUser: 38, LogoutUser: 39, + LoginRegister: 40, GetAccessTokens: 41, CreateAccessToken: 42, DeleteAccessToken: 43, LoginWithAccessToken: 44, + LoginRegisterWithAccessToken: 45, PollMessages: 100, SendMessages: 101, FlushUnsavedBuffers: 102, diff --git a/foreign/php/Cargo.toml b/foreign/php/Cargo.toml index 0384b6563..1c8245e3e 100644 --- a/foreign/php/Cargo.toml +++ b/foreign/php/Cargo.toml @@ -17,7 +17,7 @@ [package] name = "iggy-php" -version = "0.1.0" +version = "0.1.1" edition = "2024" authors = ["Iggy Committers <[email protected]>"] license = "Apache-2.0" diff --git a/foreign/php/README.md b/foreign/php/README.md index ff6e4c6e0..a339d1f3c 100644 --- a/foreign/php/README.md +++ b/foreign/php/README.md @@ -214,6 +214,8 @@ iggy+tcp://iggy:[email protected]:8090?tls=true&tls_domain=localhost&tls_ca_file=/p be read repeatedly. - Large unsigned values that can overflow PHP integers, such as message checksums, are returned as decimal strings. +- `Iggy\Client::sendRawWithResponse(int $code, string $payload): string` sends a + command code and payload and returns the raw response body. - `Iggy\Client` is synchronous and blocks the current PHP thread. - The extension owns a lazy global Tokio runtime. Do not call `pcntl_fork()` after the first Iggy SDK call; the child process inherits file descriptors but not diff --git a/foreign/php/iggy-php.stubs.php b/foreign/php/iggy-php.stubs.php index b06d3ec92..f4be69725 100644 --- a/foreign/php/iggy-php.stubs.php +++ b/foreign/php/iggy-php.stubs.php @@ -217,6 +217,16 @@ namespace Iggy { * @return void */ public function sendMessages(mixed $stream, mixed $topic, int $partition_id, array $messages): void {} + + /** + * Sends a command code with a payload and returns the raw response bytes. + * Session-control codes return an invalid-command exception. + * + * @param int $code + * @param string $payload + * @return string + */ + public function sendRawWithResponse(int $code, string $payload): string {} } /** diff --git a/foreign/php/src/client.rs b/foreign/php/src/client.rs index 1f32e97d1..730f949d9 100644 --- a/foreign/php/src/client.rs +++ b/foreign/php/src/client.rs @@ -17,7 +17,8 @@ use std::{str::FromStr, sync::Arc}; -use ext_php_rs::{exception::PhpResult, php_class, php_impl}; +use bytes::Bytes; +use ext_php_rs::{binary::Binary, exception::PhpResult, php_class, php_impl}; use iggy::prelude::{ CompressionAlgorithm, Consumer as RustConsumer, IggyClient as RustIggyClient, IggyClientBuilder, IggyDuration, IggyExpiry, IggyMessage as RustMessage, MaxTopicSize, @@ -376,6 +377,20 @@ impl IggyClient { }) }) } + + /// Sends a command code with a payload and returns the raw response bytes. + /// Session-control codes return an invalid-command exception. + pub fn send_raw_with_response(&self, code: u32, payload: Binary<u8>) -> PhpResult<Binary<u8>> { + let inner = self.inner.clone(); + + runtime().block_on(async move { + inner + .send_binary_request(code, Bytes::from(Vec::<u8>::from(payload))) + .await + .map(|response| Binary::new(response.to_vec())) + .map_err(to_php_exception) + }) + } } fn non_zero_duration_micros(field: &str, micros: u64) -> PhpResult<IggyDuration> { diff --git a/foreign/php/tests/RawCommandTest.php b/foreign/php/tests/RawCommandTest.php new file mode 100644 index 000000000..eca074a23 --- /dev/null +++ b/foreign/php/tests/RawCommandTest.php @@ -0,0 +1,77 @@ +<?php +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +declare(strict_types=1); + +use Iggy\Exception\IggyException; +use PHPUnit\Framework\Attributes\DataProvider; +use PHPUnit\Framework\Attributes\TestDox; +use PHPUnit\Framework\TestCase; + +final class RawCommandTest extends TestCase +{ + private const PING_CODE = 1; + private const GET_STATS_CODE = 10; + private const UNKNOWN_CODE = 60_000; + + #[TestDox('A raw ping command returns an empty successful response')] + public function testRawPingReturnsEmptyResponse(): void + { + $client = new_client(); + + $response = $client->sendRawWithResponse(self::PING_CODE, ''); + + assert_same('', $response); + } + + #[TestDox('A raw get-stats command returns a non-empty response')] + public function testRawGetStatsReturnsNonEmptyResponse(): void + { + $client = new_client(); + + $response = $client->sendRawWithResponse(self::GET_STATS_CODE, ''); + + assert_true($response !== '', 'expected a non-empty stats response'); + } + + #[TestDox('A session-control code is rejected before reaching the server')] + #[DataProvider('sessionControlCodes')] + public function testRawSessionControlCodeIsRejected(int $code): void + { + $client = new_client(); + + $throwable = assert_throws(static fn () => $client->sendRawWithResponse($code, '')); + + assert_instance_of(IggyException::class, $throwable); + } + + #[TestDox('An unknown command code is rejected by the server')] + public function testRawUnknownCodeIsRejectedByServer(): void + { + $client = new_client(); + + $throwable = assert_throws(static fn () => $client->sendRawWithResponse(self::UNKNOWN_CODE, '')); + + assert_instance_of(IggyException::class, $throwable); + } + + public static function sessionControlCodes(): array + { + return [[38], [39], [40], [44], [45]]; + } +} diff --git a/foreign/python/Cargo.toml b/foreign/python/Cargo.toml index 86138351a..945b6aba9 100644 --- a/foreign/python/Cargo.toml +++ b/foreign/python/Cargo.toml @@ -17,7 +17,7 @@ [package] name = "apache-iggy" -version = "0.8.1-dev2" +version = "0.8.1-dev3" edition = "2024" authors = ["Iggy Committers <[email protected]>"] license = "Apache-2.0" diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi index 5b5366938..facd0961b 100644 --- a/foreign/python/apache_iggy.pyi +++ b/foreign/python/apache_iggy.pyi @@ -635,6 +635,25 @@ class IggyClient: Creates a new consumer group consumer. Returns the consumer or a PyRuntimeError on failure. """ + def send_raw_with_response( + self, code: builtins.int, payload: builtins.bytes + ) -> collections.abc.Awaitable[bytes]: + r""" + Send a command code with a payload and return the raw response bytes. + + Session-control codes are rejected client-side. HTTP transport does not + support raw binary commands. + + Args: + code: Command code as `int`. + payload: Request payload as `bytes`. + + Returns: + An awaitable that resolves to the raw response `bytes`. + + Raises: + PyRuntimeError: If the command cannot be sent or the server returns an error. + """ @typing.final class IggyConsumer: diff --git a/foreign/python/pyproject.toml b/foreign/python/pyproject.toml index b36bff18b..785d9248d 100644 --- a/foreign/python/pyproject.toml +++ b/foreign/python/pyproject.toml @@ -22,7 +22,7 @@ build-backend = "maturin" [project] name = "apache-iggy" requires-python = ">=3.10" -version = "0.8.1.dev2" +version = "0.8.1.dev3" description = "Apache Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second." readme = "README.md" license = { file = "LICENSE" } diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs index 2a36ef333..a2c68db65 100644 --- a/foreign/python/src/client.rs +++ b/foreign/python/src/client.rs @@ -15,13 +15,14 @@ // specific language governing permissions and limitations // under the License. +use bytes::Bytes; use iggy::prelude::{ Consumer as RustConsumer, IggyClient as RustIggyClient, IggyMessage as RustMessage, PollingStrategy as RustPollingStrategy, *, }; use pyo3::PyRef; use pyo3::prelude::*; -use pyo3::types::{PyDelta, PyList, PyType}; +use pyo3::types::{PyBytes, PyDelta, PyList, PyType}; use pyo3_async_runtimes::tokio::future_into_py; use pyo3_stub_gen::define_stub_info_gatherer; use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods}; @@ -823,6 +824,37 @@ impl IggyClient { }) }) } + + /// Send a command code with a payload and return the raw response bytes. + /// + /// Session-control codes are rejected client-side. HTTP transport does not + /// support raw binary commands. + /// + /// Args: + /// code: Command code as `int`. + /// payload: Request payload as `bytes`. + /// + /// Returns: + /// An awaitable that resolves to the raw response `bytes`. + /// + /// Raises: + /// PyRuntimeError: If the command cannot be sent or the server returns an error. + #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[bytes]", imports=("collections.abc")))] + fn send_raw_with_response<'a>( + &self, + py: Python<'a>, + code: u32, + #[gen_stub(override_type(type_repr = "builtins.bytes"))] payload: Vec<u8>, + ) -> PyResult<Bound<'a, PyAny>> { + let inner = self.inner.clone(); + future_into_py(py, async move { + let response = inner + .send_binary_request(code, Bytes::from(payload)) + .await + .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?; + Ok(Python::attach(|py| PyBytes::new(py, &response).unbind())) + }) + } } define_stub_info_gatherer!(stub_info); diff --git a/foreign/python/tests/test_raw_command.py b/foreign/python/tests/test_raw_command.py new file mode 100644 index 000000000..467dd252c --- /dev/null +++ b/foreign/python/tests/test_raw_command.py @@ -0,0 +1,47 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +import pytest + +from apache_iggy import IggyClient + + [email protected] +async def test_raw_ping_returns_empty_response(iggy_client: IggyClient): + response = await iggy_client.send_raw_with_response(1, b"") + + assert response == b"" + + [email protected] +async def test_raw_get_stats_returns_non_empty_response(iggy_client: IggyClient): + response = await iggy_client.send_raw_with_response(10, b"") + + assert response + + [email protected] [email protected]("code", [38, 39, 40, 44, 45]) +async def test_raw_session_control_code_is_rejected(iggy_client: IggyClient, code: int): + with pytest.raises(RuntimeError, match="(?i)invalid command"): + await iggy_client.send_raw_with_response(code, b"") + + [email protected] +async def test_raw_unknown_code_is_rejected_by_server(iggy_client: IggyClient): + with pytest.raises(RuntimeError, match="(?i)invalid command"): + await iggy_client.send_raw_with_response(60_000, b"") diff --git a/foreign/python/uv.lock b/foreign/python/uv.lock index 3beb7ab49..55960c278 100644 --- a/foreign/python/uv.lock +++ b/foreign/python/uv.lock @@ -12,7 +12,7 @@ exclude-newer-span = "P7D" [[package]] name = "apache-iggy" -version = "0.8.1.dev2" +version = "0.8.1.dev3" source = { editable = "." } [package.optional-dependencies] diff --git a/raw-command-iggy-sdk.md b/raw-command-iggy-sdk.md new file mode 100644 index 000000000..004046d21 --- /dev/null +++ b/raw-command-iggy-sdk.md @@ -0,0 +1,461 @@ +# Raw command API across Iggy SDKs: implementation plan + +Goal: every official Iggy SDK exposes a public method that sends an +arbitrary command code (`u32`) plus an opaque byte payload to the server +and returns the raw response bytes. This mirrors what the Rust SDK +already ships and is the extension point for forked servers that add +custom command handlers. + +## 1. Reference: what Rust already has + +### 1.1 API surface + +| Layer | Item | Location | +| --- | --- | --- | +| Transport trait | `BinaryTransport::send_raw_with_response(&self, code: u32, payload: Bytes) -> Result<Bytes, IggyError>` | `core/common/src/traits/binary_transport.rs:31` | +| Trait impls | TCP / QUIC / WebSocket | `core/sdk/src/tcp/tcp_client.rs:150`, `core/sdk/src/quic/quic_client.rs:139`, `core/sdk/src/websocket/websocket_client.rs:138` | +| Public user API | `IggyClient::send_binary_request(&self, code: u32, payload: Bytes) -> Result<Bytes, IggyError>` (inherent method, no trait import needed) | `core/sdk/src/clients/client.rs:210` | +| HTTP counterpart | `IggyClient::send_http_request(...)` (REST path, binary raw send is unavailable over HTTP) | `core/sdk/src/clients/client.rs:224` | + +Behavior of `send_binary_request` (the semantics every port must copy): + +1. Session-control codes are rejected client-side with + `IggyError::InvalidCommand` before hitting the wire. The list is + `SESSION_CONTROL_CODES` (`core/sdk/src/clients/client.rs:50`): + `LOGIN_USER_CODE = 38`, `LOGOUT_USER_CODE = 39`, + `LOGIN_REGISTER_CODE = 40`, + `LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE = 44`, + `LOGIN_REGISTER_WITH_PAT_CODE = 45`. Rationale: session mutation + must go through the typed login/logout methods so client session + state (and VSR dedup counters) stay correct. +2. TCP / QUIC / WebSocket dispatch to `send_raw_with_response`. +3. HTTP transport returns `IggyError::FeatureUnavailable` (there is no + binary frame concept over REST). +4. Under the `vsr` feature, unknown codes are rejected client-side + (`operation_for_code`, `core/sdk/src/vsr.rs:147`): + unknown -> `InvalidCommand`, replicated code without an operation + mapping -> `UnknownReplicatedCommand` (14002). Custom codes are a + classic-protocol feature. + +### 1.2 Wire format (classic TCP binary protocol, QUIC/WS identical framing) + +Request: + +```text +[ length : u32 LE ] 4 bytes, = 4 (code field) + payload.len() +[ code : u32 LE ] 4 bytes +[ payload: N bytes ] +``` + +Response: + +```text +[ status : u32 LE ] 4 bytes, 0 = OK +[ length : u32 LE ] 4 bytes, response payload length +[ payload: length bytes ] length <= 1 is treated as empty +``` + +Non-zero status maps to `IggyError::from_code(status)`. Source: +`core/sdk/src/tcp/tcp_client.rs:789-829` and `handle_response` at +`:324`. Every foreign SDK already implements exactly this framing in +its own send path, so no wire work is needed anywhere. + +### 1.3 Server behavior for arbitrary codes + +- Classic server: closed `match frame.code` in + `core/server/src/binary/dispatch.rs:150`. An unknown code logs and + returns `IggyError::InvalidCommand` (`dispatch.rs:451-454`). Forked + servers add match arms there. There is no runtime handler registry. +- server-ng: code routed at `core/server-ng/src/dispatch.rs:1133`, + fallback at `:1244`. The command metadata registry is in + `core/binary_protocol/src/dispatch.rs:204` (`lookup_command`). +- Consequence for tests against a stock server: a raw round-trip can + assert (a) known codes work (`PING_CODE = 1` -> empty OK, + `GET_STATS_CODE = 10` -> non-empty body) and (b) an unknown code + (e.g. `60_000`) returns the invalid-command error. + +### 1.4 Existing tests (Rust) + +- Integration: `core/integration/tests/sdk/raw.rs`, transport matrix + `[Tcp, Quic, Http, WebSocket]` plus a vsr variant. Asserts ping + round-trip, non-empty stats body, login code rejected, unknown code + rejected under vsr, HTTP -> `FeatureUnavailable`. This file is the + canonical template for every other SDK's test. +- Unit: `core/sdk/src/clients/client.rs:470-497` (HTTP rejection, + session-code rejection). +- BDD: none. Examples: none. (Gaps addressed in sections 4 and 5.) + +## 2. Cross-SDK contract + +Every SDK adds one public method with language-native casing and types: + +| SDK | Signature | +| --- | --- | +| Rust | `send_binary_request(code: u32, payload: Bytes) -> Result<Bytes, IggyError>` (exists) | +| Python | `async def send_raw_with_response(code: int, payload: bytes) -> bytes` | +| Node | `sendRawWithResponse(code: number, payload: Buffer): Promise<Buffer>` | +| Java async | `CompletableFuture<byte[]> sendRawWithResponse(int code, byte[] payload)` | +| Java blocking | `byte[] sendRawWithResponse(int code, byte[] payload)` | +| C# | `Task<byte[]> SendRawWithResponseAsync(uint code, byte[] payload, CancellationToken token = default)` | +| Go | `SendRawWithResponse(ctx context.Context, code uint32, payload []byte) ([]byte, error)` | +| PHP | `sendRawWithResponse(int $code, string $payload): string` | +| C++ | `rust::Vec<uint8_t> send_raw_with_response(uint32_t code, rust::Vec<uint8_t> payload)` (via cxx bridge) | + +Required semantics, identical everywhere: + +1. Success (status 0): return response payload bytes. An empty payload + (length <= 1) returns an empty byte value, never null. +2. Non-zero status: raise/return the SDK's native Iggy error carrying + the numeric server error code (each SDK's existing error mapping + already does this in its send path. Reuse it instead of inventing new + error types). +3. Reject session-control codes {38, 39, 40, 44, 45} client-side with + the SDK's invalid-command error, same rationale as Rust. Define the + set as a private constant next to the method. +4. Where the SDK also has an HTTP transport behind the same public + interface (Rust, Java, C#), the HTTP implementation fails with the + SDK's feature-unavailable / operation-not-supported error. +5. Auth: the method goes through the SDK's normal send path, so + whatever auto-auth or auth gating the SDK applies to typed commands + applies here too. The per-SDK notes document this behavior. +6. Document the opaque request and response payloads, session-control + rejection, and HTTP limitation where applicable. Follow each SDK's + existing public API documentation style. + +Naming note: the Rust public method is `send_binary_request` while the +trait method is `send_raw_with_response`. For the ports we standardize +on the `send_raw_with_response` name (it describes the operation and +avoids implying an HTTP/binary dichotomy that only Rust and C# have). +Rust keeps its existing names. Its documentation may note the +correspondence. + +## 3. Per-SDK implementation plans + +Ordered roughly by effort (smallest first). All are independent and +can be done in parallel. Each item lists every file to touch. + +### 3.1 Python (`foreign/python/`, pyo3 wrapper over Rust SDK) + +State: wraps `iggy::prelude::IggyClient` as `Arc<RustIggyClient>` +(`foreign/python/src/client.rs:45-49`). Nothing raw is exposed today. +The Rust seam (`send_binary_request`) is directly reachable. + +Changes: + +1. `foreign/python/src/client.rs`: add to the + `#[gen_stub_pymethods] #[pymethods] impl IggyClient` block (ends + `:826`) an async binding following the existing pattern (clone + `Arc`, `pyo3_async_runtimes::tokio::future_into_py`): + - accept `code: u32, payload: Vec<u8>` (pyo3 converts `bytes`), + build `Bytes::from(payload)`, + - call `inner.send_binary_request(code, payload).await`, map + `IggyError` to `PyRuntimeError` like every other method, + - return `PyBytes::new(py, &response)` inside `Python::attach` + (pattern: `foreign/python/src/receive_message.rs:37-38`). A bare + `Vec<u8>` would surface as `list[int]`, + - annotate + `#[gen_stub(override_return_type(type_repr = "collections.abc.Awaitable[bytes]", imports = ("collections.abc")))]`. + Session-code rejection and HTTP handling come for free from the + wrapped Rust method. +2. Regenerate `foreign/python/apache_iggy.pyi` via the `stub_gen` bin + (`foreign/python/src/bin/stub_gen.rs`). Do not hand-edit. +3. No `lib.rs` change (methods on registered classes need no wiring). + +Tests: new `foreign/python/tests/test_raw_command.py` following +`test_message_operations.py` conventions: ping code 1 with the ping +request payload returns empty bytes. Stats code 10 returns non-empty +bytes. Login code 38 raises. Unknown code 60000 raises a server error. + +### 3.2 PHP (`foreign/php/`, ext-php-rs wrapper over Rust SDK) + +State: wraps `Arc<RustIggyClient>` (`foreign/php/src/client.rs:41`). +no raw surface. Marked experimental but usable end to end. + +Changes: + +1. `foreign/php/src/client.rs`: add to `#[php_impl] impl IggyClient`: + `pub fn send_raw_with_response(&self, code: u32, payload: Binary<u8>) -> PhpResult<Binary<u8>>` + using `runtime().block_on(...)` around + `inner.send_binary_request(code, Bytes::from(Vec::<u8>::from(payload)))`, + errors through the existing `to_php_exception` mapper + (`foreign/php/src/error.rs:55`). `Binary<u8>` is already used by + `send_message.rs:19`. Method surfaces to PHP as + `sendRawWithResponse` (ext-php-rs camelCases). +2. Regenerate `foreign/php/iggy-php.stubs.php` with `cargo php stubs`. + CI fails on stub drift (`foreign/php/README.md:39`). +3. `foreign/php/README.md`: one short API Notes entry. + +Tests: extend `foreign/php/tests/IggySdkTest.php` (or a new +`RawCommandTest.php`) with the same four assertions as Python. + +### 3.3 C++ (`foreign/cpp/`, cxx bridge over Rust SDK) + +State: usable at the low level (~40 client functions, ~242 e2e test +cases, BDD wired). Producer/Consumer are stubs but irrelevant here. +The client layer is complete enough to include. The change is two files. + +Changes: + +1. `foreign/cpp/src/lib.rs`: add to the `extern "Rust"` block (near + `:382`): + `fn send_raw_with_response(self: &Client, code: u32, payload: Vec<u8>) -> Result<Vec<u8>>;` + (cxx has no `Bytes` type, so `Vec<u8>` crosses the FFI). +2. `foreign/cpp/src/client.rs`: implement on `impl Client` with + `RUNTIME.block_on(async { self.inner.send_binary_request(code, Bytes::from(payload)).await })`, + mapping the error like sibling methods. `bytes` is already a dep + (`foreign/cpp/Cargo.toml:30`). + +Tests: extend `foreign/cpp/tests/client/low_level_e2e.cpp` with the +four standard assertions (helpers in `tests/common/test_helpers.hpp`). + +### 3.4 Node (`foreign/node/`, pure TypeScript protocol impl) + +State: framing and raw send already exist and are even public via +`getRawClient(cfg).sendCommand(code, payload)` +(`src/client/client.socket.ts:111`, exported through the barrel), but +there is no ergonomic method on the `Client` classes and the return is +the full `{status, length, data}` object. + +Changes: + +1. `foreign/node/src/wire/command-set.ts`: add to `CommandAPI` (or + `AbstractAPI`, `:202-228`): + + ```ts + async sendRawWithResponse(code: number, payload: Buffer): Promise<Buffer> { + // session-control guard here: throw on {38, 39, 40, 44, 45} + const r = await (await this.clientProvider()).sendCommand(code, payload); + return r.data; + } + ``` + + `sendCommand` with default `handleResponse = true` already rejects + non-zero status via `responseError` and strips the 8-byte header, + matching the contract. Method lands on `Client`, `SingleClient`, + and `SimpleClient` at once (all extend `CommandAPI`). +2. Auth note: codes outside `UNLOGGED_COMMAND_CODE` + (`src/client/client.socket.ts:35-39`) trigger auto-login first. + Document this behavior in the method's JSDoc. +3. Optional: re-export `COMMAND_CODE` from `src/wire/index.ts` so + users can pass named codes (currently not part of the public API). + +Tests: unit test beside `src/client/client.utils.test.ts` for the +guard + framing, plus `foreign/node/src/e2e/tcp.raw.e2e.ts` with the +four standard assertions (pattern: existing `src/e2e/tcp.system.e2e.ts`). + +### 3.5 Go (`foreign/go/`, pure Go protocol impl, TCP only) + +State: no public escape hatch. Seam is the unexported +`do(ctx, cmd)` / `sendWireAndFetchResponse(ctx, wirePayload)` pair in +`foreign/go/client/tcp/tcp_core.go:292/:333`. Commands live in an +`internal/` package so users cannot inject custom codes today. + +Changes: + +1. `foreign/go/contracts/client.go`: add to the `Client` interface + (around `:293`): + `SendRawWithResponse(ctx context.Context, code uint32, payload []byte) ([]byte, error)`. +2. `foreign/go/client/tcp/` (new `tcp_raw.go` or in `tcp_core.go`): + implement on `*IggyTcpClient` by building the frame inline the way + `encodeWireRequest` (`tcp_core.go:311`) does: acquire pooled buffer + (`acquireRequestBuf`, `:241`), 4-byte LE length + 4-byte LE code + + payload, then `c.sendWireAndFetchResponse(ctx, buf)` and release. + This reuses the existing status/length/body read path unchanged. + Add the session-control guard before sending. +3. `foreign/go/client/iggy_client.go`: no change. `IggyClient` embeds + `iggcon.Client` (`:72`) so the method is inherited. + +Tests: extend `foreign/go/client/tcp/tcp_core_test.go` using +`newTestClient` (`:41`, `net.Pipe`) and `serverRespondCapture` +(`:176`): assert emitted frame layout (length/code/payload), success +body pass-through, non-zero status -> `ierror.FromCode`, guard codes +rejected without any write. + +### 3.6 C# (`foreign/csharp/`, pure .NET impl, TCP + HTTP) + +State: private seam `TcpMessageStream.SendRawAsync` / +`SendWithResponseAsync` (`Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs:1124/:1072`) +already does framing, status parsing, error throw, body read. Nothing +public. `CommandCodes` is `internal` (`Iggy_SDK/Utils/CommandCodes.cs:20`). + +Changes: + +1. `foreign/csharp/Iggy_SDK/IggyClient/IIggyClient.cs`: add + `Task<byte[]> SendRawWithResponseAsync(uint code, byte[] payload, CancellationToken token = default);` + directly on `IIggyClient` (`:27`), not on a role interface. + Precedent for `Task<byte[]>`: `IIggySystem.GetSnapshotAsync`. +2. `TcpMessageStream.cs`: implement: session-control guard, allocate + `new byte[4 + BufferSizes.INITIAL_BYTES_LENGTH + payload.Length]`, + `TcpMessageStreamHelpers.CreatePayload(buffer, payload, (int)code)` + (`Utils/TcpMessageStreamHelpers.cs:30`). Either cast or widen the + helper's `int command` parameter to `uint`), then + `using var response = await SendWithResponseAsync(buffer, token);` + and return `response.Memory.ToArray()`. Error mapping is already + inside `SendRawAsync`. +3. `HttpMessageStream.cs`: implement the new interface member as + `throw new FeatureUnavailableException();` (precedent at + `HttpMessageStream.cs:462/:575/:591/:620`). Mandatory or the + project will not compile. + +Tests: new `foreign/csharp/Iggy_SDK.Tests.Integration/RawCommandTests.cs` +run against both transports with the existing `SkipHttp`/`SkipTcp` +attributes: TCP gets the four standard assertions, HTTP asserts +`FeatureUnavailableException`. Optional unit test for the guard in +`Iggy_SDK_Tests/`. + +### 3.7 Java (`foreign/java/`, Netty async TCP + blocking wrapper + HTTP) + +State: `AsyncTcpConnection.send(int commandCode, ByteBuf payload)` is +public and already returns the response payload with status handling +(`java-sdk/.../client/async/tcp/AsyncTcpConnection.java:195`), but the +connection is a private field of `AsyncIggyTcpClient` with no getter, +so users cannot reach it. + +Changes: + +1. `java-sdk/.../client/async/tcp/AsyncIggyTcpClient.java`: add + `public CompletableFuture<byte[]> sendRawWithResponse(int code, byte[] payload)`: + session-control guard, null-connection guard like the accessors at + `:226-231`, `connection.send(code, Unpooled.wrappedBuffer(payload))`, + then `thenApply` copying the `ByteBuf` to `byte[]` and releasing it + (pattern: `SystemTcpClient.java:54-60`). +2. `java-sdk/.../client/blocking/tcp/IggyTcpClient.java`: add + `public byte[] sendRawWithResponse(int code, byte[] payload)` + delegating via `FutureUtil.resolve(...)` (pattern: `login()`, + `:92-94`). +3. Interface decision (recommended): also add the method to + `client/blocking/IggyBaseClient.java:22` so it is reachable through + the interface users hold, and implement it in + `client/blocking/http/IggyHttpClient.java` as + `throw new IggyOperationNotSupportedException(...)` (type exists). + This matches Rust and C# behavior. The TCP-only alternative (skip + the interface) is smaller but hides the method from anyone coding + against `IggyBaseClient`. +4. Auth caveat to document: `AsyncTcpConnection.send` gates + non-login/PING/GET_STATS codes on `authenticated` + (`AsyncTcpConnection.java:207-236`). Raw sends require a prior + `login()`, consistent with the other SDKs' behavior in practice. + +Tests: new `java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/RawCommandTcpClientTest.java` +using the existing Testcontainers integration base +(`client/blocking/IntegrationTest.java`), four standard assertions, +plus an HTTP-side test asserting the unsupported exception if the +interface route is taken. Async coverage in +`client/async/AsyncClientIntegrationTest.java` style. + +### 3.8 Rust (reference, small follow-ups only) + +1. Optional rustdoc touch-up on `send_binary_request` noting it is the + canonical raw-command API mirrored by all SDKs. +2. BDD steps for the new shared feature (section 4), since + `bdd/rust/` has no raw coverage today. + +## 4. Shared BDD feature (cross-SDK) + +New file `bdd/scenarios/raw_command.feature` (third shared feature, +next to `basic_messaging.feature` and `leader_redirection.feature`): + +```gherkin +Feature: Raw command API + Background: + Given a running Iggy server + And an authenticated root client + + Scenario: Known command round-trips through the raw API + When I send a raw command with code 1 and an empty ping payload + Then I should receive an empty successful raw response + When I send a raw command with code 10 and an empty payload + Then I should receive a non-empty raw response + + Scenario: Unknown command code is rejected by the server + When I send a raw command with code 60000 and an empty payload + Then the raw command should fail with an invalid command error + + Scenario: Session-control code is rejected client-side + When I send a raw command with code 38 and an empty payload + Then the raw command should fail before reaching the server +``` + +Wording of steps to be finalized when writing the first (Rust) +implementation. The ping payload for code 1 is empty on the classic +protocol (`PingRequest` serializes to empty bytes), so all steps can +share one generic "code + empty payload" step. The 60000 unknown-code +scenario relies on stock-server behavior (`dispatch.rs:451`). + +Step definitions per runner (same files that hold basic messaging +steps): + +| Lang | File to extend | +| --- | --- | +| rust | `bdd/rust/tests/steps/` (new `raw_command.rs`, register in `steps/mod.rs`, new test entry `bdd/rust/tests/raw_command.rs`) | +| python | `bdd/python/tests/` (new `test_raw_command.py`, pytest-bdd) | +| go | `bdd/go/tests/` (new `raw_command.go` + suite entry in `suite_test.go`) | +| node | `foreign/node/src/bdd/` (new `raw.ts`) | +| csharp | `foreign/csharp/Iggy_SDK.Tests.BDD/StepDefinitions/` (new `RawCommandSteps.cs`) | +| java | `bdd/java/src/test/java/org/apache/iggy/bdd/` (new `RawCommandSteps.java`) | +| php | `bdd/php/tests/` (new `RawCommandFeatureTest.php`) | +| cpp | `bdd/cpp/features/step_definitions/` (new `raw_command_steps.cpp`) | + +Script/compose wiring: + +- `scripts/run-bdd-tests.sh`: add `raw_command` to the valid feature + list (`:48-54`) and to the compose-file selection so it pulls + `docker-compose.server.yml` like `basic_messaging` does. No cluster + compose needed. Unlike `leader_redirection`, no language gate: all + eight runners implement it (gate temporarily if rollout is staged). +- Runners that enumerate features explicitly need their config touched + (go `suite_test.go` reads `BDD_FEATURE`, while csharp uses + `--filter-trait Category=...` in `bdd/docker-compose.yml:117-123`). + java/cucumber and node/cucumber pick up features from the shared + directory automatically. Verify each during implementation. + +## 5. Test matrix summary + +| SDK | Unit/local | Integration (real server) | BDD | +| --- | --- | --- | --- | +| Rust | exists (`client.rs:470-497`) | exists (`core/integration/tests/sdk/raw.rs`) | add steps | +| Python | - | add `tests/test_raw_command.py` | add steps | +| Node | add guard/framing unit test | add `src/e2e/tcp.raw.e2e.ts` | add steps | +| Java | optional guard test | add `RawCommandTcpClientTest` (+ HTTP unsupported) | add steps | +| C# | optional guard test | add `RawCommandTests` (TCP + HTTP unavailable) | add steps | +| Go | add `tcp_core` pipe tests (frame layout, status, guard) | covered by BDD | add steps | +| PHP | - | add PHPUnit raw test | add steps | +| C++ | - | extend `low_level_e2e.cpp` | add steps | + +Standard four assertions everywhere a real server is available: + +1. code 1 (PING), empty payload -> empty success response. +2. code 10 (GET_STATS), empty payload -> non-empty response. +3. code 38 (LOGIN_USER) -> rejected client-side, nothing sent. +4. code 60000 -> server responds with invalid-command error. + +## 6. Suggested execution order + +1. **Phase 1, SDK methods** (independent, parallelizable): + Python and PHP and C++ (thin Rust-wrapper plumbing, hours each), + Node (one method on `CommandAPI`), Go (interface + tcp impl), + C# (interface + two impls), Java (async + blocking + interface + decision). Each lands with its own local/integration tests from + section 5. +2. **Phase 2, shared BDD**: add `raw_command.feature`, Rust steps + first (validates the feature wording against the stock server), + then the other seven runners, then `run-bdd-tests.sh` wiring. +3. **Phase 3, docs (optional)**: short raw-command paragraph in each + SDK README and a runnable example under `examples/<lang>/` if + desired. Rust currently has no example either, so this is a + uniform, deferrable follow-up. + +## 7. Open decisions + +1. **Java interface placement**: `IggyBaseClient` + HTTP throws + (recommended, matches Rust/C#) vs TCP-only classes. Affects + plan 3.7 step 3. +2. **Session-control guard set**: plan says mirror all five codes + {38, 39, 40, 44, 45}. Codes 40/45 are VSR login-register codes that + foreign SDKs (classic protocol only) will never see legitimately. + guarding them anyway is free and future-proof. +3. **Exposing command-code constants publicly** (Node `COMMAND_CODE` + export, C# `CommandCodes` visibility): nice-to-have, not required + since the API takes numeric codes. Default: skip in phase 1. +4. **C++ scope**: included because the client layer is mature and the + change is two files. Drop to phase 3 if C++ effort is constrained. diff --git a/scripts/run-bdd-tests.sh b/scripts/run-bdd-tests.sh index 148a564b3..29e0124d9 100755 --- a/scripts/run-bdd-tests.sh +++ b/scripts/run-bdd-tests.sh @@ -36,7 +36,7 @@ usage(){ log "Usage: $0 [--coverage] <sdk> [feature]" log "" log " sdk: rust | python | php | go | go-race | node | csharp | java | cpp | all | clean (default: all)" - log " feature: basic_messaging | leader_redirection | all (default: all)" + log " feature: basic_messaging | leader_redirection | raw_command | all (default: all)" log "" log "Examples:" log " $0 rust # run all features for Rust" @@ -46,7 +46,7 @@ usage(){ } case "$FEATURE" in - basic_messaging|leader_redirection|all) ;; + basic_messaging|leader_redirection|raw_command|all) ;; *) log "Unknown feature: ${FEATURE}" usage @@ -66,7 +66,7 @@ ALL_COMPOSE_FILES=( COMPOSE_FILES=(-f docker-compose.yml) case "$FEATURE" in - basic_messaging|leader_redirection|all) + basic_messaging|leader_redirection|raw_command|all) COMPOSE_FILES+=(-f docker-compose.server.yml) ;; esac case "$FEATURE" in
