diff --git a/.dockerignore b/.dockerignore index 30d0a2bb62..d724e1982f 100644 --- a/.dockerignore +++ b/.dockerignore @@ -17,6 +17,8 @@ **/node_modules **/.gradle /foreign/swift/.build +/bdd/swift/**/.build +/examples/swift/**/.build /foreign/java/**/build /bdd/java/build /examples/java/**/build diff --git a/.github/actions/swift/pre-merge/action.yml b/.github/actions/swift/pre-merge/action.yml index bb37f83eec..2043b5e1a5 100644 --- a/.github/actions/swift/pre-merge/action.yml +++ b/.github/actions/swift/pre-merge/action.yml @@ -33,13 +33,18 @@ runs: uses: ./.github/actions/utils/setup-swift # swift-format ships with the toolchain; the configuration lives in - # foreign/swift/.swift-format. + # foreign/swift/.swift-format and covers the examples and the BDD runner + # too, which compile in their own lanes. - name: Format Check shell: bash if: inputs.task == 'lint' run: | cd foreign/swift swift format lint --strict --recursive Sources Tests Examples Package.swift + cd ../.. + for package in examples/swift/getting-started bdd/swift/tests; do + swift format lint --strict --recursive --configuration foreign/swift/.swift-format "$package/Sources" "$package/Package.swift" + done # A warning-free build is the lint gate the Swift compiler provides: the # SDK builds in Swift 6 language mode with strict concurrency checking. diff --git a/.github/actions/utils/setup-swift/action.yml b/.github/actions/utils/setup-swift/action.yml index 24a200926d..06cd9734f8 100644 --- a/.github/actions/utils/setup-swift/action.yml +++ b/.github/actions/utils/setup-swift/action.yml @@ -22,6 +22,10 @@ inputs: description: "Cache key segment naming the package that builds in the job; one constant per package so every lane shares one cache" required: false default: "sdk" + package-path: + description: "Directory of the SwiftPM package whose build directory is cached" + required: false + default: "foreign/swift" runs: using: "composite" @@ -83,11 +87,11 @@ runs: uses: actions/cache@v6.1.0 with: path: | - foreign/swift/.build/checkouts - foreign/swift/.build/repositories - foreign/swift/.build/artifacts - foreign/swift/.build/debug - foreign/swift/.build/release - key: swiftpm-${{ runner.os }}-${{ steps.swift.outputs.version }}-${{ inputs.cache-key }}-${{ hashFiles('foreign/swift/Package.resolved') }} + ${{ inputs.package-path }}/.build/checkouts + ${{ inputs.package-path }}/.build/repositories + ${{ inputs.package-path }}/.build/artifacts + ${{ inputs.package-path }}/.build/debug + ${{ inputs.package-path }}/.build/release + key: swiftpm-${{ runner.os }}-${{ steps.swift.outputs.version }}-${{ inputs.cache-key }}-${{ hashFiles(format('{0}/Package.resolved', inputs.package-path)) }} restore-keys: | swiftpm-${{ runner.os }}-${{ steps.swift.outputs.version }}-${{ inputs.cache-key }}- diff --git a/.github/config/components.yml b/.github/config/components.yml index b0df5eb2d6..7a217cb4a7 100644 --- a/.github/config/components.yml +++ b/.github/config/components.yml @@ -472,12 +472,13 @@ components: bdd-swift: depends_on: - "rust-server" + - "rust-sdk" - "sdk-swift" - "bdd-infrastructure" - "ci-infrastructure" paths: - "bdd/swift/**" - - "bdd/scenarios/deferred_polling.feature" + - "bdd/scenarios/**" tasks: ["bdd-swift"] examples-rust: @@ -563,6 +564,18 @@ components: - "scripts/utils.sh" tasks: ["examples-java"] + examples-swift: + depends_on: + - "rust-server" + - "rust-sdk" # All SDKs depend on core SDK changes + - "sdk-swift" + - "ci-infrastructure" + paths: + - "examples/swift/**" + - "scripts/run-examples-from-readme.sh" + - "scripts/utils.sh" + tasks: ["examples-swift"] + web-ui: paths: - "web/**" diff --git a/.github/workflows/_test_examples.yml b/.github/workflows/_test_examples.yml index 057d73493f..f6b0220613 100644 --- a/.github/workflows/_test_examples.yml +++ b/.github/workflows/_test_examples.yml @@ -91,6 +91,18 @@ jobs: with: download-deps: "false" + - name: Setup Swift for examples + if: matrix.task == 'examples-swift' + uses: ./.github/actions/utils/setup-swift + with: + cache-key: "examples" + package-path: "examples/swift/getting-started" + + # Built before the server starts, so each example run only executes. + - name: Build Swift examples + if: matrix.task == 'examples-swift' + run: swift build --package-path examples/swift/getting-started -Xswiftc -warnings-as-errors + - name: Test Go examples if: matrix.task == 'examples-go' working-directory: examples/go @@ -178,6 +190,10 @@ jobs: if: matrix.task == 'examples-java' run: ./scripts/run-examples-from-readme.sh --language java + - name: Run Swift examples + if: matrix.task == 'examples-swift' + run: ./scripts/run-examples-from-readme.sh --language swift + - name: Upload reports if: always() uses: actions/upload-artifact@v7 diff --git a/README.md b/README.md index d30d8b532a..e21f035e79 100644 --- a/README.md +++ b/README.md @@ -159,7 +159,7 @@ This checkout moves the binary protocol to 0.12.0 and the metadata snapshot form - [Node.js (TypeScript)](https://www.npmjs.com/package/apache-iggy) - [Go](https://pkg.go.dev/github.com/apache/iggy/foreign/go) -[C++](https://github.com/apache/iggy/tree/master/foreign/cpp) and [PHP](https://github.com/apache/iggy/tree/master/foreign/php) are work in progress. +[C++](https://github.com/apache/iggy/tree/master/foreign/cpp), [PHP](https://github.com/apache/iggy/tree/master/foreign/php), and [Swift](https://github.com/apache/iggy/tree/master/foreign/swift) are work in progress. --- diff --git a/bdd/README.md b/bdd/README.md index a1507ae336..ce47163ee8 100644 --- a/bdd/README.md +++ b/bdd/README.md @@ -35,6 +35,9 @@ bdd/ │ ├── Dockerfile # PHP BDD test container │ ├── phpunit.xml.dist │ └── tests/ +├── swift/ # Swift SDK BDD implementation +│ ├── Dockerfile # Swift BDD test container +│ └── tests/ # SwiftPM package with the Gherkin runner and steps ├── docker-compose.yml # Base: SDK test clients (always included) ├── docker-compose.server.yml # Single iggy-server test setup ├── docker-compose.cluster.yml # Leader + follower test setup @@ -50,7 +53,7 @@ bdd/ ```bash # Usage: ../scripts/run-bdd-tests.sh [--coverage] [feature] -# sdk: rust | python | php | go | go-race | node | csharp | java | all | clean (default: all) +# sdk: rust | python | php | go | go-race | node | csharp | java | cpp | swift | all | clean (default: all) # feature: basic_messaging | leader_redirection | raw_command | session_contract | stream_crud | all (default: all) # Run all features for all SDKs @@ -64,6 +67,7 @@ bdd/ ../scripts/run-bdd-tests.sh csharp ../scripts/run-bdd-tests.sh java ../scripts/run-bdd-tests.sh php +../scripts/run-bdd-tests.sh swift # Run only basic_messaging feature for Rust SDK ../scripts/run-bdd-tests.sh rust basic_messaging @@ -135,6 +139,14 @@ The Java implementation is located in `java/src/test/` The PHP implementation is located in `bdd/php/tests/` and runs through PHPUnit. +### For Swift SDK + +The Swift implementation is located in `bdd/swift/tests/`, a SwiftPM package with a small Gherkin +runner that walks the shared feature files and drives the SDK. It runs every feature but +`leader_redirection` and `stream_crud`. Outside Docker, set `IGGY_TCP_ADDRESS`, +`IGGY_TCP_ADDRESS_NO_QUORUM`, `IGGY_ROOT_USERNAME`, and `IGGY_ROOT_PASSWORD`, then run +`swift run iggy-bdd`, with `BDD_FEATURE` to pick one feature. + ### Adding New Scenarios Add new `.feature` files to the `bdd/scenarios/` directory and implement the corresponding step definitions in each SDK's test directory. diff --git a/bdd/docker-compose.yml b/bdd/docker-compose.yml index c84273a6c6..74d948fdbf 100644 --- a/bdd/docker-compose.yml +++ b/bdd/docker-compose.yml @@ -234,37 +234,34 @@ services: networks: - iggy-bdd-network - cpp-bdd: - image: iggy-bdd-cpp + swift-bdd: + image: iggy-bdd-swift build: context: .. - dockerfile: bdd/cpp/Dockerfile + dockerfile: bdd/swift/Dockerfile 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 - - ./scenarios/stream_options.feature:/app/features/stream_options.feature - - ./scenarios/resource_options_patch.feature:/app/features/resource_options_patch.feature - - ./scenarios/metadata_changes.feature:/app/features/metadata_changes.feature - - ./scenarios/session_contract.feature:/app/features/session_contract.feature - - ./scenarios/deferred_polling.feature:/app/features/deferred_polling.feature networks: - iggy-bdd-network - # Runs only the deferred polling feature. - swift-bdd: - image: iggy-bdd-swift + cpp-bdd: + image: iggy-bdd-cpp build: context: .. - dockerfile: bdd/swift/Dockerfile + dockerfile: bdd/cpp/Dockerfile 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 + - ./scenarios/stream_options.feature:/app/features/stream_options.feature + - ./scenarios/resource_options_patch.feature:/app/features/resource_options_patch.feature + - ./scenarios/metadata_changes.feature:/app/features/metadata_changes.feature + - ./scenarios/session_contract.feature:/app/features/session_contract.feature - ./scenarios/deferred_polling.feature:/app/features/deferred_polling.feature networks: - iggy-bdd-network diff --git a/bdd/swift/Dockerfile b/bdd/swift/Dockerfile index 931c3b8105..b00a91ab02 100644 --- a/bdd/swift/Dockerfile +++ b/bdd/swift/Dockerfile @@ -15,16 +15,21 @@ # specific language governing permissions and limitations # under the License. -# Swift BDD runner: builds the Swift SDK and the runner, which parses the -# mounted feature files and runs their scenarios on the SDK. The toolchain -# matches .github/actions/utils/setup-swift. -FROM swift:6.4.0 +# The Swift version and Ubuntu release the Linux CI lanes install. +FROM swift:6.4.0-resolute WORKDIR /app + +# The SDK is a path dependency of the runner, so it keeps its place +# relative to the runner as in the repository. COPY foreign/swift ./foreign/swift -COPY bdd/swift ./bdd/swift -COPY bdd/scenarios/deferred_polling.feature ./features/ +COPY bdd/swift/tests/Package.swift bdd/swift/tests/Package.resolved ./bdd/swift/tests/ + +WORKDIR /app/bdd/swift/tests +RUN swift package resolve -RUN cd bdd/swift/runner && swift build --product iggy-bdd +COPY bdd/swift/tests/Sources ./Sources +COPY bdd/scenarios /app/bdd/scenarios +RUN swift build -Xswiftc -warnings-as-errors -CMD ["/app/bdd/swift/runner/.build/debug/iggy-bdd", "/app/features"] +CMD ["swift", "run", "--skip-build", "iggy-bdd"] diff --git a/bdd/swift/runner/Sources/Gherkin.swift b/bdd/swift/runner/Sources/Gherkin.swift deleted file mode 100644 index 180bea0ec4..0000000000 --- a/bdd/swift/runner/Sources/Gherkin.swift +++ /dev/null @@ -1,109 +0,0 @@ -// 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 Foundation - -/// One step of a scenario, without its keyword. -struct Step: Sendable { - let keyword: String - let text: String - let line: Int -} - -struct Scenario: Sendable { - let name: String - let line: Int - var steps: [Step] = [] -} - -struct Feature: Sendable { - let path: String - var name = "" - var background: [Step] = [] - var scenarios: [Scenario] = [] -} - -struct GherkinError: Error, CustomStringConvertible { - let description: String -} - -/// Reads the subset of Gherkin the shared features use: tags, comments, a -/// feature with its description, one background and scenarios of plain -/// steps. Anything else, such as tables, doc strings or outlines, is refused, -/// so a feature never runs with a step silently dropped. -enum Gherkin { - private static let stepKeywords = ["Given", "When", "Then", "And", "But"] - - static func parse(_ text: String, path: String) throws -> Feature { - var feature = Feature(path: path) - var inBackground = false - var current: Scenario? - for (index, raw) in text.components(separatedBy: .newlines).enumerated() { - let line = raw.trimmingCharacters(in: .whitespaces) - let number = index + 1 - if line.isEmpty || line.hasPrefix("#") || line.hasPrefix("@") { - continue - } - if let name = value(of: "Feature:", in: line) { - feature.name = name - } else if value(of: "Background:", in: line) != nil { - inBackground = true - } else if let name = value(of: "Scenario:", in: line) { - if let current { - feature.scenarios.append(current) - } - inBackground = false - current = Scenario(name: name, line: number) - } else if let step = step(line, number: number) { - if inBackground { - feature.background.append(step) - } else if current != nil { - current?.steps.append(step) - } else { - throw GherkinError(description: "\(path):\(number): a step outside a scenario") - } - } else if line.hasPrefix("|") || line.hasPrefix("\"\"\"") || line.hasPrefix("Scenario Outline:") || line.hasPrefix("Examples:") - || line.hasPrefix("Rule:") - { - throw GherkinError(description: "\(path):\(number): unsupported Gherkin: \(line)") - } else if current != nil || inBackground { - throw GherkinError(description: "\(path):\(number): not a step: \(line)") - } - } - if let current { - feature.scenarios.append(current) - } - guard !feature.name.isEmpty else { - throw GherkinError(description: "\(path): no feature") - } - return feature - } - - private static func value(of keyword: String, in line: String) -> String? { - guard line.hasPrefix(keyword) else { - return nil - } - return line.dropFirst(keyword.count).trimmingCharacters(in: .whitespaces) - } - - private static func step(_ line: String, number: Int) -> Step? { - for keyword in stepKeywords where line.hasPrefix(keyword + " ") { - return Step(keyword: keyword, text: String(line.dropFirst(keyword.count + 1)), line: number) - } - return nil - } -} diff --git a/bdd/swift/runner/Sources/Steps.swift b/bdd/swift/runner/Sources/Steps.swift deleted file mode 100644 index 8eae7731b5..0000000000 --- a/bdd/swift/runner/Sources/Steps.swift +++ /dev/null @@ -1,356 +0,0 @@ -// 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 Foundation -import Iggy - -/// Whether a poll ended, set by the task of the poll. -final class PollEnd: @unchecked Sendable { - private let lock = NSLock() - private var endedAt: ContinuousClock.Instant? - - func end() { - lock.withLock { endedAt = endedAt ?? .now } - } - - var ended: ContinuousClock.Instant? { - lock.withLock { endedAt } - } -} - -struct StepFailure: Error, CustomStringConvertible { - let description: String - - init(_ description: String) { - self.description = description - } -} - -/// The state of one scenario. Its steps run one after another, never at the -/// same time, so the state needs no lock. -final class World: @unchecked Sendable { - var client: IggyClient? - var stream: Identifier? - var topic: Identifier? - var poll: Task? - var pollStarted = ContinuousClock.now - var pollEnd = PollEnd() - var reader: PartitionReader? - var consumer: IggyConsumer? - /// The operation of a committing poll whose outcome is unknown. - var lostOperation: UInt128Value? - - func requireClient() throws -> IggyClient { - guard let client else { - throw StepFailure("no client is signed in") - } - return client - } - - func requireTopic() throws -> (stream: Identifier, topic: Identifier) { - guard let stream, let topic else { - throw StepFailure("no stream and topic were created") - } - return (stream, topic) - } - - func requirePoll() throws -> Task { - guard let poll else { - throw StepFailure("no deferred poll was started") - } - return poll - } - - func requireConsumer() throws -> IggyConsumer { - guard let consumer else { - throw StepFailure("no consumer is open") - } - return consumer - } - - func tearDown() async { - poll?.cancel() - if let reader { - try? await reader.close() - } - await consumer?.shutdown() - await client?.shutdown() - } -} - -/// How long a started poll gets to reach the server and park there. -private let parkTime: Duration = .milliseconds(500) - -struct StepDefinition: Sendable { - let pattern: String - let run: @Sendable (World, [String]) async throws -> Void -} - -/// Runs `body` and fails when it takes longer than `limit`. -func within(_ limit: Duration, _ what: String, _ body: @escaping @Sendable () async throws -> T) async throws -> T { - try await withThrowingTaskGroup(of: T?.self) { group in - group.addTask { try await body() } - group.addTask { - try await Task.sleep(for: limit) - return nil - } - defer { group.cancelAll() } - guard let value = try await group.next() ?? nil else { - throw StepFailure("\(what) took longer than \(limit)") - } - return value - } -} - -private func expect(_ condition: Bool, _ message: @autoclosure () -> String) throws { - if !condition { - throw StepFailure(message()) - } -} - -private func seconds(_ text: String) throws -> Duration { - guard let value = Int64(text) else { - throw StepFailure("\(text) is not a number of seconds") - } - return .seconds(value) -} - -private func number(_ text: String) throws -> T { - guard let value = T(text) else { - throw StepFailure("\(text) is not a number") - } - return value -} - -private func send(_ world: World, payload: String, to partition: UInt32) async throws { - let (stream, topic) = try world.requireTopic() - var messages = [try IggyMessage(payload)] - try await world.requireClient().sendMessages(streamID: stream, topicID: topic, partitioning: .partition(partition), messages: &messages) -} - -private func poll(_ world: World, partition: UInt32, consumer: String, commit: CommitPolicy) async throws -> PolledMessages { - let (stream, topic) = try world.requireTopic() - return try await world.requireClient().pollMessages( - streamID: stream, topicID: topic, partitionID: partition, consumer: .consumer(try Identifier(named: consumer)), strategy: .next, - options: PollOptions(count: 1, commit: commit)) -} - -/// The ``IggyError`` that `body` throws, or a failure when it throws -/// anything else or returns, which names what it returned. -private func iggyError(_ what: String, _ body: () async throws -> PolledMessages) async throws -> IggyError { - let polled: PolledMessages - do { - polled = try await body() - } catch let error as IggyError { - return error - } catch { - throw StepFailure("\(what) failed with \(error)") - } - throw StepFailure("\(what) did not fail, it returned \(polled.messages.map(\.payloadString)) from partition \(polled.partitionID)") -} - -private func openConsumer(_ world: World, group: String, options: PollOptions) async throws { - let (stream, topic) = try world.requireTopic() - let consumer = try world.requireClient().consumerGroup( - name: group, streamID: stream, topicID: topic, configuration: ConsumerConfiguration(pollOptions: options)) - try await consumer.initialize() - world.consumer = consumer -} - -/// The steps of `bdd/scenarios/deferred_polling.feature` on the Swift SDK. -let stepDefinitions: [StepDefinition] = [ - StepDefinition(pattern: #"I have a running Iggy server"#) { world, _ in - let address = ProcessInfo.processInfo.environment["IGGY_TCP_ADDRESS"] ?? "127.0.0.1:8090" - let client = IggyClient(configuration: ClientConfiguration(address: address)) - try await client.connect() - world.client = client - }, - StepDefinition(pattern: #"I am authenticated as the root user"#) { world, _ in - let environment = ProcessInfo.processInfo.environment - _ = try await world.requireClient().login( - username: environment["IGGY_ROOT_USERNAME"] ?? "iggy", password: environment["IGGY_ROOT_PASSWORD"] ?? "iggy") - }, - StepDefinition(pattern: #"a stream named "([^"]+)" with a topic named "([^"]+)" with (\d+) partitions exists"#) { world, values in - let client = try world.requireClient() - _ = try await client.createStream(name: values[0]) - let stream = try Identifier(named: values[0]) - _ = try await client.createTopic(streamID: stream, name: values[1], options: TopicCreateOptions(partitionsCount: try number(values[2]))) - world.stream = stream - world.topic = try Identifier(named: values[1]) - }, - StepDefinition(pattern: #"I send a message with payload "([^"]+)" to partition (\d+)"#) { world, values in - try await send(world, payload: values[0], to: try number(values[1])) - }, - StepDefinition(pattern: #"I start a deferred poll of partition (\d+) from offset (\d+) with a max wait of (\d+) seconds?"#) { world, values in - let (stream, topic) = try world.requireTopic() - let client = try world.requireClient() - let partition: UInt32 = try number(values[0]) - let offset: UInt64 = try number(values[1]) - let options = PollOptions(count: 10, maxWait: try seconds(values[2])) - world.pollStarted = .now - world.poll = Task { - try await client.pollMessages(streamID: stream, topicID: topic, partitionID: partition, strategy: .offset(offset), options: options) - } - }, - StepDefinition( - pattern: - #"I start a deferred poll of partition (\d+) for consumer "([^"]+)" with the on read commit policy, a min count of (\d+) and a max wait of (\d+) seconds?"# - ) { world, values in - let (stream, topic) = try world.requireTopic() - let client = try world.requireClient() - let partition: UInt32 = try number(values[0]) - let consumer = Consumer.consumer(try Identifier(named: values[1])) - let options = PollOptions(count: 10, maxWait: try seconds(values[3]), minCount: try number(values[2]), commit: .onRead) - world.pollStarted = .now - world.poll = Task { - try await client.pollMessages(streamID: stream, topicID: topic, partitionID: partition, consumer: consumer, strategy: .next, options: options) - } - }, - StepDefinition( - pattern: #"I start a deferred poll of partition (\d+) for consumer "([^"]+)" with the on read commit policy and a max wait of (\d+) seconds?"# - ) { world, values in - let (stream, topic) = try world.requireTopic() - let client = try world.requireClient() - let partition: UInt32 = try number(values[0]) - let consumer = Consumer.consumer(try Identifier(named: values[1])) - let options = PollOptions(count: 10, maxWait: try seconds(values[2]), commit: .onRead) - // A read that waits for nothing first opens the route and the - // connection, so the parked poll only writes its frame. - _ = try await client.pollMessages( - streamID: stream, topicID: topic, partitionID: partition, consumer: consumer, strategy: .offset(0), options: PollOptions(count: 1, maxWait: .zero)) - let end = PollEnd() - world.pollEnd = end - world.pollStarted = .now - world.poll = Task { - defer { end.end() } - return try await client.pollMessages( - streamID: stream, topicID: topic, partitionID: partition, consumer: consumer, strategy: .next, options: options) - } - }, - StepDefinition(pattern: #"the client disconnects before the deferred poll answers"#) { world, _ in - _ = try world.requirePoll() - try await Task.sleep(for: parkTime) - if let ended = world.pollEnd.ended { - throw StepFailure("the deferred poll ended \(ended - world.pollStarted) after it started, before the disconnect") - } - await (try world.requireClient()).disconnect() - }, - StepDefinition(pattern: #"the deferred poll should fail as uncertain"#) { world, _ in - let poll = try world.requirePoll() - world.poll = nil - let error = try await iggyError("the deferred poll") { try await within(.seconds(5), "the deferred poll") { try await poll.value } } - let ended = world.pollEnd.ended.map { "\($0 - world.pollStarted)" } ?? "never" - try expect(error.code == .fetchProgressUncertain, "the deferred poll failed with \(error) \(ended) after it started") - guard let operation = error.operationID else { - throw StepFailure("the uncertain poll names no operation: \(error)") - } - world.lostOperation = operation - }, - StepDefinition(pattern: #"the client connects again"#) { world, _ in - let client = try world.requireClient() - try await client.connect() - let environment = ProcessInfo.processInfo.environment - _ = try await client.login(username: environment["IGGY_ROOT_USERNAME"] ?? "iggy", password: environment["IGGY_ROOT_PASSWORD"] ?? "iggy") - }, - StepDefinition( - pattern: - #"a poll of partition (\d+) for consumer "([^"]+)" with the next strategy and the on read commit policy should fail as blocked by the deferred poll"# - ) { world, values in - let partition: UInt32 = try number(values[0]) - let error = try await iggyError("the poll") { try await poll(world, partition: partition, consumer: values[1], commit: .onRead) } - try expect(error.code == .partitionLaneBlocked, "the poll failed with \(error)") - try expect(error.operationID == world.lostOperation && error.partitionID == partition, "the poll was blocked by \(error)") - }, - StepDefinition(pattern: #"I abandon the operation of the deferred poll"#) { world, _ in - guard let operation = world.lostOperation else { - throw StepFailure("no deferred poll failed as uncertain") - } - try world.requireClient().abandonOperation(operation) - }, - StepDefinition(pattern: #"I cancel the deferred poll after (\d+) milliseconds"#) { world, values in - let poll = try world.requirePoll() - try await Task.sleep(for: .milliseconds(try number(values[0]) as Int64)) - poll.cancel() - // The server answers a cancelled committing poll once, with nothing - // stored when the cancellation won. - let result = await poll.result - guard case .failure(let error) = result, error is CancellationError else { - throw StepFailure("the cancelled poll ended with \(result)") - } - }, - StepDefinition(pattern: #"I wait (\d+) seconds?"#) { _, values in - try await Task.sleep(for: try seconds(values[0])) - }, - StepDefinition(pattern: #"the deferred poll should return the payload "([^"]+)" within (\d+) seconds?"#) { world, values in - let poll = try world.requirePoll() - let polled = try await within(try seconds(values[1]), "the deferred poll") { try await poll.value } - try expect(polled.messages.map(\.payloadString) == [values[0]], "the deferred poll returned \(polled.messages.map(\.payloadString))") - }, - StepDefinition(pattern: #"the deferred poll should return no messages between (\d+) and (\d+) seconds after it started"#) { world, values in - let polled = try await world.requirePoll().value - let elapsed = ContinuousClock.now - world.pollStarted - try expect(polled.messages.isEmpty, "the deferred poll returned \(polled.messages.map(\.payloadString))") - try expect( - elapsed >= (try seconds(values[0])) && elapsed <= (try seconds(values[1])), "the deferred poll returned after \(elapsed)") - }, - StepDefinition( - pattern: #"a poll of partition (\d+) for consumer "([^"]+)" with the next strategy and the on read commit policy should return the payload "([^"]+)""# - ) { world, values in - let polled = try await poll(world, partition: try number(values[0]), consumer: values[1], commit: .onRead) - try expect(polled.messages.map(\.payloadString) == [values[2]], "the poll returned \(polled.messages.map(\.payloadString))") - }, - StepDefinition(pattern: #"a poll of partition (\d+) for consumer "([^"]+)" with the next strategy should return the payload "([^"]+)""#) { - world, values in - let polled = try await poll(world, partition: try number(values[0]), consumer: values[1], commit: .manual) - try expect(polled.messages.map(\.payloadString) == [values[2]], "the poll returned \(polled.messages.map(\.payloadString))") - }, - StepDefinition(pattern: #"I open a partition reader on partitions ([\d, and]+) from offset (\d+) with a max wait of (\d+) seconds?"#) { world, values in - let (stream, topic) = try world.requireTopic() - let partitions = values[0].split(whereSeparator: { !$0.isNumber }).compactMap { UInt32($0) } - let offset: UInt64 = try number(values[1]) - world.reader = try await world.requireClient().reader( - streamID: stream, topicID: topic, consumer: .consumer(try Identifier(named: "bdd-reader")), - configuration: ReaderConfiguration(partitions: partitions, start: .offset(offset), options: .wait(try seconds(values[2])))) - }, - StepDefinition(pattern: #"the partition reader should return the payload "([^"]+)" from partition (\d+) within (\d+) seconds?"#) { - world, values in - guard let reader = world.reader else { - throw StepFailure("no partition reader is open") - } - let batch = try await within(try seconds(values[2]), "the partition reader") { try await reader.next() } - let partition: UInt32 = try number(values[1]) - try expect(batch?.partitionID == partition, "the reader returned partition \(String(describing: batch?.partitionID))") - try expect(batch?.messages.map(\.payloadString) == [values[0]], "the reader returned \(String(describing: batch?.messages.map(\.payloadString)))") - }, - StepDefinition(pattern: #"I open a consumer in group "([^"]+)" on the topic with a max wait of (\d+) seconds?"#) { world, values in - try await openConsumer(world, group: values[0], options: .wait(try seconds(values[1]))) - }, - StepDefinition(pattern: #"I open a consumer in group "([^"]+)" on the topic that commits after processing"#) { world, values in - var options = PollOptions.wait(ReaderConfiguration.defaultMaxWait) - options.commit = .auto(AutoCommit(interval: .seconds(5), after: .processing)) - try await openConsumer(world, group: values[0], options: options) - }, - StepDefinition(pattern: #"the consumer should return the payload "([^"]+)" within (\d+) seconds?"#) { world, values in - let consumer = try world.requireConsumer() - let received = try await within(try seconds(values[1]), "the consumer") { try await consumer.next() } - try expect(received?.message.payloadString == values[0], "the consumer returned \(String(describing: received?.message.payloadString))") - }, - StepDefinition(pattern: #"I close the consumer"#) { world, _ in - await (try world.requireConsumer()).shutdown() - world.consumer = nil - }, -] diff --git a/bdd/swift/runner/Sources/main.swift b/bdd/swift/runner/Sources/main.swift deleted file mode 100644 index db3174288f..0000000000 --- a/bdd/swift/runner/Sources/main.swift +++ /dev/null @@ -1,92 +0,0 @@ -// 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 Foundation - -// Runs every scenario of the feature files in the directory of the first -// argument, `/app/features` by default, or only `.feature` when -// BDD_FEATURE names one. A step without a definition fails its scenario. -// Exits with 1 when any scenario failed. - -/// The definition whose pattern matches all of `text`, with its captures. -func match(_ text: String) throws -> (StepDefinition, [String])? { - let range = NSRange(text.startIndex..., in: text) - for definition in stepDefinitions { - let expression = try NSRegularExpression(pattern: "^" + definition.pattern + "$") - guard let result = expression.firstMatch(in: text, range: range) else { - continue - } - let captures = (1.. Bool { - let world = World() - defer { print() } - print(" Scenario: \(scenario.name)") - for step in feature.background + scenario.steps { - do { - guard let (definition, captures) = try match(step.text) else { - print(" ✘ \(step.keyword) \(step.text)\n undefined step at \(feature.path):\(step.line)") - await world.tearDown() - return false - } - try await definition.run(world, captures) - print(" ✔ \(step.keyword) \(step.text)") - } catch { - print(" ✘ \(step.keyword) \(step.text)\n \(error)") - await world.tearDown() - return false - } - } - await world.tearDown() - return true -} - -let directory = CommandLine.arguments.dropFirst().first ?? "/app/features" -let selected = ProcessInfo.processInfo.environment["BDD_FEATURE"].flatMap { $0 == "all" || $0.isEmpty ? nil : $0 } -let paths = try FileManager.default.contentsOfDirectory(atPath: directory) - .filter { $0.hasSuffix(".feature") && (selected == nil || $0 == "\(selected ?? "").feature") } - .sorted() - .map { "\(directory)/\($0)" } -guard !paths.isEmpty else { - print("No feature file to run in \(directory)") - exit(1) -} - -var passed = 0 -var failed: [String] = [] -for path in paths { - let feature = try Gherkin.parse(try String(contentsOfFile: path, encoding: .utf8), path: path) - print("Feature: \(feature.name)\n") - for scenario in feature.scenarios { - if await run(scenario, of: feature) { - passed += 1 - } else { - failed.append("\(path):\(scenario.line) \(scenario.name)") - } - } -} -print("\(passed + failed.count) scenarios: \(passed) passed, \(failed.count) failed") -for scenario in failed { - print(" failed: \(scenario)") -} -exit(failed.isEmpty ? 0 : 1) diff --git a/bdd/swift/runner/Package.resolved b/bdd/swift/tests/Package.resolved similarity index 79% rename from bdd/swift/runner/Package.resolved rename to bdd/swift/tests/Package.resolved index 85cf3ef6fc..12cd9c4cc7 100644 --- a/bdd/swift/runner/Package.resolved +++ b/bdd/swift/tests/Package.resolved @@ -1,5 +1,5 @@ { - "originHash" : "2eef30a4b9e0abd3654ca61c734ec545f17b5858f0876dbec3b6318dec8d9bb9", + "originHash" : "5779d5213e3a88ea6669d73995c16d99f4400dc8c2a7362223e4d44682e2dafb", "pins" : [ { "identity" : "swift-asn1", @@ -24,8 +24,8 @@ "kind" : "remoteSourceControl", "location" : "https://github.com/apple/swift-collections.git", "state" : { - "revision" : "f96b909e1a9e469620bc5037c1e24e00153f376f", - "version" : "1.7.5" + "revision" : "a0cb0954ecb21e4e31b0070e6ed5674e8556685a", + "version" : "1.6.0" } }, { @@ -42,8 +42,8 @@ "kind" : "remoteSourceControl", "location" : "https://github.com/apple/swift-log.git", "state" : { - "revision" : "c9c6bb9d275ec0d95c3ad16391f3f045dc623104", - "version" : "1.16.2" + "revision" : "9c6fb14227f55d8f711ce3847dc2f419fb0ecacb", + "version" : "1.15.1" } }, { @@ -51,8 +51,8 @@ "kind" : "remoteSourceControl", "location" : "https://github.com/apple/swift-nio.git", "state" : { - "revision" : "6db9d9abd92ad7399b8a564cafbb87d3507245fa", - "version" : "2.104.0" + "revision" : "a931f2c1de8dd49381ce3bf2e279d033f68d8865", + "version" : "2.102.0" } }, { @@ -60,8 +60,8 @@ "kind" : "remoteSourceControl", "location" : "https://github.com/apple/swift-nio-ssl.git", "state" : { - "revision" : "322f3c2a4a21df31c84ca416bf65ee5e9059e440", - "version" : "2.37.5" + "revision" : "03827c1a9fdb2b6b00a4e93ede8861520263af8c", + "version" : "2.37.4" } }, { diff --git a/bdd/swift/runner/Package.swift b/bdd/swift/tests/Package.swift similarity index 75% rename from bdd/swift/runner/Package.swift rename to bdd/swift/tests/Package.swift index e868b791ff..1757bdd2f9 100644 --- a/bdd/swift/runner/Package.swift +++ b/bdd/swift/tests/Package.swift @@ -18,17 +18,19 @@ import PackageDescription +// A small Gherkin interpreter that walks the shared scenarios in +// bdd/scenarios and drives the SDK against a live server, as the other SDK +// suites do with cucumber or godog. It depends on the SDK by path. let package = Package( name: "iggy-bdd", platforms: [.macOS(.v13)], dependencies: [ - .package(path: "../../../foreign/swift") + .package(name: "apache-iggy", path: "../../../foreign/swift") ], targets: [ .executableTarget( name: "iggy-bdd", - dependencies: [.product(name: "Iggy", package: "swift")], - path: "Sources" + dependencies: [.product(name: "Iggy", package: "apache-iggy")] ) ], swiftLanguageModes: [.v6] diff --git a/bdd/swift/tests/Sources/iggy-bdd/DeferredPollingSteps.swift b/bdd/swift/tests/Sources/iggy-bdd/DeferredPollingSteps.swift new file mode 100644 index 0000000000..0d6b7f9972 --- /dev/null +++ b/bdd/swift/tests/Sources/iggy-bdd/DeferredPollingSteps.swift @@ -0,0 +1,265 @@ +// 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 Foundation +import Iggy + +/// Whether a poll ended, set by the task of the poll. +final class PollEnd: @unchecked Sendable { + private let lock = NSLock() + private var endedAt: ContinuousClock.Instant? + + func end() { + lock.withLock { endedAt = endedAt ?? .now } + } + + var ended: ContinuousClock.Instant? { + lock.withLock { endedAt } + } +} + +/// `deferred_polling.feature`. +enum DeferredPollingSteps { + /// How long a started poll gets to reach the server and park there. + static let parkTime: Duration = .milliseconds(500) + + static func register(into registry: inout StepRegistry) { + registry.step(#"a stream named "([^"]+)" with a topic named "([^"]+)" with (\d+) partitions exists"#) { world, values in + let client = try world.requireClient() + let stream = try await client.createStream(name: values[0]) + world.lastStreamID = stream.id + world.createdStreams.append(stream.id) + let topic = try await client.createTopic( + streamID: Identifier(numeric: stream.id), name: values[1], options: TopicCreateOptions(partitionsCount: try number(values[2]))) + world.lastTopicID = topic.id + } + registry.step(#"I send a message with payload "([^"]+)" to partition (\d+)"#) { world, values in + try await send(world, payload: values[0], to: try number(values[1])) + } + registry.step(#"I start a deferred poll of partition (\d+) from offset (\d+) with a max wait of (\d+) seconds?"#) { world, values in + let (stream, topic) = (try world.requireStream(), try world.requireTopic()) + let client = try world.requireClient() + let partition: UInt32 = try number(values[0]) + let offset: UInt64 = try number(values[1]) + let options = PollOptions(count: 10, maxWait: try seconds(values[2])) + world.pollStarted = .now + world.poll = Task { + try await client.pollMessages(streamID: stream, topicID: topic, partitionID: partition, strategy: .offset(offset), options: options) + } + } + registry.step( + #"I start a deferred poll of partition (\d+) for consumer "([^"]+)" with the on read commit policy, a min count of (\d+) and a max wait of (\d+) seconds?"# + ) { world, values in + let (stream, topic) = (try world.requireStream(), try world.requireTopic()) + let client = try world.requireClient() + let partition: UInt32 = try number(values[0]) + let consumer = Consumer.consumer(try Identifier(named: values[1])) + let options = PollOptions(count: 10, maxWait: try seconds(values[3]), minCount: try number(values[2]), commit: .onRead) + world.pollStarted = .now + world.poll = Task { + try await client.pollMessages(streamID: stream, topicID: topic, partitionID: partition, consumer: consumer, strategy: .next, options: options) + } + } + registry.step( + #"I start a deferred poll of partition (\d+) for consumer "([^"]+)" with the on read commit policy and a max wait of (\d+) seconds?"# + ) { world, values in + let (stream, topic) = (try world.requireStream(), try world.requireTopic()) + let client = try world.requireClient() + let partition: UInt32 = try number(values[0]) + let consumer = Consumer.consumer(try Identifier(named: values[1])) + let options = PollOptions(count: 10, maxWait: try seconds(values[2]), commit: .onRead) + // A read that waits for nothing first opens the route and the + // connection, so the parked poll only writes its frame. + _ = try await client.pollMessages( + streamID: stream, topicID: topic, partitionID: partition, consumer: consumer, strategy: .offset(0), + options: PollOptions(count: 1, maxWait: .zero)) + let end = PollEnd() + world.pollEnd = end + world.pollStarted = .now + world.poll = Task { + defer { end.end() } + return try await client.pollMessages( + streamID: stream, topicID: topic, partitionID: partition, consumer: consumer, strategy: .next, options: options) + } + } + registry.step(#"the client disconnects before the deferred poll answers"#) { world, _ in + _ = try world.requirePoll() + try await Task.sleep(for: parkTime) + if let ended = world.pollEnd.ended { + throw StepFailure("the deferred poll ended \(ended - world.pollStarted) after it started, before the disconnect") + } + await (try world.requireClient()).disconnect() + } + registry.step(#"the deferred poll should fail as uncertain"#) { world, _ in + let poll = try world.requirePoll() + world.poll = nil + let error = try await iggyError("the deferred poll") { try await within(.seconds(5), "the deferred poll") { try await poll.value } } + let ended = world.pollEnd.ended.map { "\($0 - world.pollStarted)" } ?? "never" + try check(error.code == .fetchProgressUncertain, "the deferred poll failed with \(error) \(ended) after it started") + guard let operation = error.operationID else { + throw StepFailure("the uncertain poll names no operation: \(error)") + } + world.lostOperation = operation + } + registry.step(#"the client connects again"#) { world, _ in + let client = try world.requireClient() + try await client.connect() + let root = try Environment.rootCredentials() + _ = try await client.login(username: root.username, password: root.password) + } + registry.step( + #"a poll of partition (\d+) for consumer "([^"]+)" with the next strategy and the on read commit policy should fail as blocked by the deferred poll"# + ) { world, values in + let partition: UInt32 = try number(values[0]) + let error = try await iggyError("the poll") { try await poll(world, partition: partition, consumer: values[1], commit: .onRead) } + try check(error.code == .partitionLaneBlocked, "the poll failed with \(error)") + try check(error.operationID == world.lostOperation && error.partitionID == partition, "the poll was blocked by \(error)") + } + registry.step(#"I abandon the operation of the deferred poll"#) { world, _ in + guard let operation = world.lostOperation else { + throw StepFailure("no deferred poll failed as uncertain") + } + try world.requireClient().abandonOperation(operation) + } + registry.step(#"I cancel the deferred poll after (\d+) milliseconds"#) { world, values in + let poll = try world.requirePoll() + try await Task.sleep(for: .milliseconds(try number(values[0], as: Int64.self))) + poll.cancel() + // The server answers a cancelled committing poll once, with nothing + // stored when the cancellation won. + let result = await poll.result + guard case .failure(let error) = result, error is CancellationError else { + throw StepFailure("the cancelled poll ended with \(result)") + } + } + registry.step(#"I wait (\d+) seconds?"#) { _, values in + try await Task.sleep(for: try seconds(values[0])) + } + registry.step(#"the deferred poll should return the payload "([^"]+)" within (\d+) seconds?"#) { world, values in + let poll = try world.requirePoll() + let polled = try await within(try seconds(values[1]), "the deferred poll") { try await poll.value } + try check(polled.messages.map(\.payloadString) == [values[0]], "the deferred poll returned \(polled.messages.map(\.payloadString))") + } + registry.step(#"the deferred poll should return no messages between (\d+) and (\d+) seconds after it started"#) { world, values in + let polled = try await world.requirePoll().value + let elapsed = ContinuousClock.now - world.pollStarted + try check(polled.messages.isEmpty, "the deferred poll returned \(polled.messages.map(\.payloadString))") + try check(elapsed >= (try seconds(values[0])) && elapsed <= (try seconds(values[1])), "the deferred poll returned after \(elapsed)") + } + registry.step( + #"a poll of partition (\d+) for consumer "([^"]+)" with the next strategy and the on read commit policy should return the payload "([^"]+)""# + ) { world, values in + let polled = try await poll(world, partition: try number(values[0]), consumer: values[1], commit: .onRead) + try check(polled.messages.map(\.payloadString) == [values[2]], "the poll returned \(polled.messages.map(\.payloadString))") + } + registry.step(#"a poll of partition (\d+) for consumer "([^"]+)" with the next strategy should return the payload "([^"]+)""#) { world, values in + let polled = try await poll(world, partition: try number(values[0]), consumer: values[1], commit: .manual) + try check(polled.messages.map(\.payloadString) == [values[2]], "the poll returned \(polled.messages.map(\.payloadString))") + } + registry.step(#"I open a partition reader on partitions ([\d, and]+) from offset (\d+) with a max wait of (\d+) seconds?"#) { world, values in + let (stream, topic) = (try world.requireStream(), try world.requireTopic()) + let partitions = values[0].split(whereSeparator: { !$0.isNumber }).compactMap { UInt32($0) } + let offset: UInt64 = try number(values[1]) + world.reader = try await world.requireClient().reader( + streamID: stream, topicID: topic, consumer: .consumer(try Identifier(named: "bdd-reader")), + configuration: ReaderConfiguration(partitions: partitions, start: .offset(offset), options: .wait(try seconds(values[2])))) + } + registry.step(#"the partition reader should return the payload "([^"]+)" from partition (\d+) within (\d+) seconds?"#) { world, values in + guard let reader = world.reader else { + throw StepFailure("no partition reader is open") + } + let batch = try await within(try seconds(values[2]), "the partition reader") { try await reader.next() } + let partition: UInt32 = try number(values[1]) + try check(batch?.partitionID == partition, "the reader returned partition \(String(describing: batch?.partitionID))") + try check( + batch?.messages.map(\.payloadString) == [values[0]], "the reader returned \(String(describing: batch?.messages.map(\.payloadString)))") + } + registry.step(#"I open a consumer in group "([^"]+)" on the topic with a max wait of (\d+) seconds?"#) { world, values in + try await openConsumer(world, group: values[0], options: .wait(try seconds(values[1]))) + } + registry.step(#"I open a consumer in group "([^"]+)" on the topic that commits after processing"#) { world, values in + var options = PollOptions.wait(ReaderConfiguration.defaultMaxWait) + options.commit = .auto(AutoCommit(interval: .seconds(5), after: .processing)) + try await openConsumer(world, group: values[0], options: options) + } + registry.step(#"the consumer should return the payload "([^"]+)" within (\d+) seconds?"#) { world, values in + let consumer = try world.requireConsumer() + let received = try await within(try seconds(values[1]), "the consumer") { try await consumer.next() } + try check(received?.message.payloadString == values[0], "the consumer returned \(String(describing: received?.message.payloadString))") + } + registry.step(#"I close the consumer"#) { world, _ in + await (try world.requireConsumer()).shutdown() + world.consumer = nil + } + } + + /// Runs `body` and fails when it takes longer than `limit`. + private static func within( + _ limit: Duration, _ what: String, _ body: @escaping @Sendable () async throws -> T + ) async throws -> T { + try await withThrowingTaskGroup(of: T?.self) { group in + group.addTask { try await body() } + group.addTask { + try await Task.sleep(for: limit) + return nil + } + defer { group.cancelAll() } + guard let value = try await group.next() ?? nil else { + throw StepFailure("\(what) took longer than \(limit)") + } + return value + } + } + + private static func seconds(_ text: String) throws -> Duration { + .seconds(try number(text, as: Int64.self)) + } + + private static func send(_ world: World, payload: String, to partition: UInt32) async throws { + var messages = [try IggyMessage(payload)] + try await world.requireClient().sendMessages( + streamID: try world.requireStream(), topicID: try world.requireTopic(), partitioning: .partition(partition), messages: &messages) + } + + private static func poll(_ world: World, partition: UInt32, consumer: String, commit: CommitPolicy) async throws -> PolledMessages { + try await world.requireClient().pollMessages( + streamID: try world.requireStream(), topicID: try world.requireTopic(), partitionID: partition, + consumer: .consumer(try Identifier(named: consumer)), strategy: .next, options: PollOptions(count: 1, commit: commit)) + } + + /// The ``IggyError`` that `body` throws, or a failure when it throws + /// anything else or returns, which names what it returned. + private static func iggyError(_ what: String, _ body: () async throws -> PolledMessages) async throws -> IggyError { + let polled: PolledMessages + do { + polled = try await body() + } catch let error as IggyError { + return error + } catch { + throw StepFailure("\(what) failed with \(error)") + } + throw StepFailure("\(what) did not fail, it returned \(polled.messages.map(\.payloadString)) from partition \(polled.partitionID)") + } + + private static func openConsumer(_ world: World, group: String, options: PollOptions) async throws { + let consumer = try world.requireClient().consumerGroup( + name: group, streamID: try world.requireStream(), topicID: try world.requireTopic(), + configuration: ConsumerConfiguration(pollOptions: options)) + try await consumer.initialize() + world.consumer = consumer + } +} diff --git a/bdd/swift/tests/Sources/iggy-bdd/Gherkin.swift b/bdd/swift/tests/Sources/iggy-bdd/Gherkin.swift new file mode 100644 index 0000000000..c3ead78563 --- /dev/null +++ b/bdd/swift/tests/Sources/iggy-bdd/Gherkin.swift @@ -0,0 +1,102 @@ +// 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. + +/// One step of a scenario: the keyword as written and the text after it. +struct Step: Sendable { + let keyword: String + let text: String + let line: Int +} + +struct Scenario: Sendable { + let name: String + let steps: [Step] +} + +struct Feature: Sendable { + let name: String + let background: [Step] + let scenarios: [Scenario] +} + +/// Parses the Gherkin the shared scenarios use: a feature with an optional +/// background, scenarios, and Given/When/Then/And/But steps. Tags and the +/// prose under a heading are skipped. A construct it cannot run, such as a +/// scenario outline, fails the parse rather than running part of it. +enum GherkinParser { + static let stepKeywords = ["Given", "When", "Then", "And", "But"] + static let unsupportedKeywords = ["Scenario Outline:", "Scenario Template:", "Examples:", "Scenarios:", "Rule:", "\"\"\"", "|"] + + static func parse(_ text: String) throws -> Feature { + var featureName: String? + var background: [Step] = [] + var scenarios: [Scenario] = [] + var current: (name: String, steps: [Step])? + var inBackground = false + + for (index, rawLine) in text.split(separator: "\n", omittingEmptySubsequences: false).enumerated() { + let line = rawLine.trimmingCharacters(in: .whitespaces) + let number = index + 1 + if line.isEmpty || line.hasPrefix("#") || line.hasPrefix("@") { + continue + } + if let keyword = unsupportedKeywords.first(where: { line.hasPrefix($0) }) { + throw ParseError(line: number, message: "\(keyword) is not supported by this runner") + } + if let name = heading("Feature:", line) { + featureName = name + } else if heading("Background:", line) != nil { + inBackground = true + } else if let name = heading("Scenario:", line) ?? heading("Example:", line) { + if let current { + scenarios.append(Scenario(name: current.name, steps: current.steps)) + } + inBackground = false + current = (name, []) + } else if let keyword = stepKeywords.first(where: { line.hasPrefix($0 + " ") }) { + let step = Step(keyword: keyword, text: String(line.dropFirst(keyword.count + 1)).trimmingCharacters(in: .whitespaces), line: number) + if inBackground { + background.append(step) + } else if current != nil { + current?.steps.append(step) + } else { + throw ParseError(line: number, message: "step outside a scenario") + } + } + } + if let current { + scenarios.append(Scenario(name: current.name, steps: current.steps)) + } + guard let featureName else { + throw ParseError(line: 0, message: "no Feature: heading") + } + return Feature(name: featureName, background: background, scenarios: scenarios) + } + + private static func heading(_ keyword: String, _ line: String) -> String? { + guard line.hasPrefix(keyword) else { + return nil + } + return line.dropFirst(keyword.count).trimmingCharacters(in: .whitespaces) + } + + struct ParseError: Error, CustomStringConvertible { + let line: Int + let message: String + var description: String { "line \(line): \(message)" } + } +} diff --git a/bdd/swift/tests/Sources/iggy-bdd/MessagingSteps.swift b/bdd/swift/tests/Sources/iggy-bdd/MessagingSteps.swift new file mode 100644 index 0000000000..2d81e6f0d2 --- /dev/null +++ b/bdd/swift/tests/Sources/iggy-bdd/MessagingSteps.swift @@ -0,0 +1,216 @@ +// 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 Foundation +import Iggy + +/// The background every feature shares. +enum CommonSteps { + static func register(into registry: inout StepRegistry) { + registry.step("I have a running Iggy server") { world, _ in + world.serverAddress = try Environment.serverAddress() + } + registry.step("I am authenticated as the root user") { world, _ in + guard let address = world.serverAddress else { + throw StepFailure("the server address was not read; the background is out of order") + } + let client = try await world.connect(to: address) + try await client.ping() + let root = try Environment.rootCredentials() + _ = try await client.login(username: root.username, password: root.password) + } + } +} + +/// `basic_messaging.feature`. +enum BasicMessagingSteps { + static func register(into registry: inout StepRegistry) { + registry.step("I have no streams in the system") { world, _ in + let streams = try await world.requireClient().getStreams() + try check(streams.isEmpty, "the system should have no streams initially, found \(streams.count)") + } + registry.step("I create a stream with name \"([^\"]*)\"") { world, groups in + let stream = try await world.requireClient().createStream(name: groups[0]) + world.lastStreamID = stream.id + world.lastStreamName = stream.name + } + registry.step("the stream should be created successfully") { world, _ in + try check(world.lastStreamID != nil, "the stream should have been created") + } + registry.step("the stream should have name \"([^\"]*)\"") { world, groups in + try check(world.lastStreamName == groups[0], "expected stream name \(groups[0]), got \(world.lastStreamName ?? "none")") + } + registry.step("I create a topic with name \"([^\"]*)\" in stream (\\d+) with (\\d+) partitions") { world, groups in + let topic = try await world.requireClient().createTopic( + streamID: Identifier(numeric: try number(groups[1])), name: groups[0], + options: TopicCreateOptions(partitionsCount: try number(groups[2]), messageExpiry: .never)) + world.lastTopicID = topic.id + world.lastTopicName = topic.name + world.lastTopicPartitions = topic.partitionsCount + } + registry.step("the topic should be created successfully") { world, _ in + try check(world.lastTopicID != nil, "the topic should have been created") + } + registry.step("the topic should have name \"([^\"]*)\"") { world, groups in + try check(world.lastTopicName == groups[0], "expected topic name \(groups[0]), got \(world.lastTopicName ?? "none")") + } + registry.step("the topic should have (\\d+) partitions") { world, groups in + let expected: UInt32 = try number(groups[0]) + try check(world.lastTopicPartitions == expected, "expected \(expected) partitions, got \(world.lastTopicPartitions.map { "\($0)" } ?? "none")") + } + registry.step("I send (\\d+) messages to stream (\\d+), topic (\\d+), partition (\\d+)") { world, groups in + let count: UInt32 = try number(groups[0]) + var messages = try (0.. PolledMessages { + guard let polled = world.lastPolled else { + throw StepFailure("no messages were polled in this scenario") + } + return polled + } +} + +/// `raw_command.feature`. +enum RawCommandSteps { + static func register(into registry: inout StepRegistry) { + registry.step("I send a raw command with code (\\d+) and an empty payload") { world, groups in + let client = try world.requireClient() + let code: UInt32 = try number(groups[0]) + await world.captureReply { try await client.sendRawRequest(code: code, payload: []) } + } + registry.step("I send a raw command with code (\\d+) and an empty payload to replica (\\d+)") { world, groups in + let client = try world.requireClient() + let code: UInt32 = try number(groups[0]) + let replica: UInt8 = try number(groups[1]) + await world.captureReply { try await client.sendRawRequest(code: code, payload: [], replica: replica) } + } + registry.step("the raw command should succeed with an empty response") { world, _ in + try check(world.lastError == nil, "the raw command failed: \(world.lastError.map { "\($0)" } ?? "")") + try check(world.lastRawResponse?.isEmpty == true, "expected an empty response, got \(world.lastRawResponse?.count ?? 0) bytes") + } + registry.step("the raw command should succeed with a non-empty response") { world, _ in + try check(world.lastError == nil, "the raw command failed: \(world.lastError.map { "\($0)" } ?? "")") + try check(world.lastRawResponse?.isEmpty == false, "expected a non-empty response") + } + registry.step("the raw command should fail with an invalid command error") { world, _ in + try world.expectFailure(.invalidCommand) + } + registry.step("the raw command should fail with an invalid identifier error") { world, _ in + try world.expectFailure(.invalidIdentifier) + } + } +} + +/// `session_contract.feature`. +enum SessionContractSteps { + static func register(into registry: inout StepRegistry) { + registry.step("I have a cluster node that cannot reach quorum") { world, _ in + world.serverAddress = try Environment.noQuorumAddress() + } + registry.step("I sign out") { world, _ in + try await world.requireClient().logout() + } + registry.step("I request the stream with ID (\\d+)") { world, groups in + let client = try world.requireClient() + let streamID = Identifier(numeric: try number(groups[0])) + await world.capture { _ = try await client.getStream(streamID) } + } + registry.step("I sign in to that node as the root user") { world, _ in + guard let address = world.serverAddress else { + throw StepFailure("the node address was not read; the steps are out of order") + } + let client = try await world.connect(to: address) + let root = try Environment.rootCredentials() + await world.capture { _ = try await client.login(username: root.username, password: root.password) } + } + registry.step("the call should fail with error code (\\d+)") { world, groups in + let code: UInt32 = try number(groups[0]) + guard let error = world.lastError as? IggyError else { + throw StepFailure("expected error code \(code), got \(world.lastError.map { "\($0)" } ?? "success")") + } + try check(error.rawCode == code, "expected error code \(code), got \(error)") + } + } +} + +extension World { + /// Runs a call whose failure a later step checks. + func capture(_ call: () async throws -> Void) async { + lastError = nil + do { + try await call() + } catch { + lastError = error + } + } + + /// Runs a raw request whose reply or failure a later step checks. + func captureReply(_ call: () async throws -> [UInt8]) async { + lastRawResponse = nil + await capture { lastRawResponse = try await call() } + } + + func expectFailure(_ code: IggyErrorCode) throws { + guard let error = lastError as? IggyError else { + throw StepFailure("expected \(code.name), got \(lastError.map { "\($0)" } ?? "success")") + } + try check(error.code == code, "expected \(code.name), got \(error)") + } +} diff --git a/bdd/swift/tests/Sources/iggy-bdd/MetadataChangesSteps.swift b/bdd/swift/tests/Sources/iggy-bdd/MetadataChangesSteps.swift new file mode 100644 index 0000000000..a6cf2908e7 --- /dev/null +++ b/bdd/swift/tests/Sources/iggy-bdd/MetadataChangesSteps.swift @@ -0,0 +1,265 @@ +// 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 Foundation +import Iggy + +/// `metadata_changes.feature`, on top of the stream and topic steps of +/// ``ResourceOptionsSteps``. +enum MetadataChangesSteps { + /// A two phase delete commits its record after the call returns. + static let followAttempts = 100 + static let followInterval: Duration = .milliseconds(100) + static let rootUserID: UInt32 = 0 + + static func register(into registry: inout StepRegistry) { + registry.step("I hold the latest metadata change cursor for the groups \"([^\"]+)\"") { world, groups in + world.changeGroups = try parseGroups(groups[0]) + let changes = try await read(world, from: .latest) + guard changes.status == .ok, let next = changes.next else { + throw StepFailure("latest answered \(changes.status) with cursor \(changes.next.map { "\($0)" } ?? "none")") + } + world.changeCursor = next + } + registry.step("I delete the stream") { world, _ in + let streamID = try world.requireStream() + try await world.requireClient().deleteStream(streamID) + world.createdStreams.removeAll { $0 == world.lastStreamID } + } + registry.step("I create and delete (\\d+) streams named \"([^\"]+)\"") { world, groups in + let client = try world.requireClient() + for index in 0..<(try number(groups[0], as: Int.self)) { + let stream = try await client.createStream(name: "\(groups[1])-\(index)") + try await client.deleteStream(Identifier(numeric: stream.id)) + } + } + registry.step("following the metadata changes should return the kinds \"([^\"]+)\" for the stream") { world, groups in + try await followKinds(world, groups[0]) + } + registry.step("each followed change should name the root user as its actor") { world, _ in + for record in world.followedChanges { + try check(record.actorUserID == rootUserID, "op \(record.op) names actor \(record.actorUserID.map { "\($0)" } ?? "none"), want the root user") + } + } + registry.step("the followed \"([^\"]+)\" change should carry a requested op before its own op") { world, groups in + let record = try followedRecord(world, groups[0]) + guard case .streamDeleted(let deleted) = record.change else { + throw StepFailure("expected a stream deletion, got \(record.change)") + } + guard let requestedOp = deleted.requestedOp, requestedOp < record.op else { + throw StepFailure("requested op \(deleted.requestedOp.map { "\($0)" } ?? "none") is not before op \(record.op)") + } + } + registry.step("the followed \"([^\"]+)\" change should report encryption key source \"([^\"]+)\"") { world, groups in + let source = try followedStreamOptions(world, groups[0])[StreamOptionKey.encryptionKeySource.rawValue]?.value.stringValue + try check(source == groups[1], "encryption key source is \(source ?? "unreported"), want \(groups[1])") + } + registry.step("the followed \"([^\"]+)\" change should carry the stream key fingerprint") { world, groups in + let fingerprint = try followedStreamOptions(world, groups[0])[StreamOptionKey.encryptionKeyFingerprint.rawValue]?.value.stringValue + let expected = try EncryptionKey.fingerprint(try ResourceOptionsSteps.streamKey(world)) + try check(fingerprint == expected, "encryption key fingerprint is \(fingerprint ?? "unreported"), want \(expected)") + } + registry.step("the followed changes should expose no encryption key material") { world, _ in + for record in world.followedChanges { + guard let stream = streamChange(record.change) else { + continue + } + guard case .readable(let options) = stream.options else { + throw StepFailure("op \(record.op) options must stay readable") + } + try ResourceOptionsSteps.checkNoKeyMaterial(options, key: world.streamKey, label: "op \(record.op)") + } + } + registry.step("following the metadata changes should answer \"([^\"]+)\"") { world, groups in + let changes = try await read(world, from: .after(try cursor(world))) + try check(changes.status.description == groups[0], "follow answered \(changes.status), want \(groups[0])") + try check( + changes.status == .ok || (changes.records.isEmpty && changes.next == nil), + "a \(changes.status) answer carries \(changes.records.count) records and a cursor") + } + registry.step("I read the current state for the groups \"([^\"]+)\"") { world, groups in + world.changeGroups = try parseGroups(groups[0]) + world.currentItems = [] + var token: MetadataCurrentPageToken? + while true { + let changes = try await read(world, from: .current(token)) + try check(changes.status == .ok, "current state answered \(changes.status)") + world.currentItems += changes.items + guard let next = changes.currentPageToken else { + guard let anchor = changes.next else { + throw StepFailure("the last current state page returned no anchor") + } + world.changeCursor = anchor + return + } + token = next + } + } + registry.step("the current state should list the same streams as the stream list") { world, _ in + let listed = Dictionary(uniqueKeysWithValues: try await world.requireClient().getStreams().map { ($0.id, $0.name) }) + var current: [UInt32: String] = [:] + for case .stream(let stream) in world.currentItems { + current[stream.streamID] = stream.name + } + try check(current == listed, "current state lists \(current), the stream list \(listed)") + } + registry.step("a user named \"([^\"]+)\" with password \"([^\"]+)\" and no server permissions exists") { world, groups in + let permissions = Permissions(global: GlobalPermissions(readServers: true)) + try await ResourceOptionsSteps.createUser(world, username: groups[0], password: groups[1], permissions: permissions) + } + registry.step("I authenticate as \"([^\"]+)\" with password \"([^\"]+)\"") { world, groups in + let client = try world.requireClient() + try await client.logout() + world.signedInAsAnotherUser = true + _ = try await client.login(username: groups[0], password: groups[1]) + } + registry.step("reading the metadata changes should be refused as unauthorized") { world, _ in + world.changeGroups = [] + await world.capture { _ = try await read(world, from: .earliest) } + try world.expectFailure(.unauthorized) + } + } + + /// Follows the log after the held cursor until the scenario's stream has + /// as many records as `kinds` names, then compares their kinds in order. + private static func followKinds(_ world: World, _ kinds: String) async throws { + let want = try kinds.split(separator: ",").map { name in + let trimmed = name.trimmingCharacters(in: .whitespaces) + guard let kind = MetadataChangeKind.allCases.first(where: { $0.description == trimmed }) else { + throw StepFailure("unknown change kind \(trimmed)") + } + return kind + } + let key = try await createdStreamKey(world) + var position = try cursor(world) + var records: [MetadataChangeRecord] = [] + var mine: [MetadataChangeRecord] = [] + for _ in 0..= want.count { + break + } + try await Task.sleep(for: followInterval) + } + for record in records where !world.changeGroups.isEmpty { + try check( + record.change.group.map(world.changeGroups.contains) == true, + "op \(record.op) of group \(record.change.group.map { "\($0)" } ?? "none") is outside the requested \(world.changeGroups)") + } + let got = mine.map(\.change.kind) + try check(got == want, "followed kinds \(got), want \(want)") + world.followedChanges = mine + } + + /// The stream id and creating op of the scenario's stream, read from its + /// creation record. Ids are reused after a delete, so an id alone can + /// match a stream another scenario created and deleted meanwhile. + private static func createdStreamKey(_ world: World) async throws -> StreamKey { + guard let streamID = world.lastStreamID, let name = world.lastStreamName else { + throw StepFailure("no stream was created in this scenario") + } + var query = MetadataChangesQuery(start: .after(try cursor(world)), groups: [.streams], limit: MetadataChangesQuery.maxLimit) + while true { + let changes = try await world.requireClient().getMetadataChanges(query) + guard changes.status == .ok, let next = changes.next else { + throw StepFailure("stream changes answered \(changes.status) with cursor \(changes.next.map { "\($0)" } ?? "none")") + } + for record in changes.records { + if case .streamCreated(let created) = record.change, created.streamID == streamID, created.name == name { + return StreamKey(streamID: created.streamID, createdOp: created.streamCreatedOp) + } + } + guard changes.hasMore else { + throw StepFailure("no stream_created change for stream \(streamID) \(name)") + } + query.start = .after(next) + } + } + + struct StreamKey: Equatable { + let streamID: UInt32 + let createdOp: UInt64 + } + + /// The stream a stream, topic or partitions record belongs to. + private static func streamKey(_ change: MetadataChange) -> StreamKey? { + switch change { + case .streamCreated(let body), .streamUpdated(let body): StreamKey(streamID: body.streamID, createdOp: body.streamCreatedOp) + case .streamDeleted(let body): StreamKey(streamID: body.streamID, createdOp: body.streamCreatedOp) + case .topicCreated(let body), .topicUpdated(let body): StreamKey(streamID: body.streamID, createdOp: body.streamCreatedOp) + case .topicDeleted(let body): StreamKey(streamID: body.streamID, createdOp: body.streamCreatedOp) + case .partitionsChanged(let body): StreamKey(streamID: body.streamID, createdOp: body.streamCreatedOp) + default: nil + } + } + + private static func streamChange(_ change: MetadataChange) -> MetadataStreamChange? { + switch change { + case .streamCreated(let body), .streamUpdated(let body): body + default: nil + } + } + + private static func followedRecord(_ world: World, _ kind: String) throws -> MetadataChangeRecord { + guard let record = world.followedChanges.first(where: { $0.change.kind?.description == kind }) else { + throw StepFailure("no followed \(kind) change") + } + return record + } + + private static func followedStreamOptions(_ world: World, _ kind: String) throws -> ResourceOptions { + let record = try followedRecord(world, kind) + guard let stream = streamChange(record.change) else { + throw StepFailure("expected a stream change, got \(record.change)") + } + guard case .readable(let options) = stream.options else { + throw StepFailure("stream options must stay readable") + } + return options + } + + private static func read(_ world: World, from start: MetadataChangeStart) async throws -> MetadataChanges { + try await world.requireClient().getMetadataChanges( + MetadataChangesQuery(start: start, groups: world.changeGroups, limit: MetadataChangesQuery.maxLimit)) + } + + private static func cursor(_ world: World) throws -> MetadataChangeCursor { + guard let cursor = world.changeCursor else { + throw StepFailure("no metadata change cursor is held") + } + return cursor + } + + private static func parseGroups(_ names: String) throws -> [MetadataChangeGroup] { + try names.split(separator: ",").map { name in + let trimmed = name.trimmingCharacters(in: .whitespaces) + guard let group = MetadataChangeGroup.allCases.first(where: { $0.description == trimmed }) else { + throw StepFailure("unknown change group \(trimmed)") + } + return group + } + } +} diff --git a/bdd/swift/tests/Sources/iggy-bdd/ResourceOptionsSteps.swift b/bdd/swift/tests/Sources/iggy-bdd/ResourceOptionsSteps.swift new file mode 100644 index 0000000000..6499beff1b --- /dev/null +++ b/bdd/swift/tests/Sources/iggy-bdd/ResourceOptionsSteps.swift @@ -0,0 +1,289 @@ +// 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 Foundation +import Iggy + +/// `stream_options.feature` and `resource_options_patch.feature`, whose +/// stream and topic steps `metadata_changes.feature` shares. +enum ResourceOptionsSteps { + /// The partition every message of these scenarios goes to. + static let partitionID: UInt32 = 0 + /// Option keys no read may ever return. + static let hiddenKeys = ["encryption_key", "encryption_key_wrapped", "critical_keys"] + + static func register(into registry: inout StepRegistry) { + registerStreams(into: ®istry) + registerMessages(into: ®istry) + registerTopics(into: ®istry) + registerUsers(into: ®istry) + } + + private static func registerStreams(into registry: inout StepRegistry) { + let keyed: StepHandler = { world, groups in + let key = Data(repeating: try number(groups[1], as: UInt8.self), count: EncryptionKey.length) + try await createStream(world, name: groups[0], options: StreamCreateOptions(encryptionKey: key)) + world.streamKey = key + } + let keyless: StepHandler = { world, groups in + world.streamKey = nil + try await createStream(world, name: groups[0], options: StreamCreateOptions()) + } + registry.step("I create a stream named \"([^\"]+)\" with an encryption key of 32 bytes valued (\\d+)", keyed) + registry.step("a stream named \"([^\"]+)\" with an encryption key of 32 bytes valued (\\d+) exists", keyed) + registry.step("I create a stream named \"([^\"]+)\" without an encryption key", keyless) + registry.step("a stream named \"([^\"]+)\" exists", keyless) + registry.step("the stream should report encryption key source \"([^\"]+)\"") { world, groups in + let stream = try await fetchStream(world) + try check( + stream.encryptionKeySource?.rawValue == groups[0], + "encryption key source is \(stream.encryptionKeySource.map { $0.rawValue } ?? "unreported"), want \(groups[0])") + } + registry.step("the stream encryption key fingerprint should match the key") { world, _ in + let stream = try await fetchStream(world) + let expected = try EncryptionKey.fingerprint(try streamKey(world)) + try check( + stream.encryptionKeyFingerprint == expected, + "encryption key fingerprint is \(stream.encryptionKeyFingerprint ?? "unreported"), want \(expected)") + } + registry.step("the stream should report an encryption key fingerprint") { world, _ in + let fingerprint = try await fetchStream(world).encryptionKeyFingerprint + try check(fingerprint?.count == 16, "encryption key fingerprint is \(fingerprint ?? "unreported"), want 16 hex characters") + } + registry.step("the stream should report at least one KEK fingerprint") { world, _ in + try check(!(try await fetchStream(world).encryptionKekFingerprints.isEmpty), "the stream reports no KEK fingerprint") + } + registry.step("the stream should expose no encryption key material") { world, _ in + try checkNoKeyMaterial(try await fetchStream(world).options, key: world.streamKey, label: "the stream") + } + registry.step("I rename the stream to \"([^\"]+)\"") { world, groups in + try await world.requireClient().updateStream(try world.requireStream(), options: StreamUpdateOptions(name: groups[0])) + } + registry.step("the stream name should be \"([^\"]+)\"") { world, groups in + let name = try await fetchStream(world).name + try check(name == groups[0], "stream name is \(name), want \(groups[0])") + } + registry.step("I request a reset of the stream option \"([^\"]+)\"") { world, groups in + let client = try world.requireClient() + let streamID = try world.requireStream() + let options = StreamOptionKey(rawValue: groups[0]).map { StreamUpdateOptions(reset: [$0]) } ?? StreamUpdateOptions(rawReset: [groups[0]]) + await world.capture { try await client.updateStream(streamID, options: options) } + } + registry.step("the stream update should be rejected") { world, _ in + try check(world.lastError != nil, "the update must fail") + } + registry.step("I describe the stream options") { world, _ in + world.optionSpecs = try await world.requireClient().describeOptions(scope: .stream) + } + registry.step("the stream option \"([^\"]+)\" should be updatable and not secret") { world, groups in + let flags = try optionFlags(world, groups[0]) + try check(flags.contains(.updatable) && !flags.contains(.secret) && !flags.contains(.readOnly), "\(groups[0]) has flags \(flags.rawValue)") + } + registry.step("the stream option \"([^\"]+)\" should be create only and secret") { world, groups in + let flags = try optionFlags(world, groups[0]) + try check( + flags.contains(.secret) && !flags.contains(.updatable) && !flags.contains(.resettable) && !flags.contains(.readOnly), + "\(groups[0]) has flags \(flags.rawValue)") + } + registry.step("the stream option \"([^\"]+)\" should be read only") { world, groups in + let flags = try optionFlags(world, groups[0]) + try check(flags.contains(.readOnly) && !flags.contains(.updatable) && !flags.contains(.secret), "\(groups[0]) has flags \(flags.rawValue)") + } + } + + private static func registerMessages(into registry: inout StepRegistry) { + registry.step("I send a message with payload \"([^\"]+)\" to the topic") { world, groups in + try await send(world, try IggyMessage(groups[0])) + } + registry.step("I send a message with payload \"([^\"]+)\" and header \"([^\"]+)\" set to \"([^\"]+)\" to the topic") { world, groups in + try await send(world, try IggyMessage(groups[0], userHeaders: [try HeaderKey(groups[1]): try HeaderValue.string(groups[2])])) + } + registry.step("polling the topic should return the payload \"([^\"]+)\"") { world, groups in + let message = try await pollFirst(world) + try check(message.payloadString == groups[0], "polled payload \(message.payloadString), want \(groups[0])") + } + registry.step("polling the topic should return the payload \"([^\"]+)\" and header \"([^\"]+)\" set to \"([^\"]+)\"") { world, groups in + let message = try await pollFirst(world) + try check(message.payloadString == groups[0], "polled payload \(message.payloadString), want \(groups[0])") + let value = message.userHeaders?[try HeaderKey(groups[1])] + try check(value == (try HeaderValue.string(groups[2])), "polled headers \(message.userHeaders ?? [:]) do not carry \(groups[1])=\(groups[2])") + } + } + + private static func registerTopics(into registry: inout StepRegistry) { + registry.step("I create a topic named \"([^\"]+)\" with (\\d+) partitions? in the stream") { world, groups in + let options = TopicCreateOptions(partitionsCount: try number(groups[1]), messageExpiry: .never) + world.lastTopicID = try await world.requireClient().createTopic(streamID: try world.requireStream(), name: groups[0], options: options).id + } + registry.step( + "a topic named \"([^\"]+)\" exists in the stream with message expiry (\\d+) microseconds and max topic size (\\d+) bytes" + ) { world, groups in + let options = TopicCreateOptions( + partitionsCount: 1, messageExpiry: .after(.microseconds(try number(groups[1], as: UInt64.self))), + maxTopicSize: .bytes(try number(groups[2]))) + world.lastTopicID = try await world.requireClient().createTopic(streamID: try world.requireStream(), name: groups[0], options: options).id + } + registry.step("I update the topic message expiry to (\\d+) microseconds without a name") { world, groups in + try await updateTopic(world, TopicUpdateOptions(messageExpiry: .after(.microseconds(try number(groups[0], as: UInt64.self))))) + } + registry.step("I reset the topic option \"([^\"]+)\"") { world, groups in + try await updateTopic( + world, TopicOptionKey(rawValue: groups[0]).map { TopicUpdateOptions(reset: [$0]) } ?? TopicUpdateOptions(rawReset: [groups[0]])) + } + registry.step("I rename the topic to \"([^\"]+)\"") { world, groups in + try await updateTopic(world, TopicUpdateOptions(name: groups[0])) + } + registry.step("the topic name should be \"([^\"]+)\"") { world, groups in + let name = try await fetchTopic(world).name + try check(name == groups[0], "topic name is \(name), want \(groups[0])") + } + registry.step("the topic message expiry should be (\\d+) microseconds set explicitly") { world, groups in + let topic = try await fetchTopic(world) + let expected: UInt64 = try number(groups[0]) + try check(topic.messageExpiry == .after(.microseconds(expected)), "message expiry is \(topic.messageExpiry), want \(expected) microseconds") + try check(topic.options[TopicOptionKey.messageExpiry.rawValue]?.explicit == true, "message expiry must be set explicitly") + } + registry.step("the topic max topic size should be (\\d+) bytes set explicitly") { world, groups in + let topic = try await fetchTopic(world) + let expected: UInt64 = try number(groups[0]) + try check(topic.maxTopicSize == .bytes(expected), "max topic size is \(topic.maxTopicSize), want \(expected) bytes") + try check(topic.options[TopicOptionKey.maxTopicSize.rawValue]?.explicit == true, "max topic size must be set explicitly") + } + registry.step("the topic max topic size should be derived from the server default") { world, _ in + let option = try await fetchTopic(world).options[TopicOptionKey.maxTopicSize.rawValue] + try check(option?.explicit == false, "max topic size must be derived from the server default, got \(option.map { "\($0)" } ?? "nothing")") + } + } + + private static func registerUsers(into registry: inout StepRegistry) { + registry.step("a user named \"([^\"]+)\" with password \"([^\"]+)\" exists") { world, groups in + try await createUser(world, username: groups[0], password: groups[1], permissions: nil) + } + registry.step("I update the user username to \"([^\"]+)\" and status to \"([^\"]+)\"") { world, groups in + let options = UserUpdateOptions(username: groups[0], status: try userStatus(groups[1])) + try await world.requireClient().updateUser(try world.requireUser(), options: options) + } + registry.step("the user should have username \"([^\"]+)\"") { world, groups in + let username = try await fetchUser(world).username + try check(username == groups[0], "username is \(username), want \(groups[0])") + } + registry.step("the user should have status \"([^\"]+)\"") { world, groups in + let status = try await fetchUser(world).status + try check(status == (try userStatus(groups[0])), "user status is \(status), want \(groups[0])") + } + } + + static func createStream(_ world: World, name: String, options: StreamCreateOptions) async throws { + let stream = try await world.requireClient().createStream(name: name, options: options) + world.lastStreamID = stream.id + world.lastStreamName = name + world.createdStreams.append(stream.id) + } + + static func createUser(_ world: World, username: String, password: String, permissions: Permissions?) async throws { + let user = try await world.requireClient().createUser(username: username, password: password, permissions: permissions) + world.userID = user.id + world.createdUsers.append(user.id) + } + + static func streamKey(_ world: World) throws -> Data { + guard let key = world.streamKey else { + throw StepFailure("the scenario created no keyed stream") + } + return key + } + + /// Fails when `options` returns a hidden key or carries `key` raw, in + /// base64 or in hex. + static func checkNoKeyMaterial(_ options: ResourceOptions, key: Data?, label: String) throws { + for hidden in hiddenKeys { + try check(options[hidden] == nil, "\(label) returns \(hidden)") + } + guard let key else { + return + } + let spellings = [Array(key), Array(key.base64EncodedString().utf8), Array(key.map { String(format: "%02x", $0) }.joined().utf8)] + for (name, option) in options { + for spelling in spellings { + try check(!contains(option.value.bytes, spelling), "\(label) option \(name) carries the stream key") + } + } + } + + private static func contains(_ bytes: [UInt8], _ needle: [UInt8]) -> Bool { + guard bytes.count >= needle.count else { + return false + } + return (0...(bytes.count - needle.count)).contains { start in bytes[start..<(start + needle.count)].elementsEqual(needle) } + } + + private static func send(_ world: World, _ message: IggyMessage) async throws { + var messages = [message] + try await world.requireClient().sendMessages( + streamID: try world.requireStream(), topicID: try world.requireTopic(), partitioning: .partition(partitionID), messages: &messages) + } + + private static func pollFirst(_ world: World) async throws -> IggyMessage { + let polled = try await world.requireClient().pollMessages( + streamID: try world.requireStream(), topicID: try world.requireTopic(), partitionID: partitionID, strategy: .offset(0), + options: PollOptions(count: 1)) + guard let message = polled.messages.first else { + throw StepFailure("the poll returned no message") + } + return message + } + + private static func updateTopic(_ world: World, _ options: TopicUpdateOptions) async throws { + try await world.requireClient().updateTopic(streamID: try world.requireStream(), topicID: try world.requireTopic(), options: options) + } + + private static func optionFlags(_ world: World, _ key: String) throws -> OptionFlags { + guard let spec = world.optionSpecs.first(where: { $0.key == key }) else { + throw StepFailure("the catalog does not describe \(key)") + } + return spec.flags + } + + private static func fetchStream(_ world: World) async throws -> StreamDetails { + guard let stream = try await world.requireClient().getStream(try world.requireStream()) else { + throw StepFailure("the stream is gone") + } + return stream + } + + private static func fetchTopic(_ world: World) async throws -> TopicDetails { + guard let topic = try await world.requireClient().getTopic(streamID: try world.requireStream(), topicID: try world.requireTopic()) else { + throw StepFailure("the topic is gone") + } + return topic + } + + private static func fetchUser(_ world: World) async throws -> UserInfoDetails { + guard let user = try await world.requireClient().getUser(try world.requireUser()) else { + throw StepFailure("the user is gone") + } + return user + } + + private static func userStatus(_ name: String) throws -> UserStatus { + switch name { + case "active": .active + case "inactive": .inactive + default: throw StepFailure("unknown user status \(name)") + } + } +} diff --git a/bdd/swift/tests/Sources/iggy-bdd/Runner.swift b/bdd/swift/tests/Sources/iggy-bdd/Runner.swift new file mode 100644 index 0000000000..99463fbc74 --- /dev/null +++ b/bdd/swift/tests/Sources/iggy-bdd/Runner.swift @@ -0,0 +1,267 @@ +// 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 Foundation +import Iggy + +/// The endpoints and credentials a suite runs against. A default here would +/// turn a dropped compose variable into a run against whatever listens on +/// the fallback address, so a missing value fails the step instead. +enum Environment { + static func required(_ name: String) throws -> String { + guard let value = ProcessInfo.processInfo.environment[name], !value.isEmpty else { + throw StepFailure("\(name) must be set; run the suite via scripts/run-bdd-tests.sh") + } + return value + } + + static func serverAddress() throws -> String { + try required("IGGY_TCP_ADDRESS") + } + + static func noQuorumAddress() throws -> String { + try required("IGGY_TCP_ADDRESS_NO_QUORUM") + } + + static func rootCredentials() throws -> (username: String, password: String) { + (try required("IGGY_ROOT_USERNAME"), try required("IGGY_ROOT_PASSWORD")) + } +} + +struct StepFailure: Error, CustomStringConvertible { + let description: String + + init(_ description: String) { + self.description = description + } +} + +/// Fails the step unless `condition` holds. +func check(_ condition: Bool, _ message: @autoclosure () -> String) throws { + guard condition else { + throw StepFailure(message()) + } +} + +/// Parses a captured decimal group. +func number(_ text: String, as type: Value.Type = Value.self) throws -> Value { + guard let value = Value(text) else { + throw StepFailure("\(text) is not a \(Value.self)") + } + return value +} + +/// The state one scenario's steps share. Steps run one after another, so +/// nothing touches it concurrently. +final class World: @unchecked Sendable { + var serverAddress: String? + var client: IggyClient? + /// The outcome of the last call a step made to see it fail. + var lastError: (any Error)? + + var lastSentMessage: IggyMessage? + var lastPolled: PolledMessages? + var lastRawResponse: [UInt8]? + var lastStreamID: UInt32? + var lastStreamName: String? + var lastTopicID: UInt32? + var lastTopicName: String? + var lastTopicPartitions: UInt32? + + var streamKey: Data? + var userID: UInt32? + var createdStreams: [UInt32] = [] + var createdUsers: [UInt32] = [] + var optionSpecs: [OptionSpec] = [] + + var changeGroups: [MetadataChangeGroup] = [] + var changeCursor: MetadataChangeCursor? + var followedChanges: [MetadataChangeRecord] = [] + var currentItems: [MetadataCurrentItem] = [] + var signedInAsAnotherUser = false + + var poll: Task? + var pollStarted = ContinuousClock.now + var pollEnd = PollEnd() + var reader: PartitionReader? + var consumer: IggyConsumer? + /// The operation of a committing poll whose outcome is unknown. + var lostOperation: UInt128Value? + + func requireClient() throws -> IggyClient { + guard let client else { + throw StepFailure("the client is not connected; the background must sign in first") + } + return client + } + + func requireStream() throws -> Identifier { + guard let lastStreamID else { + throw StepFailure("no stream was created in this scenario") + } + return Identifier(numeric: lastStreamID) + } + + func requireTopic() throws -> Identifier { + guard let lastTopicID else { + throw StepFailure("no topic was created in this scenario") + } + return Identifier(numeric: lastTopicID) + } + + func requireUser() throws -> Identifier { + guard let userID else { + throw StepFailure("no user was created in this scenario") + } + return Identifier(numeric: userID) + } + + func requirePoll() throws -> Task { + guard let poll else { + throw StepFailure("no deferred poll was started") + } + return poll + } + + func requireConsumer() throws -> IggyConsumer { + guard let consumer else { + throw StepFailure("no consumer is open") + } + return consumer + } + + /// A client for `address`, connected but not signed in. + func connect(to address: String) async throws -> IggyClient { + let client = IggyClient(address: address) + self.client = client + try await client.connect() + return client + } + + /// Signs back in as root when a step switched users, deletes what the + /// scenario created, and shuts the client down. A failed cleanup fails + /// the scenario, since it would leak into the next one. + func tearDown() async -> [any Error] { + poll?.cancel() + if let reader { + try? await reader.close() + } + await consumer?.shutdown() + guard let client else { + return [] + } + var failures: [any Error] = [] + if signedInAsAnotherUser { + do { + let root = try Environment.rootCredentials() + try await client.logout() + _ = try await client.login(username: root.username, password: root.password) + } catch { + failures.append(error) + } + } + for id in createdStreams { + do { try await client.deleteStream(Identifier(numeric: id)) } catch { failures.append(error) } + } + for id in createdUsers { + do { try await client.deleteUser(Identifier(numeric: id)) } catch { failures.append(error) } + } + await client.shutdown() + self.client = nil + return failures + } +} + +typealias StepHandler = @Sendable (World, [String]) async throws -> Void + +/// The step definitions of one feature, matched by regular expression. +struct StepRegistry { + private var definitions: [(pattern: NSRegularExpression, handler: StepHandler)] = [] + + mutating func step(_ pattern: String, _ handler: @escaping StepHandler) { + // Anchored, so one text cannot match two definitions that differ only + // in a suffix. + guard let expression = try? NSRegularExpression(pattern: "^" + pattern + "$") else { + preconditionFailure("invalid step pattern \(pattern)") + } + definitions.append((expression, handler)) + } + + func match(_ text: String) throws -> (StepHandler, [String]) { + let range = NSRange(text.startIndex.. (StepHandler, [String])? in + guard let match = definition.pattern.firstMatch(in: text, range: range) else { + return nil + } + let groups = (1.. String in + guard let groupRange = Range(match.range(at: index), in: text) else { + return "" + } + return String(text[groupRange]) + } + return (definition.handler, groups) + } + switch matches.count { + case 0: throw StepFailure("undefined step") + case 1: return matches[0] + default: throw StepFailure("ambiguous step, \(matches.count) definitions match") + } + } +} + +/// Runs a feature and prints a cucumber-style report. Strict: an undefined +/// or ambiguous step fails its scenario. +struct Runner { + let registry: StepRegistry + + func run(_ feature: Feature) async -> (passed: Int, failed: Int) { + print("Feature: \(feature.name)") + var passed = 0 + var failed = 0 + for scenario in feature.scenarios { + print("") + print(" Scenario: \(scenario.name)") + let world = World() + var scenarioFailed = false + for step in feature.background + scenario.steps { + if scenarioFailed { + print(" - \(step.keyword) \(step.text) (skipped)") + continue + } + do { + let (handler, groups) = try registry.match(step.text) + try await handler(world, groups) + print(" \u{2713} \(step.keyword) \(step.text)") + } catch { + scenarioFailed = true + print(" \u{2717} \(step.keyword) \(step.text) (line \(step.line))") + print(" \(error)") + } + } + for failure in await world.tearDown() { + scenarioFailed = true + print(" \u{2717} cleanup: \(failure)") + } + if scenarioFailed { + failed += 1 + } else { + passed += 1 + } + } + return (passed, failed) + } +} diff --git a/bdd/swift/tests/Sources/iggy-bdd/main.swift b/bdd/swift/tests/Sources/iggy-bdd/main.swift new file mode 100644 index 0000000000..698180276f --- /dev/null +++ b/bdd/swift/tests/Sources/iggy-bdd/main.swift @@ -0,0 +1,80 @@ +// 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 Foundation + +// Runs the shared scenarios the Swift SDK supports, each feature with its +// own step definitions as the godog suites do. `BDD_FEATURE` picks one +// feature or `all`; `BDD_SCENARIOS_DIR` points at the feature files and +// defaults to the repository's bdd/scenarios. +let features: [(name: String, register: (inout StepRegistry) -> Void)] = [ + ("basic_messaging", BasicMessagingSteps.register), + ("raw_command", RawCommandSteps.register), + ("session_contract", SessionContractSteps.register), + ("stream_options", ResourceOptionsSteps.register), + ("resource_options_patch", ResourceOptionsSteps.register), + ( + "metadata_changes", + { registry in + ResourceOptionsSteps.register(into: ®istry) + MetadataChangesSteps.register(into: ®istry) + } + ), + ("deferred_polling", DeferredPollingSteps.register), +] + +let selected = ProcessInfo.processInfo.environment["BDD_FEATURE"].flatMap { $0.isEmpty || $0 == "all" ? nil : $0 } +let toRun = features.filter { selected == nil || $0.name == selected } +guard !toRun.isEmpty else { + print("unknown BDD_FEATURE=\(selected ?? ""); supported: \(features.map(\.name).joined(separator: ", "))") + exit(2) +} + +let scenariosDirectory: URL +if let configured = ProcessInfo.processInfo.environment["BDD_SCENARIOS_DIR"], !configured.isEmpty { + scenariosDirectory = URL(fileURLWithPath: configured) +} else { + scenariosDirectory = URL(fileURLWithPath: #filePath) + .deletingLastPathComponent() // main.swift + .deletingLastPathComponent() // iggy-bdd + .deletingLastPathComponent() // Sources + .deletingLastPathComponent() // tests + .deletingLastPathComponent() // swift + .appendingPathComponent("scenarios") +} + +var totalPassed = 0 +var totalFailed = 0 +for feature in toRun { + let path = scenariosDirectory.appendingPathComponent("\(feature.name).feature") + let parsed: Feature + do { + parsed = try GherkinParser.parse(try String(contentsOf: path, encoding: .utf8)) + } catch { + print("cannot load \(path.path): \(error)") + exit(2) + } + var registry = StepRegistry() + CommonSteps.register(into: ®istry) + feature.register(®istry) + let result = await Runner(registry: registry).run(parsed) + totalPassed += result.passed + totalFailed += result.failed + print("") +} +print("\(totalPassed + totalFailed) scenarios (\(totalPassed) passed, \(totalFailed) failed)") +exit(totalFailed == 0 ? 0 : 1) diff --git a/examples/swift/README.md b/examples/swift/README.md new file mode 100644 index 0000000000..ea505d9f29 --- /dev/null +++ b/examples/swift/README.md @@ -0,0 +1,67 @@ +# Iggy Examples + +This directory contains a getting-started producer and consumer using the Swift SDK over TCP, with optional TLS. To learn more about building applications with Iggy, please refer to the [getting started](https://iggy.apache.org/docs/introduction/getting-started) guide. + +## Running Examples + +Use Swift 6.1 or newer. The example package depends on `../../../foreign/swift` by path, so build the server from the same checkout. Start it from the repository root, then run the Swift commands from `examples/swift`. + +For disposable development data and no root credential environment overrides: + +```bash +cargo run --bin iggy-server -- --fresh --with-default-root-credentials +``` + +`--fresh` wipes this replica's local data directory. `IGGY_ROOT_USERNAME` and `IGGY_ROOT_PASSWORD` take precedence over the flag. Bootstrap settings do not replace recovered credentials, including credentials recovered from cluster peers. The examples sign in as `iggy`/`iggy` and address the stream and topic by name, so they run on a server that already holds other streams. + +For server configuration options and help: + +```bash +cargo run --bin iggy-server -- --help +``` + +Each producer creates `sample-stream` and `sample-topic` when they are missing and sends 5 batches of 10 messages to partition `0`. Each consumer reads that partition from its first message and exits after 5 batches. It stores no offset and does not join a consumer group, so multiple consumers read the same messages. + +## Basic Examples + +### Getting Started + +Perfect introduction for newcomers to Iggy: + +```bash +swift run --package-path getting-started producer +swift run --package-path getting-started consumer +``` + +## TLS Examples + +From the repository root, start a disposable TLS server with the development certificate pair and the same credential prerequisites: + +```bash +IGGY_TCP_TLS_ENABLED=true \ +IGGY_TCP_TLS_CERT_FILE=core/certs/iggy_cert.pem \ +IGGY_TCP_TLS_KEY_FILE=core/certs/iggy_key.pem \ +cargo run --bin iggy-server -- --fresh --with-default-root-credentials +``` + +These certificates are for development only. From `examples/swift`, run: + +```bash +swift run --package-path getting-started producer --tcp-server-address localhost:8090 --tls --tls-ca-file ../../core/certs/iggy_ca_cert.pem +swift run --package-path getting-started consumer --tcp-server-address localhost:8090 --tls --tls-ca-file ../../core/certs/iggy_ca_cert.pem +``` + +## Example Structure + +All examples can be executed directly from the repository. Follow these steps: + +1. **Start the Iggy server**: the Swift SDK speaks the VSR wire protocol, so the + examples need a VSR server. + Use the source-server command and prerequisites above. +2. **Run the producer, then the consumer**: use the commands above from `examples/swift`. Both take + `--tcp-server-address host:port`, `--tls`, `--tls-ca-file path` and `--tls-domain name`. +3. **Check source code**: Examples include detailed comments explaining concepts and usage patterns + +These examples use `IggyClient` with TCP transport and automatic sign-in, an `IggyProducer` that creates the stream and topic, and an `IggyConsumer` read with `for try await`. + +The examples are automatically tested via `scripts/run-examples-from-readme.sh --language swift` to ensure they remain functional and up-to-date with the latest API changes. diff --git a/examples/swift/getting-started/Package.resolved b/examples/swift/getting-started/Package.resolved new file mode 100644 index 0000000000..12cd9c4cc7 --- /dev/null +++ b/examples/swift/getting-started/Package.resolved @@ -0,0 +1,78 @@ +{ + "originHash" : "5779d5213e3a88ea6669d73995c16d99f4400dc8c2a7362223e4d44682e2dafb", + "pins" : [ + { + "identity" : "swift-asn1", + "kind" : "remoteSourceControl", + "location" : "https://github.com/apple/swift-asn1.git", + "state" : { + "revision" : "3b6410f7dee09eb33cdd26260c5fd47fda19b0e2", + "version" : "1.7.3" + } + }, + { + "identity" : "swift-atomics", + "kind" : "remoteSourceControl", + "location" : "https://github.com/apple/swift-atomics.git", + "state" : { + "revision" : "0442cb5a3f98ab802acb777929fdb446bda11a34", + "version" : "1.3.1" + } + }, + { + "identity" : "swift-collections", + "kind" : "remoteSourceControl", + "location" : "https://github.com/apple/swift-collections.git", + "state" : { + "revision" : "a0cb0954ecb21e4e31b0070e6ed5674e8556685a", + "version" : "1.6.0" + } + }, + { + "identity" : "swift-crypto", + "kind" : "remoteSourceControl", + "location" : "https://github.com/apple/swift-crypto.git", + "state" : { + "revision" : "95ba0316a9b733e92bb6b071255ff46263bbe7dc", + "version" : "3.15.1" + } + }, + { + "identity" : "swift-log", + "kind" : "remoteSourceControl", + "location" : "https://github.com/apple/swift-log.git", + "state" : { + "revision" : "9c6fb14227f55d8f711ce3847dc2f419fb0ecacb", + "version" : "1.15.1" + } + }, + { + "identity" : "swift-nio", + "kind" : "remoteSourceControl", + "location" : "https://github.com/apple/swift-nio.git", + "state" : { + "revision" : "a931f2c1de8dd49381ce3bf2e279d033f68d8865", + "version" : "2.102.0" + } + }, + { + "identity" : "swift-nio-ssl", + "kind" : "remoteSourceControl", + "location" : "https://github.com/apple/swift-nio-ssl.git", + "state" : { + "revision" : "03827c1a9fdb2b6b00a4e93ede8861520263af8c", + "version" : "2.37.4" + } + }, + { + "identity" : "swift-system", + "kind" : "remoteSourceControl", + "location" : "https://github.com/apple/swift-system.git", + "state" : { + "revision" : "869129b7bf4ecc57b97d0193ad29690ca2134750", + "version" : "1.8.1" + } + } + ], + "version" : 3 +} diff --git a/examples/swift/getting-started/Package.swift b/examples/swift/getting-started/Package.swift new file mode 100644 index 0000000000..f223dbcb4d --- /dev/null +++ b/examples/swift/getting-started/Package.swift @@ -0,0 +1,43 @@ +// swift-tools-version: 6.0 +// 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 PackageDescription + +// Depends on the SDK by path, so the examples build from the checkout. +let package = Package( + name: "getting-started", + platforms: [.macOS(.v13)], + dependencies: [ + .package(name: "apache-iggy", path: "../../../foreign/swift") + ], + targets: [ + .target( + name: "Common", + dependencies: [.product(name: "Iggy", package: "apache-iggy")] + ), + .executableTarget( + name: "producer", + dependencies: ["Common", .product(name: "Iggy", package: "apache-iggy")] + ), + .executableTarget( + name: "consumer", + dependencies: ["Common", .product(name: "Iggy", package: "apache-iggy")] + ), + ], + swiftLanguageModes: [.v6] +) diff --git a/examples/swift/getting-started/Sources/Common/ExampleOptions.swift b/examples/swift/getting-started/Sources/Common/ExampleOptions.swift new file mode 100644 index 0000000000..aff1cd80a3 --- /dev/null +++ b/examples/swift/getting-started/Sources/Common/ExampleOptions.swift @@ -0,0 +1,90 @@ +// 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 Foundation +import Iggy + +/// What the producer and the consumer agree on. +public enum Sample { + public static let rootUsername = "iggy" + public static let rootPassword = "iggy" + public static let streamName = "sample-stream" + public static let topicName = "sample-topic" + public static let partitionID: UInt32 = 0 + public static let messagesPerBatch = 10 + public static let batchesLimit = 5 + public static let interval: Duration = .milliseconds(500) +} + +/// The command-line flags of both examples: +/// +/// ```text +/// --tcp-server-address host:port server to connect to (127.0.0.1:8090) +/// --tls enable TLS +/// --tls-ca-file path CA certificate that signed the server's +/// --tls-domain name server name to verify (defaults to the host) +/// ``` +public struct ExampleOptions: Sendable { + public var serverAddress = "127.0.0.1:8090" + public var tls = false + public var tlsCAFile: String? + public var tlsDomain: String? + + /// Parses the process arguments, and exits with the usage on a mistake. + public static func parse(_ arguments: [String] = Array(CommandLine.arguments.dropFirst())) -> ExampleOptions { + var options = ExampleOptions() + var index = arguments.startIndex + func value(for flag: String) -> String { + index += 1 + guard index < arguments.endIndex else { + usage("\(flag) needs a value") + } + return arguments[index] + } + while index < arguments.endIndex { + switch arguments[index] { + case "--tcp-server-address": options.serverAddress = value(for: "--tcp-server-address") + case "--tls": options.tls = true + case "--tls-ca-file": options.tlsCAFile = value(for: "--tls-ca-file") + case "--tls-domain": options.tlsDomain = value(for: "--tls-domain") + case "--help", "-h": usage(nil) + default: usage("unknown argument \(arguments[index])") + } + index += 1 + } + return options + } + + /// A client for the flags that signs in as root when it connects. + public func makeClient() -> IggyClient { + var configuration = ClientConfiguration( + address: serverAddress, autoLogin: .usernamePassword(username: Sample.rootUsername, password: Sample.rootPassword)) + if tls { + configuration.tls = TLSOptions(domain: tlsDomain, caFile: tlsCAFile) + print("TLS enabled with CA file: \(tlsCAFile ?? "the system roots"), domain: \(tlsDomain ?? "the host")") + } + return IggyClient(configuration: configuration) + } + + private static func usage(_ problem: String?) -> Never { + if let problem { + print("error: \(problem)") + } + print("usage: [--tcp-server-address host:port] [--tls] [--tls-ca-file path] [--tls-domain name]") + exit(problem == nil ? 0 : 2) + } +} diff --git a/examples/swift/getting-started/Sources/consumer/main.swift b/examples/swift/getting-started/Sources/consumer/main.swift new file mode 100644 index 0000000000..3705e8c942 --- /dev/null +++ b/examples/swift/getting-started/Sources/consumer/main.swift @@ -0,0 +1,50 @@ +// 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 Common +import Iggy + +// Reads partition 0 of the sample topic from its first message and exits +// after 5 batches. The consumer stores no offset, so every run reads the +// same messages, and it waits for the producer to create the stream. +let options = ExampleOptions.parse() +let client = options.makeClient() +print("Connecting to server at \(options.serverAddress)") +try await client.connect() + +let consumer = try client.consumer( + name: "getting-started", streamID: try Identifier(named: Sample.streamName), topicID: try Identifier(named: Sample.topicName), + partitions: [Sample.partitionID], + configuration: ConsumerConfiguration( + pollingStrategy: .first, pollOptions: PollOptions(count: UInt32(Sample.messagesPerBatch), maxWait: Sample.interval), + initRetries: 20, initRetryInterval: Sample.interval)) +try await consumer.initialize() + +print( + "Messages will be consumed from stream: \(Sample.streamName), topic: \(Sample.topicName), partition: \(Sample.partitionID) with interval \(Sample.interval)." +) +var consumed = 0 +for try await received in consumer { + print("Handling message at offset: \(received.message.offset), payload: \(received.message.payloadString)...") + consumed += 1 + if consumed == Sample.batchesLimit * Sample.messagesPerBatch { + break + } +} +print("Consumed \(Sample.batchesLimit) batches of messages, exiting.") +await consumer.shutdown() +await client.shutdown() diff --git a/examples/swift/getting-started/Sources/producer/main.swift b/examples/swift/getting-started/Sources/producer/main.swift new file mode 100644 index 0000000000..8dfbb3e2eb --- /dev/null +++ b/examples/swift/getting-started/Sources/producer/main.swift @@ -0,0 +1,50 @@ +// 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 Common +import Iggy + +// Connects, creates the sample stream and topic when they are missing, and +// sends 5 batches of 10 messages to partition 0 before exiting. +let options = ExampleOptions.parse() +let client = options.makeClient() +print("Connecting to server at \(options.serverAddress)") +try await client.connect() + +let producer = try client.producer( + streamID: try Identifier(named: Sample.streamName), topicID: try Identifier(named: Sample.topicName), + configuration: ProducerConfiguration(partitioning: .partition(Sample.partitionID), topicPartitionsCount: 1, topicMessageExpiry: .never)) +try await producer.initialize() +print("Stream and topic are ready.") + +print( + "Messages will be sent to stream: \(Sample.streamName), topic: \(Sample.topicName), partition: \(Sample.partitionID) with interval \(Sample.interval)." +) +var currentID = 0 +for _ in 0..@[::1]:8090?reconnection_retries=unlimited + /// ``` + /// + /// The credentials become ``autoLogin``. The options are `tls`, + /// `tls_domain`, `tls_ca_file`, `reconnection_retries` (a count or + /// `unlimited`), `reconnection_interval`, `reestablish_after`, + /// `heartbeat_interval` (durations such as `5s`, `500ms` or `1h 30m`), + /// `nodelay`, `max_pending_operations` and `max_pending_operation_bytes`. + /// An unknown option fails the parse, so one string is valid in every + /// SDK or in none. + /// + /// Some options read differently than in Rust: + /// + /// - `tls` and `nodelay` take `true` or `false`, and any other value + /// fails, where Rust reads it as false. + /// - Without `nodelay` the client keeps its own default, Nagle off. + /// - `reconnection_interval` is the full pause between rounds over the + /// endpoints. Rust takes it as the base of a jittered backoff capped at + /// 10 seconds. + /// - `reestablish_after` is how long after the last successful + /// connection the client waits before redialing that endpoint. Rust + /// takes it as the uptime that resets the backoff. + /// - A request that needs the connection waits for the redial, up to the + /// request timeout (30 seconds by default), and fails when the wait, or + /// the `reconnection_interval` pause, outlasts it. An explicit + /// `connect()` waits it out. + /// - The pending operation limits are checked and then ignored, because + /// the client keeps no operations to replay after a session reset. + public init(connectionString: String) throws { + self = try ConnectionString.parse(connectionString) + } +} + +/// The parser of `core/common/src/types/configuration/auth_config/connection_string.rs` +/// and the TCP options of `tcp_connection_string_options.rs`. +enum ConnectionString { + private static let defaultPrefix = "iggy://" + private static let transportPrefix = "iggy+" + + static func parse(_ connectionString: String) throws -> ClientConfiguration { + try checkTransport(connectionString) + let protocolParts = connectionString.components(separatedBy: "://") + guard protocolParts.count == 2 else { + throw IggyError(.invalidConnectionString, context: "expected one ://") + } + let parts = protocolParts[1].split(separator: "@", omittingEmptySubsequences: false) + guard parts.count == 2 else { + throw IggyError(.invalidConnectionString, context: "expected one @ after the credentials") + } + let credentials = try parseCredentials(parts[0]) + let serverAndOptions = parts[1].split(separator: "?", omittingEmptySubsequences: false) + guard serverAndOptions.count <= 2 else { + throw IggyError(.invalidConnectionString, context: "expected at most one ?") + } + let address = String(serverAndOptions[0]) + try validate(address: address) + + var configuration = ClientConfiguration(address: address, autoLogin: credentials) + if serverAndOptions.count == 2 { + try parseOptions(serverAndOptions[1], into: &configuration) + } + return configuration + } + + /// Accepts `iggy://` and `iggy+tcp://`. Only a transport spelled in + /// letters is named in the error, since a malformed scheme can hold the + /// credentials. + private static func checkTransport(_ connectionString: String) throws { + if connectionString.hasPrefix(defaultPrefix) { + return + } + guard connectionString.hasPrefix(transportPrefix) else { + throw IggyError(.invalidConnectionString, context: "expected an iggy:// or iggy+tcp:// scheme") + } + let transport = connectionString.dropFirst(transportPrefix.count).components(separatedBy: "://")[0] + guard transport != "tcp" else { + return + } + guard !transport.isEmpty, transport.allSatisfy({ $0.isASCII && $0.isLetter }) else { + throw IggyError(.invalidConnectionString, context: "expected an iggy:// or iggy+tcp:// scheme") + } + throw IggyError(.invalidConnectionString, context: "transport \(transport) is not supported by the Swift SDK, use tcp") + } + + /// `username:password`, or a personal access token alone. + private static func parseCredentials(_ text: Substring) throws -> Credentials { + let pieces = text.split(separator: ":", omittingEmptySubsequences: false) + switch pieces.count { + case 1: + // Rust accepts an empty token and fails at the sign-in instead. + guard !pieces[0].isEmpty else { + throw IggyError(.invalidConnectionString, context: "credentials are required") + } + return .personalAccessToken(String(pieces[0])) + case 2: + guard !pieces[0].isEmpty, !pieces[1].isEmpty else { + throw IggyError(.invalidConnectionString, context: "username and password are required") + } + return .usernamePassword(username: String(pieces[0]), password: String(pieces[1])) + default: + throw IggyError(.invalidConnectionString, context: "expected username:password or a personal access token") + } + } + + /// `host:port`, or `[v6]:port` with an IPv6 literal in the brackets. + private static func validate(address: String) throws { + let host: Substring + let port: Substring + if address.hasPrefix("[") { + guard let close = address.firstIndex(of: "]"), address[address.index(after: close)...].hasPrefix(":") else { + throw IggyError(.invalidConnectionString, context: "expected [address]:port") + } + host = address[address.index(after: address.startIndex).. 0 else { + throw IggyError(.invalidConnectionString, context: "\(pair[0]) must be a positive integer") + } + case "tls": tlsEnabled = try flag(value, option: pair[0]) + case "tls_domain": tls.domain = value + case "tls_ca_file": tls.caFile = value + case "reconnection_retries": retries = value + case "reconnection_interval": interval = value + case "reestablish_after": reestablishAfter = value + case "heartbeat_interval": heartbeatInterval = value + case "nodelay": configuration.noDelay = try flag(value, option: pair[0]) + default: throw IggyError(.invalidConnectionString, context: "unknown option \(pair[0])") + } + } + + configuration.reconnection.enabled = true + if retries == "unlimited" { + configuration.reconnection.maxRetries = nil + } else if let count = UInt32(retries) { + configuration.reconnection.maxRetries = count + } else { + throw IggyError(.invalidNumberValue, context: "reconnection_retries must be a count or unlimited") + } + configuration.reconnection.interval = try duration(interval, option: "reconnection_interval", nonZero: true) + configuration.reconnection.reestablishAfter = try duration(reestablishAfter, option: "reestablish_after", nonZero: false) + configuration.heartbeatInterval = try duration(heartbeatInterval, option: "heartbeat_interval", nonZero: true) + if tlsEnabled { + configuration.tls = tls + } + } + + /// Rust reads any value but `true` as false, so `tls=1` would sign in + /// over plaintext. The Go, Node, Java and C# parsers refuse it instead. + private static func flag(_ value: String, option: Substring) throws -> Bool { + switch value { + case "true": return true + case "false": return false + default: throw IggyError(.invalidConnectionString, context: "\(option) must be true or false") + } + } + + /// Rust types the intervals as `NonZeroIggyDuration`, which refuses zero + /// and anything shorter than the whole microseconds it serializes. A + /// duration past what those microseconds hold would trap in `Task.sleep`. + private static func duration(_ text: String, option: String, nonZero: Bool) throws -> Duration { + guard let value = IggyDuration.parse(text) else { + throw IggyError(.invalidConnectionString, context: "\(option)=\(text) is not a duration") + } + if nonZero, value < .microseconds(1) { + throw IggyError(.invalidConnectionString, context: "\(option) must be at least one microsecond") + } + guard value <= ProducerConfiguration.maxDuration else { + throw IggyError(.invalidConnectionString, context: "\(option) must be at most \(ProducerConfiguration.maxDuration)") + } + return value + } +} + +/// `IggyDuration::from_str` of `core/common/src/utils/duration.rs`: the +/// zero spellings, or the grammar of the humantime crate 2.4 the Rust SDK +/// parses with, on the lowercased text. +enum IggyDuration { + static func parse(_ text: String) -> Duration? { + let lowered = text.lowercased() + if ["0", "unlimited", "disabled", "none"].contains(lowered) { + return .zero + } + var parser = HumantimeParser(lowered) + guard let (seconds, nanoseconds) = parser.parse(), seconds <= UInt64(Int64.max) else { + return nil + } + return Duration(secondsComponent: Int64(seconds), attosecondsComponent: Int64(nanoseconds) * 1_000_000_000) + } +} + +/// A port of `humantime::duration::Parser`. Amounts and units run together +/// or spaced, as in `1h30m`, `1h 30m` or `5 s`; a fraction must come out to +/// whole nanoseconds, or whole seconds for hours and longer units. +private struct HumantimeParser { + private let scalars: [Unicode.Scalar] + private var position = 0 + private var seconds: UInt64 = 0 + private var nanoseconds: UInt64 = 0 + + init(_ text: String) { + scalars = Array(text.unicodeScalars) + } + + mutating func parse() -> (UInt64, UInt64)? { + guard var amount = firstDigit() ?? nil else { + return nil + } + outer: while true { + var fraction: (numerator: UInt64, denominator: UInt64)? + var offset = position + while let scalar = next() { + if let digit = Self.digit(scalar) { + guard let grown = Self.append(digit, to: amount) else { + return nil + } + amount = grown + } else if Self.isUnitLetter(scalar) { + break + } else if scalar == "." { + guard let parsed = fractionalPart(&offset) else { + return nil + } + fraction = parsed + break + } else if !scalar.properties.isWhitespace { + return nil + } + offset = position + } + let start = offset + offset = position + while let scalar = next() { + if let digit = Self.digit(scalar) { + guard add(amount, fraction, unit: start.. UInt64?? { + while let scalar = next() { + if let digit = Self.digit(scalar) { + return .some(digit) + } + guard scalar.properties.isWhitespace else { + return nil + } + } + return .some(nil) + } + + private mutating func fractionalPart(_ offset: inout Int) -> (numerator: UInt64, denominator: UInt64)? { + var numerator: UInt64 = 0 + var denominator: UInt64 = 1 + var leadingZeros = true + while let scalar = next() { + if let digit = Self.digit(scalar) { + guard case (let scaled, false) = denominator.multipliedReportingOverflow(by: 10) else { + return nil + } + denominator = scaled + if digit != 0 || !leadingZeros { + leadingZeros = false + guard let grown = Self.append(digit, to: numerator) else { + return nil + } + numerator = grown + } + } else if Self.isUnitLetter(scalar) { + break + } else if !scalar.properties.isWhitespace { + return nil + } + offset = position + } + // A separator with no digit after it, as in `1.`. + guard denominator != 1 else { + return nil + } + return (numerator, denominator) + } + + private mutating func add(_ amount: UInt64, _ fraction: (numerator: UInt64, denominator: UInt64)?, unit range: Range) -> Bool { + let unit = String(String.UnicodeScalarView(scalars[range])) + guard let (secondsPerUnit, nanosecondsPerUnit) = Self.units[unit] else { + return false + } + // Whole amounts of sub-second units are counted in nanoseconds. + let whole = + secondsPerUnit == 0 + ? Self.multiply(amount, nanosecondsPerUnit).map { (UInt64(0), $0) } + : Self.multiply(amount, secondsPerUnit).map { ($0, UInt64(0)) } + guard let whole, accumulate(whole) else { + return false + } + guard let (numerator, denominator) = fraction else { + return true + } + let part: (UInt64, UInt64)? + switch unit { + case "nanos", "nsec", "ns": + part = nil + case _ where secondsPerUnit >= 3600: + part = Self.multiply(numerator, secondsPerUnit).flatMap { Self.divideExactly($0, denominator) }.map { ($0, UInt64(0)) } + default: + let nanosecondsPerWhole = secondsPerUnit == 0 ? nanosecondsPerUnit : secondsPerUnit * 1_000_000_000 + part = Self.multiply(numerator, nanosecondsPerWhole).flatMap { Self.divideExactly($0, denominator) }.map { (UInt64(0), $0) } + } + guard let part else { + return false + } + return accumulate(part) + } + + private mutating func accumulate(_ part: (seconds: UInt64, nanoseconds: UInt64)) -> Bool { + guard case (var total, false) = nanoseconds.addingReportingOverflow(part.nanoseconds) else { + return false + } + var extra = part.seconds + if total > 1_000_000_000 { + guard case (let carried, false) = extra.addingReportingOverflow(total / 1_000_000_000) else { + return false + } + extra = carried + total %= 1_000_000_000 + } + guard case (let sum, false) = seconds.addingReportingOverflow(extra) else { + return false + } + // `Duration::new` carries a full second of nanoseconds. + if total == 1_000_000_000 { + guard case (let carried, false) = sum.addingReportingOverflow(1) else { + return false + } + seconds = carried + nanoseconds = 0 + } else { + seconds = sum + nanoseconds = total + } + return true + } + + private mutating func next() -> Unicode.Scalar? { + guard position < scalars.count else { + return nil + } + defer { position += 1 } + return scalars[position] + } + + /// Seconds per unit, or nanoseconds per unit for units below a second. + private static let units: [String: (seconds: UInt64, nanoseconds: UInt64)] = { + var units: [String: (UInt64, UInt64)] = [:] + for (names, seconds, nanoseconds) in [ + (["nanos", "nsec", "ns"], UInt64(0), UInt64(1)), + (["usec", "us", "\u{b5}s"], 0, 1_000), + (["millis", "msec", "ms"], 0, 1_000_000), + (["seconds", "second", "secs", "sec", "s"], 1, 0), + (["minutes", "minute", "min", "mins", "m"], 60, 0), + (["hours", "hour", "hr", "hrs", "h"], 3_600, 0), + (["days", "day", "d"], 86_400, 0), + (["weeks", "week", "wk", "wks", "w"], 604_800, 0), + (["months", "month", "M"], 2_630_016, 0), + (["years", "year", "yr", "yrs", "y"], 31_557_600, 0), + ] { + for name in names { + units[name] = (seconds, nanoseconds) + } + } + return units + }() + + private static func isUnitLetter(_ scalar: Unicode.Scalar) -> Bool { + ("a"..."z").contains(scalar) || ("A"..."Z").contains(scalar) || scalar == "\u{b5}" + } + + private static func digit(_ scalar: Unicode.Scalar) -> UInt64? { + ("0"..."9").contains(scalar) ? UInt64(scalar.value - 48) : nil + } + + private static func append(_ digit: UInt64, to value: UInt64) -> UInt64? { + multiply(value, 10).flatMap { scaled in + let (sum, overflow) = scaled.addingReportingOverflow(digit) + return overflow ? nil : sum + } + } + + private static func multiply(_ lhs: UInt64, _ rhs: UInt64) -> UInt64? { + let (product, overflow) = lhs.multipliedReportingOverflow(by: rhs) + return overflow ? nil : product + } + + private static func divideExactly(_ lhs: UInt64, _ rhs: UInt64) -> UInt64? { + lhs % rhs == 0 ? lhs / rhs : nil + } +} diff --git a/foreign/swift/Sources/Iggy/Client/IggyClient.swift b/foreign/swift/Sources/Iggy/Client/IggyClient.swift index 3353a6770d..6823a9ab24 100644 --- a/foreign/swift/Sources/Iggy/Client/IggyClient.swift +++ b/foreign/swift/Sources/Iggy/Client/IggyClient.swift @@ -80,6 +80,13 @@ public final class IggyClient: Sendable { self.init(configuration: ClientConfiguration(address: address, tls: tls, autoLogin: autoLogin)) } + /// Creates a client from a connection string such as + /// `iggy://iggy:iggy@localhost:8090?tls=true`, whose credentials become the + /// auto-login. See ``ClientConfiguration/init(connectionString:)``. + public convenience init(connectionString: String, logger: Logger = Logger(label: "org.apache.iggy")) throws { + self.init(configuration: try ClientConfiguration(connectionString: connectionString), logger: logger) + } + init(configuration: ClientConfiguration, connector: any IggyConnector, logger: Logger) { self.configuration = configuration self.logger = logger diff --git a/foreign/swift/Tests/IggyE2ETests/ClientE2ETests.swift b/foreign/swift/Tests/IggyE2ETests/ClientE2ETests.swift index 0c8b752e7a..343b6c04a3 100644 --- a/foreign/swift/Tests/IggyE2ETests/ClientE2ETests.swift +++ b/foreign/swift/Tests/IggyE2ETests/ClientE2ETests.swift @@ -67,6 +67,26 @@ struct SessionE2ETests { } } + @Test func aConnectionStringSignsInWithAPasswordOrAToken() async throws { + let address = try #require(E2EEnvironment.address) + var options = "reconnection_interval=200ms&reestablish_after=0" + if let tls = E2EEnvironment.tlsOptions, let domain = tls.domain, let caFile = tls.caFile { + options += "&tls=true&tls_domain=\(domain)&tls_ca_file=\(caFile)" + } + let client = try IggyClient(connectionString: "iggy://\(E2EEnvironment.rootUsername):\(E2EEnvironment.rootPassword)@\(address)?\(options)") + try await client.connect() + #expect(await client.state == .authenticated) + let name = E2EEnvironment.uniqueName("swift-cs") + let token = try await client.createPersonalAccessToken(name: name) + await client.shutdown() + + let tokenClient = try IggyClient(connectionString: "iggy+tcp://\(token.token)@\(address)?\(options)") + try await tokenClient.connect() + #expect(await tokenClient.state == .authenticated) + try await tokenClient.deletePersonalAccessToken(name: name) + await tokenClient.shutdown() + } + @Test func invalidCredentialsAreRejected() async throws { try await E2EEnvironment.withSession(login: false) { session in let client = session.client diff --git a/foreign/swift/Tests/IggyTests/ConnectionStringTests.swift b/foreign/swift/Tests/IggyTests/ConnectionStringTests.swift new file mode 100644 index 0000000000..f6f036442e --- /dev/null +++ b/foreign/swift/Tests/IggyTests/ConnectionStringTests.swift @@ -0,0 +1,238 @@ +// 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 Testing + +@testable import Iggy + +@Suite("Connection strings") +struct ConnectionStringTests { + @Test(arguments: ["127.0.0.1:8090", "localhost:8090", "iggy.default.svc.cluster.local:8090", "my_iggy_server:8090"]) + func acceptsHostnamesAndIPv4(_ address: String) throws { + let configuration = try ClientConfiguration(connectionString: "iggy://user:secret@\(address)") + #expect(configuration.address == address) + #expect(configuration.autoLogin == .usernamePassword(username: "user", password: "secret")) + #expect(configuration.tls == nil) + } + + @Test(arguments: ["[2001:db8::1]:8090", "[2001:db8:0:0:0:0:0:1]:65535", "[::ffff:192.0.2.1]:8090", "[::1]:8090"]) + func acceptsBracketedIPv6(_ address: String) throws { + let configuration = try ClientConfiguration(connectionString: "iggy://user:secret@\(address)?reconnection_retries=3&heartbeat_interval=10s") + #expect(configuration.address == address) + #expect(configuration.reconnection.maxRetries == 3) + #expect(configuration.heartbeatInterval == .seconds(10)) + _ = try Endpoint(parsing: configuration.address) + } + + @Test func aLoneSecretIsAPersonalAccessToken() throws { + let configuration = try ClientConfiguration(connectionString: "iggy+tcp://iggypat-1234567890abcdef@[::1]:8090") + #expect(configuration.autoLogin == .personalAccessToken("iggypat-1234567890abcdef")) + #expect(configuration.address == "[::1]:8090") + } + + @Test(arguments: ["", "?max_pending_operations=1"]) + func unsetOptionsTakeTheDefaults(_ options: String) throws { + let configuration = try ClientConfiguration(connectionString: "iggy://user:secret@127.0.0.1:1234\(options)") + #expect(configuration.reconnection == ReconnectionOptions(enabled: true, maxRetries: nil, interval: .seconds(1), reestablishAfter: .seconds(5))) + #expect(configuration.heartbeatInterval == .seconds(5)) + #expect(configuration.noDelay) + #expect(configuration.tls == nil) + } + + @Test func readsEveryOption() throws { + let configuration = try ClientConfiguration( + connectionString: + "iggy://user:secret@localhost:8090?tls=true&tls_domain=iggy.example&tls_ca_file=/etc/iggy/ca.pem&reconnection_retries=7" + + "&reconnection_interval=250ms&reestablish_after=1m 30s&heartbeat_interval=1.5s&nodelay=false" + + "&max_pending_operations=256&max_pending_operation_bytes=67108864") + #expect(configuration.tls == TLSOptions(domain: "iggy.example", caFile: "/etc/iggy/ca.pem", validateCertificate: true)) + #expect(configuration.reconnection == ReconnectionOptions(enabled: true, maxRetries: 7, interval: .milliseconds(250), reestablishAfter: .seconds(90))) + #expect(configuration.heartbeatInterval == .milliseconds(1500)) + #expect(!configuration.noDelay) + } + + @Test func flagsAreTrueOrFalse() throws { + for flag in ["tls", "nodelay"] { + for value in ["1", "yes", "TRUE", "True", "0", ""] { + #expect(throws: IggyError(.invalidConnectionString), "\(flag)=\(value)") { + try ClientConfiguration(connectionString: "iggy://user:secret@localhost:8090?\(flag)=\(value)") + } + } + } + // TLS options alone do not turn TLS on, as in every SDK. + let off = try ClientConfiguration(connectionString: "iggy://user:secret@localhost:8090?tls=false&tls_domain=localhost&nodelay=false") + #expect(off.tls == nil) + #expect(!off.noDelay) + #expect(try ClientConfiguration(connectionString: "iggy://user:secret@localhost:8090?tls_ca_file=/etc/iggy/ca.pem").tls == nil) + let on = try ClientConfiguration(connectionString: "iggy://user:secret@localhost:8090?nodelay=true&tls=true") + #expect(on.tls == TLSOptions()) + #expect(on.noDelay) + // The last spelling of an option wins. + let last = try ClientConfiguration(connectionString: "iggy://user:secret@localhost:8090?reconnection_retries=2&reconnection_retries=unlimited") + #expect(last.reconnection.maxRetries == nil) + } + + @Test func zeroIsAllowedOnlyWhereRustTakesAPlainDuration() throws { + let configuration = try ClientConfiguration(connectionString: "iggy://user:secret@localhost:8090?reestablish_after=disabled") + #expect(configuration.reconnection.reestablishAfter == .zero) + for option in ["heartbeat_interval", "reconnection_interval"] { + for value in ["0", "none", "0s", "999ns"] { + #expect(throws: IggyError(.invalidConnectionString), "\(option)=\(value)") { + try ClientConfiguration(connectionString: "iggy://user:secret@localhost:8090?\(option)=\(value)") + } + } + } + #expect(try ClientConfiguration(connectionString: "iggy://user:secret@localhost:8090?heartbeat_interval=1us").heartbeatInterval == .microseconds(1)) + } + + @Test func refusesDurationsTaskSleepCannotWaitOut() throws { + // u64 microseconds hold about 584,554 years. + #expect(try ClientConfiguration(connectionString: "iggy://user:secret@localhost:8090?heartbeat_interval=584000y").heartbeatInterval > .zero) + #expect(throws: IggyError(.invalidConnectionString)) { + try ClientConfiguration(connectionString: "iggy://user:secret@localhost:8090?heartbeat_interval=585000y") + } + } + + @Test func aBadRetryCountIsANumberError() { + for value in ["many", "-1", "4294967296", ""] { + #expect(throws: IggyError(.invalidNumberValue), "\(value)") { + try ClientConfiguration(connectionString: "iggy://user:secret@localhost:8090?reconnection_retries=\(value)") + } + } + } + + @Test(arguments: [ + "[::1]", "[::1]:", "[::1]:invalid", "[::1]:65536", "[::1]:0", "[::1]:-1", "[::1]:8090:1234", "[::1:8090", "::1:8090", "2001:db8::1:8090", + "[]:8090", "[localhost]:8090", "[gggg::1]:8090", "[192.168.1.1]:8090", "localhost", "localhost:0", "127.0.0.1:0", ":8090", "host:+8090", + "localhost:8090:1", + ]) + func refusesMalformedAddresses(_ address: String) { + #expect(throws: IggyError(.invalidConnectionString)) { + try ClientConfiguration(connectionString: "iggy://user:secret@\(address)") + } + } + + @Test(arguments: [ + "", "localhost:8090", "iggy:/user:secret@localhost:8090", "tcp://user:secret@localhost:8090", "iggy+quic://user:secret@localhost:8090", + "iggy+://user:secret@localhost:8090", "iggy://user:secret@localhost:8090://x", "iggy://@localhost:8090", "iggy://:secret@localhost:8090", + "iggy://user:@localhost:8090", "iggy://a:b:c@localhost:8090", "iggy://user:secret@localhost:8090?", "iggy://user:secret@localhost:8090?tls", + "iggy://user:secret@localhost:8090?tls=true=false", "iggy://user:secret@localhost:8090?unknown=1", "iggy://user:secret@localhost:8090?a=1?b=2", + "iggy://user:secret@localhost:8090?tls=true&", "iggy://user:secret@localhost:8090?heartbeat_interval=soon", + "iggy://user:secret@localhost:8090?reestablish_after=5", "iggy://us@er:secret@localhost:8090", + "iggy://user:secret@localhost:8090?max_pending_operations=0", "iggy://user:secret@localhost:8090?max_pending_operation_bytes=lots", + ]) + func refusesMalformedStrings(_ connectionString: String) { + #expect(throws: IggyError(.invalidConnectionString)) { + try ClientConfiguration(connectionString: connectionString) + } + } + + @Test func errorsNeverRepeatTheSecret() { + for connectionString in [ + "iggy+s3cr3t-token://user:s3cr3t-token@localhost:8090", "iggy://user:s3cr3t-token@localhost", "iggy://user:s3cr3t-token:x@localhost:8090", + "iggy://user:s3cr3t-token@@localhost:8090", + ] { + do { + _ = try ClientConfiguration(connectionString: connectionString) + Issue.record("\(connectionString) parsed") + } catch { + #expect(!"\(error)".contains("s3cr3t"), "\(error)") + } + } + #expect(throws: IggyError(.invalidConnectionString)) { try ClientConfiguration(connectionString: "iggy+quic://u:p@localhost:8080") } + do { + _ = try ClientConfiguration(connectionString: "iggy+quic://u:p@localhost:8080") + } catch { + #expect("\(error)".contains("quic")) + } + } + + @Test func aClientTakesItsConfigurationFromTheString() throws { + let client = try IggyClient(connectionString: "iggy://iggy:iggy@localhost:8090?heartbeat_interval=2s") + #expect(client.configuration.address == "localhost:8090") + #expect(client.configuration.autoLogin == .usernamePassword(username: "iggy", password: "iggy")) + #expect(client.configuration.heartbeatInterval == .seconds(2)) + } +} + +/// The cases of the humantime crate's own tests, plus Iggy's zero spellings. +@Suite("Iggy durations") +struct IggyDurationTests { + static func duration(_ seconds: Int64, _ nanoseconds: Int64 = 0) -> Duration { + .seconds(seconds) + .nanoseconds(nanoseconds) + } + + @Test func parsesEveryUnitSpelling() { + let cases: [(String, Duration)] = [ + ("17nsec", Self.duration(0, 17)), ("17nanos", Self.duration(0, 17)), ("33ns", Self.duration(0, 33)), + ("3usec", Self.duration(0, 3000)), ("78us", Self.duration(0, 78000)), ("163\u{b5}s", Self.duration(0, 163_000)), + ("31msec", Self.duration(0, 31_000_000)), ("31millis", Self.duration(0, 31_000_000)), ("6ms", Self.duration(0, 6_000_000)), + ("3000s", Self.duration(3000)), ("300sec", Self.duration(300)), ("300secs", Self.duration(300)), ("50seconds", Self.duration(50)), + ("1second", Self.duration(1)), ("100m", Self.duration(6000)), ("12min", Self.duration(720)), ("12mins", Self.duration(720)), + ("1minute", Self.duration(60)), ("7minutes", Self.duration(420)), ("2h", Self.duration(7200)), ("7hr", Self.duration(25200)), + ("7hrs", Self.duration(25200)), ("1hour", Self.duration(3600)), ("24hours", Self.duration(86400)), ("1day", Self.duration(86400)), + ("2days", Self.duration(172_800)), ("365d", Self.duration(31_536_000)), ("1week", Self.duration(604_800)), + ("7weeks", Self.duration(4_233_600)), ("104wks", Self.duration(2 * 31_449_600)), ("100wk", Self.duration(60_480_000)), + ("52w", Self.duration(31_449_600)), ("1month", Self.duration(2_630_016)), ("3months", Self.duration(3 * 2_630_016)), + ("1year", Self.duration(31_557_600)), ("7years", Self.duration(7 * 31_557_600)), ("15yrs", Self.duration(15 * 31_557_600)), + ("10yr", Self.duration(10 * 31_557_600)), ("17y", Self.duration(536_479_200)), + // Iggy lowercases first, so `M` reads as minutes. + ("12M", Self.duration(720)), ("5S", Self.duration(5)), + ] + for (text, expected) in cases { + #expect(IggyDuration.parse(text) == expected, "\(text)") + } + } + + @Test func parsesFractionsThatLandOnWholeUnits() { + let cases: [(String, Duration)] = [ + ("3.1us", Self.duration(0, 3100)), ("3.001us", Self.duration(0, 3001)), + ("31.000001ms", Self.duration(0, 31_000_001)), ("300.0s", Self.duration(300)), ("0.0s", .zero), ("1.123456789s", Self.duration(1, 123_456_789)), + ("12.1min", Self.duration(726)), ("1.5minutes", Self.duration(90)), ("2.5h", Self.duration(9000)), ("0.5h", Self.duration(1800)), + ("1.5day", Self.duration(129_600)), ("7.120us 3ns", Self.duration(0, 7123)), ("1.234s0.345ms0.678us0ns", Self.duration(1, 234_345_678)), + ("1.999999999s 0.000000001s", Self.duration(2)), ("999999999ns 1ns", Self.duration(1)), + ] + for (text, expected) in cases { + #expect(IggyDuration.parse(text) == expected, "\(text)") + } + } + + @Test func spacesMaySeparateEveryPart() { + #expect(IggyDuration.parse("1h 30m") == .seconds(5400)) + #expect(IggyDuration.parse("1h30m") == .seconds(5400)) + #expect(IggyDuration.parse("5 s") == .seconds(5)) + #expect(IggyDuration.parse("20 min 17 nsec ") == Self.duration(1200, 17)) + // humantime joins digits across a space. + #expect(IggyDuration.parse("1 0s") == .seconds(10)) + } + + @Test func zeroSpellings() { + for text in ["0", "unlimited", "Unlimited", "DISABLED", "none"] { + #expect(IggyDuration.parse(text) == .zero, "\(text)") + } + } + + @Test func refusesWhatRustRefuses() { + for text in [ + "", " ", "5", "s", ".", "1.", "1..s", ".1s", "1.s", "123", "10 months 1", "5x", "five seconds", "1.5", "-1s", "+1s", "1e3s", "1s,2s", + "0.000123456789s", "1.0000000002s", "31.0000001ms", "0.5ns", "0.0001h", "0.00000005d", "1\u{3bc}s", "100000000000000000000ns", + "18446744073709551616s", + ] { + #expect(IggyDuration.parse(text) == nil, "\(text)") + } + } +} diff --git a/scripts/run-bdd-tests.sh b/scripts/run-bdd-tests.sh index f0d8acbfd6..046f08efbf 100755 --- a/scripts/run-bdd-tests.sh +++ b/scripts/run-bdd-tests.sh @@ -140,11 +140,6 @@ run_suite(){ esac fi - if [ "$svc" = "swift-bdd" ] && [ "$FEATURE" != "deferred_polling" ] && [ "$FEATURE" != "all" ]; then - unsupported "$FEATURE" "$svc" || return 1 - return 0 - fi - if [ "$FEATURE" = "resource_options_patch" ] && [ "$svc" = "php-bdd" ]; then unsupported "$FEATURE" "$svc" || return 1 return 0 diff --git a/scripts/run-examples-from-readme.sh b/scripts/run-examples-from-readme.sh index 9e9ce521f3..fc928dda5d 100755 --- a/scripts/run-examples-from-readme.sh +++ b/scripts/run-examples-from-readme.sh @@ -21,7 +21,7 @@ set -euo pipefail # Unified script to run SDK examples from README.md files. # Usage: ./scripts/run-examples-from-readme.sh [OPTIONS] # -# --language LANG Language to test: rust|go|node|python|php|java|csharp (default: all) +# --language LANG Language to test: rust|go|node|python|php|java|csharp|swift (default: all) # --target TARGET Cargo target architecture for the server binary # --skip-tls Skip TLS example tests # @@ -257,6 +257,26 @@ run_go_examples() { SERVER_READY_PROBE="log" } +# shellcheck disable=SC2329 +run_swift_examples() { + # The Swift SDK speaks only the VSR wire protocol. + resolve_server_binary "${TARGET}" "iggy-server" + SERVER_READY_PROBE="tcp" + unset -f TRANSFORM_COMMAND 2>/dev/null || true + + run_language_examples \ + "Swift" \ + "examples/swift" \ + "README.md" \ + "^swift run" \ + "--tls" \ + "^swift run.*--tls" \ + 0 \ + "" + + SERVER_READY_PROBE="log" +} + # shellcheck disable=SC2329 run_python_examples() { # The Python SDK speaks only the VSR wire protocol, so examples run @@ -428,14 +448,15 @@ case "${LANGUAGE}" in php) run_one run_php_examples "PHP" ;; java) run_one run_java_examples "Java" ;; csharp) run_one run_csharp_examples "C#" ;; + swift) run_one run_swift_examples "Swift" ;; all) - for lang in rust node go python php java csharp; do + for lang in rust node go python php java csharp swift; do run_one "run_${lang}_examples" "${lang}" done ;; *) echo "Unknown language: ${LANGUAGE}" - echo "Supported: rust, node, go, python, php, java, csharp, all" + echo "Supported: rust, node, go, python, php, java, csharp, swift, all" exit 1 ;; esac