You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/09/29 19:36:48 UTC
(iggy) branch master updated: feat(swift): add the domain models and their wire encodings (#4335)
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 3a00b5ba1 feat(swift): add the domain models and their wire encodings (#4335)
3a00b5ba1 is described below
commit 3a00b5ba1ebcca9ccfaab532b406c9591a862dcb
Author: RustToMetal <th...@gmail.com>
AuthorDate: Tue Sep 29 14:36:42 2026 -0500
feat(swift): add the domain models and their wire encodings (#4335)
---
foreign/swift/README.md | 8 +-
foreign/swift/Sources/Iggy/Models/Identifier.swift | 118 ++++
.../swift/Sources/Iggy/Models/IggyMessage.swift | 223 ++++++++
foreign/swift/Sources/Iggy/Models/Messaging.swift | 241 ++++++++
.../swift/Sources/Iggy/Models/Permissions.swift | 186 ++++++
.../Sources/Iggy/Models/ResourceOptions.swift | 168 ++++++
foreign/swift/Sources/Iggy/Models/Resources.swift | 628 +++++++++++++++++++++
.../swift/Sources/Iggy/Models/TopicOptions.swift | 206 +++++++
.../swift/Sources/Iggy/Models/UserHeaders.swift | 427 ++++++++++++++
foreign/swift/Sources/Iggy/Wire/ByteCodec.swift | 7 +
foreign/swift/Tests/IggyTests/Fixtures/golden.json | 2 +
foreign/swift/Tests/IggyTests/ModelTests.swift | 379 +++++++++++++
.../swift/Tests/IggyTests/ModelVectorTests.swift | 116 ++++
.../Tests/IggyTests/Support/GoldenFixture.swift | 32 ++
foreign/swift/Tools/golden-vectors/src/main.rs | 7 +-
15 files changed, 2744 insertions(+), 4 deletions(-)
diff --git a/foreign/swift/README.md b/foreign/swift/README.md
index f6657e0e2..c7333530b 100644
--- a/foreign/swift/README.md
+++ b/foreign/swift/README.md
@@ -11,9 +11,11 @@
Official Swift client SDK for [Apache Iggy](https://iggy.apache.org) message streaming.
The SDK is built up in stages. This stage carries the package scaffold, the error table,
-the byte codec the wire protocol is encoded with, and native XXH3-64 and XXH32 hashes
-pinned to the Rust crates by golden vectors. The protocol layer, the TCP and TLS client,
-the producer and consumer, examples, and BDD scenarios follow in later changes.
+the byte codec the wire protocol is encoded with, native XXH3-64 and XXH32 hashes pinned
+to the Rust crates by golden vectors, and the domain models (streams, topics, messages,
+user headers, options, permissions) with their wire encodings. The protocol layer, the
+TCP and TLS client, the producer and consumer, examples, and BDD scenarios follow in
+later changes.
## Requirements
diff --git a/foreign/swift/Sources/Iggy/Models/Identifier.swift b/foreign/swift/Sources/Iggy/Models/Identifier.swift
new file mode 100644
index 000000000..1d816bb57
--- /dev/null
+++ b/foreign/swift/Sources/Iggy/Models/Identifier.swift
@@ -0,0 +1,118 @@
+// 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.
+
+/// Names a stream, topic, user, or consumer group either by its numeric id
+/// or by its name.
+///
+/// Both literals work wherever an identifier is expected:
+///
+/// ```swift
+/// try await client.getTopic(streamID: "orders", topicID: 1)
+/// ```
+///
+/// A name is 1 to 255 bytes of UTF-8. The string-literal form traps on an
+/// invalid name; use ``init(named:)`` to validate a runtime value.
+public struct Identifier: Sendable, Hashable, CustomStringConvertible {
+ public enum Kind: Sendable, Hashable {
+ case numeric(UInt32)
+ case named(String)
+ }
+
+ public static let maxNameLength = 255
+
+ public let kind: Kind
+
+ public init(numeric id: UInt32) {
+ kind = .numeric(id)
+ }
+
+ /// Validates the name length before accepting it.
+ public init(named name: String) throws {
+ let length = name.utf8.count
+ guard length >= 1, length <= Self.maxNameLength else {
+ throw IggyError(.invalidIdentifier, context: "name must be 1-\(Self.maxNameLength) bytes, got \(length)")
+ }
+ kind = .named(name)
+ }
+
+ /// Parses text the way the CLI does: a decimal number becomes a numeric
+ /// identifier, anything else a name.
+ public init(parsing text: String) throws {
+ if let number = UInt32(text) {
+ self.init(numeric: number)
+ } else {
+ try self.init(named: text)
+ }
+ }
+
+ public var numericValue: UInt32? {
+ if case .numeric(let id) = kind { id } else { nil }
+ }
+
+ public var name: String? {
+ if case .named(let name) = kind { name } else { nil }
+ }
+
+ public var description: String {
+ switch kind {
+ case .numeric(let id): String(id)
+ case .named(let name): name
+ }
+ }
+}
+
+extension Identifier: ExpressibleByIntegerLiteral {
+ public init(integerLiteral value: UInt32) {
+ self.init(numeric: value)
+ }
+}
+
+extension Identifier: ExpressibleByStringLiteral {
+ public init(stringLiteral value: String) {
+ do {
+ try self.init(named: value)
+ } catch {
+ preconditionFailure("invalid identifier literal: \(error)")
+ }
+ }
+}
+
+extension Identifier {
+ private static let numericKindCode: UInt8 = 1
+ private static let stringKindCode: UInt8 = 2
+
+ /// `[kind:1][length:1][value:N]`.
+ func encode(into writer: inout ByteWriter) throws {
+ switch kind {
+ case .numeric(let id):
+ writer.write(Self.numericKindCode)
+ writer.write(UInt8(4))
+ writer.write(id)
+ case .named(let name):
+ writer.write(Self.stringKindCode)
+ try writer.writeName(name)
+ }
+ }
+
+ var encodedSize: Int {
+ switch kind {
+ case .numeric: 6
+ case .named(let name): 2 + name.utf8.count
+ }
+ }
+
+}
diff --git a/foreign/swift/Sources/Iggy/Models/IggyMessage.swift b/foreign/swift/Sources/Iggy/Models/IggyMessage.swift
new file mode 100644
index 000000000..2d6060d1a
--- /dev/null
+++ b/foreign/swift/Sources/Iggy/Models/IggyMessage.swift
@@ -0,0 +1,223 @@
+// 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
+
+/// A message: what a producer sends and what a consumer receives.
+///
+/// Create one from a payload and optional user headers:
+///
+/// ```swift
+/// let message = try IggyMessage("hello", userHeaders: ["trace-id": "abc"])
+/// let event = try IggyMessage(encoding: order) // any Encodable, as JSON
+/// ```
+///
+/// The server stamps `offset`, `timestamp`, and `checksum` when the batch is
+/// persisted; they are zero on a message that has not been sent yet. The id
+/// is minted by the producer when left zero.
+public struct IggyMessage: Sendable, Hashable {
+ /// Largest payload the server accepts.
+ public static let maxPayloadSize = 64 * 1000 * 1000
+ /// Largest encoded user-headers block the server accepts.
+ public static let maxUserHeadersSize = 100 * 1000
+
+ /// Identifier, unique within the partition. Zero until the producer mints one.
+ public var id: MessageID
+ /// Offset within the partition, assigned by the server.
+ public var offset: UInt64
+ /// When the server persisted the message.
+ public var timestamp: IggyTimestamp
+ /// When the producer created the message.
+ public var originTimestamp: IggyTimestamp
+ /// Checksum the server verified, zero before the message is sent.
+ public var checksum: UInt64
+ /// The payload bytes, never empty.
+ public var payload: [UInt8]
+ /// The user headers exactly as they travel on the wire, when any. This
+ /// is what the send path reads; ``userHeaders`` is a view over it.
+ public private(set) var rawUserHeaders: [UInt8]?
+
+ /// The user headers decoded on access, or nil when there are none or a
+ /// header this SDK cannot interpret leaves them undecodable; the raw
+ /// bytes stay available either way.
+ public var userHeaders: UserHeaders? {
+ guard let rawUserHeaders else {
+ return nil
+ }
+ return try? HeaderTLV.decode(rawUserHeaders[...])
+ }
+
+ /// Creates a message to send.
+ public init(payload: [UInt8], id: MessageID = .zero, userHeaders: UserHeaders? = nil) throws {
+ if payload.isEmpty {
+ throw IggyError(.invalidMessagePayloadLength, context: "payload cannot be empty")
+ }
+ if payload.count > Self.maxPayloadSize {
+ throw IggyError(.tooBigMessagePayload)
+ }
+ var raw: [UInt8]?
+ if let userHeaders, !userHeaders.isEmpty {
+ let encoded = HeaderTLV.encode(userHeaders)
+ if encoded.count > Self.maxUserHeadersSize {
+ throw IggyError(.tooBigUserHeaders)
+ }
+ raw = encoded
+ }
+ self.id = id
+ self.offset = 0
+ self.timestamp = .zero
+ self.originTimestamp = .now
+ self.checksum = 0
+ self.payload = payload
+ self.rawUserHeaders = raw
+ }
+
+ /// Creates a message whose payload is the UTF-8 encoding of `text`.
+ public init(_ text: String, id: MessageID = .zero, userHeaders: UserHeaders? = nil) throws {
+ try self.init(payload: Array(text.utf8), id: id, userHeaders: userHeaders)
+ }
+
+ /// Creates a message whose payload is `value` encoded as JSON.
+ public init<T: Encodable>(encoding value: T, encoder: JSONEncoder = JSONEncoder(), id: MessageID = .zero, userHeaders: UserHeaders? = nil) throws {
+ try self.init(payload: Array(try encoder.encode(value)), id: id, userHeaders: userHeaders)
+ }
+
+ /// Builds a received message from its decoded frame.
+ init(id: MessageID, offset: UInt64, timestamp: IggyTimestamp, originTimestamp: IggyTimestamp, checksum: UInt64, payload: [UInt8], rawUserHeaders: [UInt8]?)
+ {
+ self.id = id
+ self.offset = offset
+ self.timestamp = timestamp
+ self.originTimestamp = originTimestamp
+ self.checksum = checksum
+ self.payload = payload
+ self.rawUserHeaders = rawUserHeaders.flatMap { $0.isEmpty ? nil : $0 }
+ }
+
+ /// The payload interpreted as UTF-8, replacing invalid sequences.
+ public var payloadString: String {
+ String(decoding: payload, as: UTF8.self)
+ }
+
+ /// The payload decoded as JSON.
+ public func decode<T: Decodable>(_ type: T.Type = T.self, decoder: JSONDecoder = JSONDecoder()) throws -> T {
+ try decoder.decode(type, from: Data(payload))
+ }
+
+ /// One user header by its string key.
+ public subscript(header key: String) -> HeaderValue? {
+ guard let headerKey = try? HeaderKey(key) else {
+ return nil
+ }
+ return userHeaders?[headerKey]
+ }
+}
+
+extension IggyMessage: ExpressibleByStringLiteral {
+ /// A string literal must be a non-empty payload within
+ /// ``maxPayloadSize``; the literal form traps otherwise. Use
+ /// ``init(_:id:userHeaders:)`` to validate a runtime value.
+ public init(stringLiteral value: String) {
+ do {
+ try self.init(value)
+ } catch {
+ preconditionFailure("invalid message literal: \(error)")
+ }
+ }
+}
+
+extension IggyMessage: CustomStringConvertible {
+ private static let previewLength = 47
+
+ /// Shows a bounded prefix of the payload, so describing a large message
+ /// costs no more than a small one.
+ public var description: String {
+ if payload.count <= Self.previewLength + 3, let text = String(bytes: payload, encoding: .utf8) {
+ return "[\(offset)] ID:\(id) '\(text)'"
+ }
+ if let text = String(bytes: Self.utf8Prefix(payload, Self.previewLength), encoding: .utf8) {
+ return "[\(offset)] ID:\(id) '\(text)... (\(payload.count)B)'"
+ }
+ return "[\(offset)] ID:\(id) <binary \(payload.count)B>"
+ }
+
+ /// The longest prefix of at most `count` bytes that does not end inside
+ /// a multi-byte UTF-8 sequence.
+ private static func utf8Prefix(_ bytes: [UInt8], _ count: Int) -> ArraySlice<UInt8> {
+ var end = min(count, bytes.count)
+ // Step back over continuation bytes, then over a lead byte whose
+ // sequence the cut left incomplete.
+ while end > 0, bytes[end - 1] & 0xC0 == 0x80 {
+ end -= 1
+ }
+ if end > 0, bytes[end - 1] >= 0xC0 {
+ end -= 1
+ }
+ return bytes[..<end]
+ }
+}
+
+/// The messages one poll returned.
+public struct PolledMessages: Sendable, Hashable {
+ /// The partition that was read. Empty consumer-group polls can carry a
+ /// sentinel instead of a real id, see ``noAssignedPartition``.
+ public var partitionID: UInt32
+ /// The newest offset the partition had when the batch was read.
+ public var currentOffset: UInt64
+ /// Number of messages the server reported.
+ public var count: UInt32
+ public var messages: [IggyMessage]
+
+ public init(partitionID: UInt32, currentOffset: UInt64, count: UInt32, messages: [IggyMessage]) {
+ self.partitionID = partitionID
+ self.currentOffset = currentOffset
+ self.count = count
+ self.messages = messages
+ }
+
+ public static let empty = PolledMessages(partitionID: 0, currentOffset: 0, count: 0, messages: [])
+
+ /// Partition id of an empty poll by a consumer-group member that currently
+ /// holds no partitions: `NO_ASSIGNED_PARTITION` in `core/common`.
+ public static let noAssignedPartition: UInt32 = UInt32.max - 1
+}
+
+/// One confirmation of a committed batch.
+public struct SendConfirmation: Sendable, Hashable {
+ public var streamID: UInt32
+ public var topicID: UInt32
+ public var partitionID: UInt32
+ /// Offset the first message of the batch landed at.
+ public var baseOffset: UInt64
+
+ public init(streamID: UInt32, topicID: UInt32, partitionID: UInt32, baseOffset: UInt64) {
+ self.streamID = streamID
+ self.topicID = topicID
+ self.partitionID = partitionID
+ self.baseOffset = baseOffset
+ }
+}
+
+/// Per-partition commit confirmations of a send. The list may be empty: the
+/// server can commit a batch it has no offsets to describe.
+public struct SendMessagesResponse: Sendable, Hashable {
+ public var confirmations: [SendConfirmation]
+
+ public init(confirmations: [SendConfirmation] = []) {
+ self.confirmations = confirmations
+ }
+}
diff --git a/foreign/swift/Sources/Iggy/Models/Messaging.swift b/foreign/swift/Sources/Iggy/Models/Messaging.swift
new file mode 100644
index 000000000..a7f4e66d2
--- /dev/null
+++ b/foreign/swift/Sources/Iggy/Models/Messaging.swift
@@ -0,0 +1,241 @@
+// 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
+
+/// Who is polling: a standalone consumer or a member of a consumer group.
+public struct Consumer: Sendable, Hashable, CustomStringConvertible {
+ public enum Kind: UInt8, Sendable, Hashable {
+ case consumer = 1
+ case consumerGroup = 2
+ }
+
+ public var kind: Kind
+ public var id: Identifier
+
+ public init(kind: Kind, id: Identifier) {
+ self.kind = kind
+ self.id = id
+ }
+
+ /// A standalone consumer.
+ public static func consumer(_ id: Identifier) -> Consumer {
+ Consumer(kind: .consumer, id: id)
+ }
+
+ /// A member of a consumer group.
+ public static func group(_ id: Identifier) -> Consumer {
+ Consumer(kind: .consumerGroup, id: id)
+ }
+
+ /// The consumer every other SDK and the CLI use when none is named:
+ /// consumer id 0, so offsets are shared with them.
+ public static let `default` = Consumer(kind: .consumer, id: Identifier(numeric: 0))
+
+ public var description: String {
+ switch kind {
+ case .consumer: "consumer|\(id)"
+ case .consumerGroup: "consumer_group|\(id)"
+ }
+ }
+
+ func encode(into writer: inout ByteWriter) throws {
+ writer.write(kind.rawValue)
+ try id.encode(into: &writer)
+ }
+}
+
+/// How a batch of messages picks its partition.
+///
+/// Under the consensus protocol the server only routes explicit partitions,
+/// so ``balanced`` and ``messagesKey(_:)`` are resolved on the client before
+/// the batch is sent: balanced round-robins over the topic's partitions and a
+/// key is hashed with XXH32, the same way every other SDK does it.
+public enum Partitioning: Sendable, Hashable {
+ /// Round-robin over the topic's partitions.
+ case balanced
+ /// A specific partition.
+ case partition(UInt32)
+ /// Hash of the key selects the partition, so one key always lands on the
+ /// same partition. The key is 1 to 255 bytes.
+ case messagesKey([UInt8])
+
+ public static func messagesKey(_ key: String) -> Partitioning {
+ .messagesKey(Array(key.utf8))
+ }
+
+ /// The width is part of the label because it is part of the key: the
+ /// same number as a `uint64` lands on a different partition.
+ public static func messagesKey(uint32 key: UInt32) -> Partitioning {
+ .messagesKey(key.littleEndianBytes)
+ }
+
+ public static func messagesKey(uint64 key: UInt64) -> Partitioning {
+ .messagesKey(key.littleEndianBytes)
+ }
+
+ public static func messagesKey(uint128 key: UInt128Value) -> Partitioning {
+ .messagesKey(key.littleEndianBytes)
+ }
+
+ private static let balancedCode: UInt8 = 1
+ private static let partitionCode: UInt8 = 2
+ private static let messagesKeyCode: UInt8 = 3
+
+ /// A messages key outside 1...255 bytes cannot be framed. The client
+ /// checks this before it resolves the key to a partition, and `encode`
+ /// checks it again before writing the length byte.
+ func validate() throws {
+ if case .messagesKey(let key) = self, key.isEmpty || key.count > Int(UInt8.max) {
+ throw IggyError(.invalidKeyValueLength, context: "messages key must be 1-255 bytes, got \(key.count)")
+ }
+ }
+
+ /// `[kind:1][length:1][value]`.
+ func encode(into writer: inout ByteWriter) throws {
+ try validate()
+ switch self {
+ case .balanced:
+ writer.write(Self.balancedCode)
+ writer.write(UInt8(0))
+ case .partition(let id):
+ writer.write(Self.partitionCode)
+ writer.write(UInt8(4))
+ writer.write(id)
+ case .messagesKey(let key):
+ writer.write(Self.messagesKeyCode)
+ writer.write(UInt8(key.count))
+ writer.write(key)
+ }
+ }
+
+ var encodedSize: Int {
+ switch self {
+ case .balanced: 2
+ case .partition: 6
+ case .messagesKey(let key): 2 + key.count
+ }
+ }
+}
+
+/// Where a poll starts reading a partition.
+public struct PollingStrategy: Sendable, Hashable, CustomStringConvertible {
+ public enum Kind: UInt8, Sendable, Hashable {
+ case offset = 1
+ case timestamp = 2
+ case first = 3
+ case last = 4
+ case next = 5
+ }
+
+ public var kind: Kind
+ public var value: UInt64
+
+ public init(kind: Kind, value: UInt64) {
+ self.kind = kind
+ self.value = value
+ }
+
+ /// Resume from a specific offset.
+ public static func offset(_ offset: UInt64) -> PollingStrategy {
+ PollingStrategy(kind: .offset, value: offset)
+ }
+
+ /// Resume from the first message at or after a timestamp.
+ public static func timestamp(_ timestamp: IggyTimestamp) -> PollingStrategy {
+ PollingStrategy(kind: .timestamp, value: timestamp.microseconds)
+ }
+
+ /// Start from the beginning of the partition.
+ public static let first = PollingStrategy(kind: .first, value: 0)
+ /// Start from the end of the partition.
+ public static let last = PollingStrategy(kind: .last, value: 0)
+ /// Continue after the consumer's stored offset.
+ public static let next = PollingStrategy(kind: .next, value: 0)
+
+ public var description: String {
+ switch kind {
+ case .offset: "offset|\(value)"
+ case .timestamp: "timestamp|\(value)"
+ case .first: "first|0"
+ case .last: "last|0"
+ case .next: "next|0"
+ }
+ }
+
+ /// `[kind:1][value:8]`.
+ func encode(into writer: inout ByteWriter) {
+ writer.write(kind.rawValue)
+ writer.write(value)
+ }
+}
+
+/// Acknowledgement policy for consumer-offset write commands.
+public enum AckLevel: UInt8, Sendable, Hashable {
+ /// Local fast path for a single-replica partition. Replicated partitions
+ /// commit offset writes through VSR before replying, including when this
+ /// acknowledgement value is selected. On a single replica, a
+ /// directory-sync failure can be reported after the mutation became
+ /// visible; its crash durability is then unknown, and retrying a deletion
+ /// that already took effect can return `consumerOffsetNotFound`.
+ case noAck = 0
+ /// Submit through the partition VSR consensus pipeline and respond only
+ /// after the write has been committed by a quorum of replicas. The
+ /// default for explicit client writes.
+ case quorum = 1
+}
+
+/// Microseconds since the Unix epoch, the timestamp unit of the whole
+/// platform.
+public struct IggyTimestamp: Sendable, Hashable, Comparable, CustomStringConvertible {
+ public var microseconds: UInt64
+
+ public init(microseconds: UInt64) {
+ self.microseconds = microseconds
+ }
+
+ public static var now: IggyTimestamp {
+ IggyTimestamp(date: .init())
+ }
+
+ public static let zero = IggyTimestamp(microseconds: 0)
+
+ /// Dates before the epoch clamp to zero and dates past the `UInt64`
+ /// microsecond range clamp to `UInt64.max` instead of trapping.
+ public init(date: Date) {
+ let interval = date.timeIntervalSince1970 * 1_000_000
+ if !(interval > 0) {
+ microseconds = 0
+ } else if interval >= 0x1p64 {
+ microseconds = UInt64.max
+ } else {
+ microseconds = UInt64(interval)
+ }
+ }
+
+ public var date: Date {
+ Date(timeIntervalSince1970: Double(microseconds) / 1_000_000)
+ }
+
+ public static func < (lhs: IggyTimestamp, rhs: IggyTimestamp) -> Bool {
+ lhs.microseconds < rhs.microseconds
+ }
+
+ public var description: String {
+ String(microseconds)
+ }
+}
diff --git a/foreign/swift/Sources/Iggy/Models/Permissions.swift b/foreign/swift/Sources/Iggy/Models/Permissions.swift
new file mode 100644
index 000000000..4232b62bd
--- /dev/null
+++ b/foreign/swift/Sources/Iggy/Models/Permissions.swift
@@ -0,0 +1,186 @@
+// 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.
+
+/// Server-wide permissions of a user.
+public struct GlobalPermissions: Sendable, Hashable {
+ public var manageServers = false
+ public var readServers = false
+ public var manageUsers = false
+ public var readUsers = false
+ public var manageStreams = false
+ public var readStreams = false
+ public var manageTopics = false
+ public var readTopics = false
+ public var pollMessages = false
+ public var sendMessages = false
+
+ public init(
+ manageServers: Bool = false, readServers: Bool = false, manageUsers: Bool = false, readUsers: Bool = false,
+ manageStreams: Bool = false, readStreams: Bool = false, manageTopics: Bool = false, readTopics: Bool = false,
+ pollMessages: Bool = false, sendMessages: Bool = false
+ ) {
+ self.manageServers = manageServers
+ self.readServers = readServers
+ self.manageUsers = manageUsers
+ self.readUsers = readUsers
+ self.manageStreams = manageStreams
+ self.readStreams = readStreams
+ self.manageTopics = manageTopics
+ self.readTopics = readTopics
+ self.pollMessages = pollMessages
+ self.sendMessages = sendMessages
+ }
+
+ /// Every permission granted.
+ public static let all = GlobalPermissions(
+ manageServers: true, readServers: true, manageUsers: true, readUsers: true, manageStreams: true,
+ readStreams: true, manageTopics: true, readTopics: true, pollMessages: true, sendMessages: true)
+}
+
+/// Permissions of a user on one topic.
+public struct TopicPermissions: Sendable, Hashable {
+ public var manageTopic = false
+ public var readTopic = false
+ public var pollMessages = false
+ public var sendMessages = false
+
+ public init(manageTopic: Bool = false, readTopic: Bool = false, pollMessages: Bool = false, sendMessages: Bool = false) {
+ self.manageTopic = manageTopic
+ self.readTopic = readTopic
+ self.pollMessages = pollMessages
+ self.sendMessages = sendMessages
+ }
+}
+
+/// Permissions of a user on one stream and, optionally, on its topics.
+public struct StreamPermissions: Sendable, Hashable {
+ public var manageStream = false
+ public var readStream = false
+ public var manageTopics = false
+ public var readTopics = false
+ public var pollMessages = false
+ public var sendMessages = false
+ public var topics: [UInt32: TopicPermissions] = [:]
+
+ public init(
+ manageStream: Bool = false, readStream: Bool = false, manageTopics: Bool = false, readTopics: Bool = false,
+ pollMessages: Bool = false, sendMessages: Bool = false, topics: [UInt32: TopicPermissions] = [:]
+ ) {
+ self.manageStream = manageStream
+ self.readStream = readStream
+ self.manageTopics = manageTopics
+ self.readTopics = readTopics
+ self.pollMessages = pollMessages
+ self.sendMessages = sendMessages
+ self.topics = topics
+ }
+}
+
+/// The full permission set of a user.
+public struct Permissions: Sendable, Hashable {
+ public var global: GlobalPermissions
+ public var streams: [UInt32: StreamPermissions]
+
+ public init(global: GlobalPermissions = GlobalPermissions(), streams: [UInt32: StreamPermissions] = [:]) {
+ self.global = global
+ self.streams = streams
+ }
+}
+
+extension Permissions {
+ private static let hasNext: UInt8 = 1
+ private static let noNext: UInt8 = 0
+
+ /// Streams and topics are written sorted by id, the deterministic layout
+ /// the Rust SDK produces.
+ func encode() -> [UInt8] {
+ var writer = ByteWriter()
+ writer.write(global.manageServers)
+ writer.write(global.readServers)
+ writer.write(global.manageUsers)
+ writer.write(global.readUsers)
+ writer.write(global.manageStreams)
+ writer.write(global.readStreams)
+ writer.write(global.manageTopics)
+ writer.write(global.readTopics)
+ writer.write(global.pollMessages)
+ writer.write(global.sendMessages)
+ let streams = self.streams.sorted { $0.key < $1.key }
+ writer.write(streams.isEmpty ? Self.noNext : Self.hasNext)
+ for (index, (streamID, stream)) in streams.enumerated() {
+ writer.write(streamID)
+ writer.write(stream.manageStream)
+ writer.write(stream.readStream)
+ writer.write(stream.manageTopics)
+ writer.write(stream.readTopics)
+ writer.write(stream.pollMessages)
+ writer.write(stream.sendMessages)
+ let topics = stream.topics.sorted { $0.key < $1.key }
+ writer.write(topics.isEmpty ? Self.noNext : Self.hasNext)
+ for (topicIndex, (topicID, topic)) in topics.enumerated() {
+ writer.write(topicID)
+ writer.write(topic.manageTopic)
+ writer.write(topic.readTopic)
+ writer.write(topic.pollMessages)
+ writer.write(topic.sendMessages)
+ writer.write(topicIndex == topics.count - 1 ? Self.noNext : Self.hasNext)
+ }
+ writer.write(index == streams.count - 1 ? Self.noNext : Self.hasNext)
+ }
+ return writer.bytes
+ }
+
+ static func decode(from reader: inout ByteReader) throws -> Permissions {
+ var global = GlobalPermissions()
+ global.manageServers = try reader.readBool()
+ global.readServers = try reader.readBool()
+ global.manageUsers = try reader.readBool()
+ global.readUsers = try reader.readBool()
+ global.manageStreams = try reader.readBool()
+ global.readStreams = try reader.readBool()
+ global.manageTopics = try reader.readBool()
+ global.readTopics = try reader.readBool()
+ global.pollMessages = try reader.readBool()
+ global.sendMessages = try reader.readBool()
+ var streams: [UInt32: StreamPermissions] = [:]
+ if try reader.readBool() {
+ repeat {
+ let streamID = try reader.readUInt32()
+ var stream = StreamPermissions()
+ stream.manageStream = try reader.readBool()
+ stream.readStream = try reader.readBool()
+ stream.manageTopics = try reader.readBool()
+ stream.readTopics = try reader.readBool()
+ stream.pollMessages = try reader.readBool()
+ stream.sendMessages = try reader.readBool()
+ if try reader.readBool() {
+ repeat {
+ let topicID = try reader.readUInt32()
+ var topic = TopicPermissions()
+ topic.manageTopic = try reader.readBool()
+ topic.readTopic = try reader.readBool()
+ topic.pollMessages = try reader.readBool()
+ topic.sendMessages = try reader.readBool()
+ stream.topics[topicID] = topic
+ } while try reader.readBool()
+ }
+ streams[streamID] = stream
+ } while try reader.readBool()
+ }
+ return Permissions(global: global, streams: streams)
+ }
+}
diff --git a/foreign/swift/Sources/Iggy/Models/ResourceOptions.swift b/foreign/swift/Sources/Iggy/Models/ResourceOptions.swift
new file mode 100644
index 000000000..9d133b3fb
--- /dev/null
+++ b/foreign/swift/Sources/Iggy/Models/ResourceOptions.swift
@@ -0,0 +1,168 @@
+// 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 option of a stream, topic, or user as reported by the server, with
+/// whether the client set it explicitly or admission derived it.
+public struct OptionValue: Sendable, Hashable {
+ public var value: HeaderValue
+ public var explicit: Bool
+
+ public init(value: HeaderValue, explicit: Bool) {
+ self.value = value
+ self.explicit = explicit
+ }
+
+ public static func explicit(_ value: HeaderValue) -> OptionValue {
+ OptionValue(value: value, explicit: true)
+ }
+
+ public static func derived(_ value: HeaderValue) -> OptionValue {
+ OptionValue(value: value, explicit: false)
+ }
+}
+
+/// Options of a resource keyed by option name.
+public typealias ResourceOptions = [String: OptionValue]
+
+/// The option catalog scope a ``IggyClient/describeOptions(scope:)`` call asks
+/// about.
+public enum OptionsScope: UInt8, Sendable, Hashable {
+ case topic = 1
+ case stream = 2
+ case user = 3
+}
+
+/// One entry of the option catalog: a key the create command accepts, its
+/// canonical kind, and the server's current default.
+public struct OptionSpec: Sendable, Hashable {
+ public var key: String
+ public var kind: HeaderKind
+ public var defaultValue: [UInt8]
+ public var description: String
+
+ public init(key: String, kind: HeaderKind, defaultValue: [UInt8], description: String) {
+ self.key = key
+ self.kind = kind
+ self.defaultValue = defaultValue
+ self.description = description
+ }
+}
+
+/// Key-value options block for create and update requests: the user-headers
+/// TLV with the stricter contract options need. Mirrors
+/// `core/binary_protocol/src/primitives/options.rs`.
+enum OptionsBlock {
+ static let maxOptions = 1024
+ static let maxOptionsBytes = 100 * 1000
+
+ /// Encodes the explicit entries in key order; derived values are the
+ /// server's to compute and are never sent back.
+ static func encode(_ options: ResourceOptions) throws -> [UInt8] {
+ let entries = options.filter { $0.value.explicit }.sorted { $0.key.utf8.lexicographicallyPrecedes($1.key.utf8) }
+ if entries.isEmpty {
+ return []
+ }
+ if entries.count > maxOptions {
+ throw IggyError(.optionsBlockTooLarge, context: "\(entries.count) entries, maximum \(maxOptions)")
+ }
+ let size = entries.reduce(0) { $0 + HeaderTLV.entryOverhead + $1.key.utf8.count + $1.value.value.bytes.count }
+ if size > maxOptionsBytes {
+ throw IggyError(.optionsBlockTooLarge, context: "\(size) bytes, maximum \(maxOptionsBytes)")
+ }
+ var writer = ByteWriter(capacity: size)
+ for (key, option) in entries {
+ let keyBytes = Array(key.utf8)
+ guard !keyBytes.isEmpty, keyBytes.count <= HeaderTLV.maxFieldLength else {
+ throw IggyError(.unsupportedOptionKey, context: "option key must be 1-255 bytes")
+ }
+ writer.write(HeaderKind.string.rawValue)
+ writer.write(UInt32(keyBytes.count))
+ writer.write(keyBytes)
+ writer.write(option.value.kind.rawValue)
+ writer.write(UInt32(option.value.bytes.count))
+ writer.write(option.value.bytes)
+ }
+ return writer.bytes
+ }
+
+ /// Structural validation plus the options contract: string keys, no
+ /// duplicates, bounded count and size. Returns the entries.
+ static func validate(_ bytes: ArraySlice<UInt8>) throws -> [HeaderTLV.Entry] {
+ if bytes.count > maxOptionsBytes {
+ throw WireError.validation("options block is \(bytes.count) bytes, exceeds maximum \(maxOptionsBytes)")
+ }
+ let entries = try HeaderTLV.validate(bytes)
+ if entries.isEmpty {
+ return entries
+ }
+ if entries.count > maxOptions {
+ throw WireError.validation("options block has \(entries.count) entries, exceeds maximum \(maxOptions)")
+ }
+ var keys = Set<[UInt8]>()
+ for entry in entries {
+ guard entry.keyKind == HeaderKind.string.rawValue else {
+ throw WireError.validation("option key kind \(entry.keyKind) is not a string")
+ }
+ guard String(bytes: entry.key, encoding: .utf8) != nil else {
+ throw WireError.validation("option key is not valid UTF-8")
+ }
+ guard keys.insert(Array(entry.key)).inserted else {
+ throw WireError.validation("duplicate option key: \(String(decoding: entry.key, as: UTF8.self))")
+ }
+ }
+ return entries
+ }
+
+ /// Decodes a block into domain options, marking every entry with the
+ /// given provenance. Entries with an unknown value kind are skipped so a
+ /// resource written by a newer server can still be read back.
+ static func decode(_ bytes: ArraySlice<UInt8>, explicit: Bool) throws -> ResourceOptions {
+ var options: ResourceOptions = [:]
+ for entry in try validate(bytes) {
+ guard let kind = HeaderKind(rawValue: entry.valueKind) else {
+ continue
+ }
+ if let expected = kind.expectedSize, entry.value.count != expected {
+ throw IggyError(.invalidHeaderValue)
+ }
+ let key = String(decoding: entry.key, as: UTF8.self)
+ options[key] = OptionValue(value: HeaderValue(uncheckedKind: kind, bytes: Array(entry.value)), explicit: explicit)
+ }
+ return options
+ }
+
+ /// Merges the `(explicit, derived)` blocks a response carries.
+ static func decodeSplit(explicit: ArraySlice<UInt8>, derived: ArraySlice<UInt8>) throws -> ResourceOptions {
+ var options = try decode(derived, explicit: false)
+ options.merge(try decode(explicit, explicit: true)) { _, explicitValue in explicitValue }
+ return options
+ }
+
+ /// Reads a `u32`-length-prefixed block; every caller decodes it next,
+ /// which validates it.
+ static func readPrefixed(from reader: inout ByteReader) throws -> ArraySlice<UInt8> {
+ let declared = try reader.readUInt32()
+ // A length that does not fit the platform's Int cannot be in the
+ // buffer either; report it as truncation rather than trapping.
+ guard let length = Int(exactly: declared) else {
+ throw WireError.truncated(offset: reader.position, need: Int.max, have: reader.remaining)
+ }
+ return try reader.readBytes(length)
+ }
+}
diff --git a/foreign/swift/Sources/Iggy/Models/Resources.swift b/foreign/swift/Sources/Iggy/Models/Resources.swift
new file mode 100644
index 000000000..506e7e861
--- /dev/null
+++ b/foreign/swift/Sources/Iggy/Models/Resources.swift
@@ -0,0 +1,628 @@
+// 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.
+
+/// Compression applied to a topic's messages.
+public enum CompressionAlgorithm: UInt8, Sendable, Hashable, CustomStringConvertible {
+ case none = 1
+ case gzip = 2
+
+ public var description: String {
+ switch self {
+ case .none: "none"
+ case .gzip: "gzip"
+ }
+ }
+
+ public init?(name: String) {
+ switch name.lowercased() {
+ case "none": self = .none
+ case "gzip": self = .gzip
+ default: return nil
+ }
+ }
+}
+
+/// How long messages are retained.
+public enum IggyExpiry: Sendable, Hashable {
+ /// Whatever the server is configured with.
+ case serverDefault
+ /// Messages older than the duration are removed.
+ case after(Duration)
+ /// Messages never expire.
+ case never
+
+ /// The wire value: microseconds, `UInt64.max` for never, nil for the
+ /// server default. A zero or negative duration would travel as 0, which
+ /// the server reads as its default, so it is rejected instead.
+ func wireMicroseconds() throws -> UInt64? {
+ switch self {
+ case .serverDefault:
+ return nil
+ case .after(let duration):
+ // Anything under a microsecond would travel as 0 (server default)
+ // and anything past the u64 range would wrap; both are refused.
+ guard duration >= .microseconds(1), let microseconds = duration.wholeMicroseconds, microseconds < UInt64.max else {
+ throw IggyError(.invalidOptionValue, context: "message expiry must be between 1 microsecond and 2^64 - 2 microseconds, got \(duration)")
+ }
+ return microseconds
+ case .never:
+ return UInt64.max
+ }
+ }
+
+ init(wireMicroseconds value: UInt64) {
+ switch value {
+ case 0: self = .serverDefault
+ case UInt64.max: self = .never
+ default: self = .after(.microseconds(Int64(clamping: value)))
+ }
+ }
+}
+
+/// Upper bound on a topic's size.
+public enum MaxTopicSize: Sendable, Hashable {
+ case serverDefault
+ case bytes(UInt64)
+ case unlimited
+
+ var wireValue: UInt64 {
+ switch self {
+ case .serverDefault: 0
+ case .bytes(let bytes): bytes
+ case .unlimited: UInt64.max
+ }
+ }
+
+ init(wireValue value: UInt64) {
+ switch value {
+ case 0: self = .serverDefault
+ case UInt64.max: self = .unlimited
+ default: self = .bytes(value)
+ }
+ }
+}
+
+/// A stream as listed. Named `StreamInfo` so that a file importing
+/// Foundation, which has its own `Stream`, never sees an ambiguous type.
+public struct StreamInfo: Sendable, Hashable {
+ public var id: UInt32
+ public var createdAt: IggyTimestamp
+ public var name: String
+ public var sizeBytes: UInt64
+ public var messagesCount: UInt64
+ public var topicsCount: UInt32
+ public var options: ResourceOptions
+
+ public init(
+ id: UInt32, createdAt: IggyTimestamp, name: String, sizeBytes: UInt64, messagesCount: UInt64, topicsCount: UInt32, options: ResourceOptions = [:]
+ ) {
+ self.id = id
+ self.createdAt = createdAt
+ self.name = name
+ self.sizeBytes = sizeBytes
+ self.messagesCount = messagesCount
+ self.topicsCount = topicsCount
+ self.options = options
+ }
+}
+
+/// A stream with its topics.
+public struct StreamDetails: Sendable, Hashable {
+ public var id: UInt32
+ public var createdAt: IggyTimestamp
+ public var name: String
+ public var sizeBytes: UInt64
+ public var messagesCount: UInt64
+ public var topicsCount: UInt32
+ public var topics: [Topic]
+ public var options: ResourceOptions
+
+ public init(
+ id: UInt32, createdAt: IggyTimestamp, name: String, sizeBytes: UInt64, messagesCount: UInt64, topicsCount: UInt32, topics: [Topic],
+ options: ResourceOptions = [:]
+ ) {
+ self.id = id
+ self.createdAt = createdAt
+ self.name = name
+ self.sizeBytes = sizeBytes
+ self.messagesCount = messagesCount
+ self.topicsCount = topicsCount
+ self.topics = topics
+ self.options = options
+ }
+}
+
+/// A topic as listed.
+public struct Topic: Sendable, Hashable {
+ public var id: UInt32
+ public var createdAt: IggyTimestamp
+ public var name: String
+ public var sizeBytes: UInt64
+ public var messageExpiry: IggyExpiry
+ public var compressionAlgorithm: CompressionAlgorithm
+ public var maxTopicSize: MaxTopicSize
+ public var messagesCount: UInt64
+ public var partitionsCount: UInt32
+ public var options: ResourceOptions
+
+ public init(
+ id: UInt32, createdAt: IggyTimestamp, name: String, sizeBytes: UInt64, messageExpiry: IggyExpiry, compressionAlgorithm: CompressionAlgorithm,
+ maxTopicSize: MaxTopicSize, messagesCount: UInt64, partitionsCount: UInt32, options: ResourceOptions = [:]
+ ) {
+ self.id = id
+ self.createdAt = createdAt
+ self.name = name
+ self.sizeBytes = sizeBytes
+ self.messageExpiry = messageExpiry
+ self.compressionAlgorithm = compressionAlgorithm
+ self.maxTopicSize = maxTopicSize
+ self.messagesCount = messagesCount
+ self.partitionsCount = partitionsCount
+ self.options = options
+ }
+}
+
+/// A topic with its partitions.
+public struct TopicDetails: Sendable, Hashable {
+ public var id: UInt32
+ public var createdAt: IggyTimestamp
+ public var name: String
+ public var sizeBytes: UInt64
+ public var messageExpiry: IggyExpiry
+ public var compressionAlgorithm: CompressionAlgorithm
+ public var maxTopicSize: MaxTopicSize
+ public var messagesCount: UInt64
+ public var partitionsCount: UInt32
+ public var partitions: [Partition]
+ public var options: ResourceOptions
+
+ public init(topic: Topic, partitions: [Partition]) {
+ id = topic.id
+ createdAt = topic.createdAt
+ name = topic.name
+ sizeBytes = topic.sizeBytes
+ messageExpiry = topic.messageExpiry
+ compressionAlgorithm = topic.compressionAlgorithm
+ maxTopicSize = topic.maxTopicSize
+ messagesCount = topic.messagesCount
+ partitionsCount = topic.partitionsCount
+ self.partitions = partitions
+ options = topic.options
+ }
+}
+
+/// One partition of a topic.
+public struct Partition: Sendable, Hashable {
+ public var id: UInt32
+ public var createdAt: IggyTimestamp
+ public var segmentsCount: UInt32
+ public var currentOffset: UInt64
+ public var sizeBytes: UInt64
+ public var messagesCount: UInt64
+
+ public init(id: UInt32, createdAt: IggyTimestamp, segmentsCount: UInt32, currentOffset: UInt64, sizeBytes: UInt64, messagesCount: UInt64) {
+ self.id = id
+ self.createdAt = createdAt
+ self.segmentsCount = segmentsCount
+ self.currentOffset = currentOffset
+ self.sizeBytes = sizeBytes
+ self.messagesCount = messagesCount
+ }
+}
+
+/// A consumer group as listed.
+public struct ConsumerGroup: Sendable, Hashable {
+ public var id: UInt32
+ public var name: String
+ public var partitionsCount: UInt32
+ public var membersCount: UInt32
+
+ public init(id: UInt32, name: String, partitionsCount: UInt32, membersCount: UInt32) {
+ self.id = id
+ self.name = name
+ self.partitionsCount = partitionsCount
+ self.membersCount = membersCount
+ }
+}
+
+/// A consumer group with its members.
+public struct ConsumerGroupDetails: Sendable, Hashable {
+ public var id: UInt32
+ public var name: String
+ public var partitionsCount: UInt32
+ public var membersCount: UInt32
+ public var members: [ConsumerGroupMember]
+
+ public init(id: UInt32, name: String, partitionsCount: UInt32, membersCount: UInt32, members: [ConsumerGroupMember]) {
+ self.id = id
+ self.name = name
+ self.partitionsCount = partitionsCount
+ self.membersCount = membersCount
+ self.members = members
+ }
+}
+
+/// One member of a consumer group and the partitions it owns.
+public struct ConsumerGroupMember: Sendable, Hashable {
+ public var id: UInt32
+ public var partitionsCount: UInt32
+ public var partitions: [UInt32]
+
+ public init(id: UInt32, partitionsCount: UInt32, partitions: [UInt32]) {
+ self.id = id
+ self.partitionsCount = partitionsCount
+ self.partitions = partitions
+ }
+}
+
+/// The stored offset of a consumer on a partition.
+public struct ConsumerOffsetInfo: Sendable, Hashable {
+ public var partitionID: UInt32
+ public var currentOffset: UInt64
+ public var storedOffset: UInt64
+
+ public init(partitionID: UInt32, currentOffset: UInt64, storedOffset: UInt64) {
+ self.partitionID = partitionID
+ self.currentOffset = currentOffset
+ self.storedOffset = storedOffset
+ }
+}
+
+/// The partitions a consumer-group member currently owns.
+public struct ConsumerGroupAssignment: Sendable, Hashable {
+ public var generation: UInt64
+ public var partitions: [UInt32]
+
+ public init(generation: UInt64, partitions: [UInt32]) {
+ self.generation = generation
+ self.partitions = partitions
+ }
+}
+
+/// Whether a user can sign in.
+public enum UserStatus: UInt8, Sendable, Hashable, CustomStringConvertible {
+ case active = 1
+ case inactive = 2
+
+ public var description: String {
+ switch self {
+ case .active: "active"
+ case .inactive: "inactive"
+ }
+ }
+}
+
+/// A user as listed.
+public struct UserInfo: Sendable, Hashable {
+ public var id: UInt32
+ public var createdAt: IggyTimestamp
+ public var status: UserStatus
+ public var username: String
+ public var options: ResourceOptions
+
+ public init(id: UInt32, createdAt: IggyTimestamp, status: UserStatus, username: String, options: ResourceOptions = [:]) {
+ self.id = id
+ self.createdAt = createdAt
+ self.status = status
+ self.username = username
+ self.options = options
+ }
+}
+
+/// A user with its permissions.
+public struct UserInfoDetails: Sendable, Hashable {
+ public var id: UInt32
+ public var createdAt: IggyTimestamp
+ public var status: UserStatus
+ public var username: String
+ public var permissions: Permissions?
+ public var options: ResourceOptions
+
+ public init(id: UInt32, createdAt: IggyTimestamp, status: UserStatus, username: String, permissions: Permissions?, options: ResourceOptions = [:]) {
+ self.id = id
+ self.createdAt = createdAt
+ self.status = status
+ self.username = username
+ self.permissions = permissions
+ self.options = options
+ }
+}
+
+/// Who a sign-in authenticated as.
+public struct IdentityInfo: Sendable, Hashable {
+ public var userID: UInt32
+
+ public init(userID: UInt32) {
+ self.userID = userID
+ }
+}
+
+/// A personal access token as listed. The secret is only shown at creation.
+public struct PersonalAccessTokenInfo: Sendable, Hashable {
+ public var name: String
+ /// When the token expires, nil when it never does.
+ public var expiresAt: IggyTimestamp?
+
+ public init(name: String, expiresAt: IggyTimestamp?) {
+ self.name = name
+ self.expiresAt = expiresAt
+ }
+}
+
+/// The secret of a freshly created personal access token.
+public struct RawPersonalAccessToken: Sendable, Hashable {
+ public var token: String
+
+ public init(token: String) {
+ self.token = token
+ }
+}
+
+/// When a personal access token stops working.
+public typealias PersonalAccessTokenExpiry = IggyExpiry
+
+/// A connected client as listed.
+public struct ClientInfo: Sendable, Hashable {
+ public var clientID: UInt32
+ /// The signed-in user, nil before sign-in.
+ public var userID: UInt32?
+ public var address: String
+ public var transport: String
+ public var consumerGroupsCount: UInt32
+
+ public init(clientID: UInt32, userID: UInt32?, address: String, transport: String, consumerGroupsCount: UInt32) {
+ self.clientID = clientID
+ self.userID = userID
+ self.address = address
+ self.transport = transport
+ self.consumerGroupsCount = consumerGroupsCount
+ }
+}
+
+/// A connected client with the consumer groups it joined.
+public struct ClientInfoDetails: Sendable, Hashable {
+ public var clientID: UInt32
+ public var userID: UInt32?
+ public var address: String
+ public var transport: String
+ public var consumerGroupsCount: UInt32
+ public var consumerGroups: [ConsumerGroupInfo]
+
+ public init(clientID: UInt32, userID: UInt32?, address: String, transport: String, consumerGroupsCount: UInt32, consumerGroups: [ConsumerGroupInfo]) {
+ self.clientID = clientID
+ self.userID = userID
+ self.address = address
+ self.transport = transport
+ self.consumerGroupsCount = consumerGroupsCount
+ self.consumerGroups = consumerGroups
+ }
+}
+
+/// A consumer group a client is a member of.
+public struct ConsumerGroupInfo: Sendable, Hashable {
+ public var streamID: UInt32
+ public var topicID: UInt32
+ public var groupID: UInt32
+
+ public init(streamID: UInt32, topicID: UInt32, groupID: UInt32) {
+ self.streamID = streamID
+ self.topicID = topicID
+ self.groupID = groupID
+ }
+}
+
+/// Cache hit statistics of one partition.
+public struct CacheMetrics: Sendable, Hashable {
+ public var streamID: UInt32
+ public var topicID: UInt32
+ public var partitionID: UInt32
+ public var hits: UInt64
+ public var misses: UInt64
+ public var hitRatio: Float
+
+ public init(streamID: UInt32, topicID: UInt32, partitionID: UInt32, hits: UInt64, misses: UInt64, hitRatio: Float) {
+ self.streamID = streamID
+ self.topicID = topicID
+ self.partitionID = partitionID
+ self.hits = hits
+ self.misses = misses
+ self.hitRatio = hitRatio
+ }
+}
+
+/// Server statistics.
+public struct Stats: Sendable, Hashable {
+ public var processID: UInt32
+ public var cpuUsage: Float
+ public var totalCPUUsage: Float
+ public var memoryUsage: UInt64
+ public var totalMemory: UInt64
+ public var availableMemory: UInt64
+ public var runTime: Duration
+ public var startTime: IggyTimestamp
+ public var readBytes: UInt64
+ public var writtenBytes: UInt64
+ public var messagesSizeBytes: UInt64
+ public var streamsCount: UInt32
+ public var topicsCount: UInt32
+ public var partitionsCount: UInt32
+ public var segmentsCount: UInt32
+ public var messagesCount: UInt64
+ public var clientsCount: UInt32
+ public var consumerGroupsCount: UInt32
+ public var hostname: String
+ public var osName: String
+ public var osVersion: String
+ public var kernelVersion: String
+ public var iggyServerVersion: String
+ public var iggyServerSemver: UInt32?
+ public var cacheMetrics: [CacheMetrics]
+ public var threadsCount: UInt32
+ public var freeDiskSpace: UInt64
+ public var totalDiskSpace: UInt64
+
+ public init(
+ processID: UInt32, cpuUsage: Float, totalCPUUsage: Float, memoryUsage: UInt64, totalMemory: UInt64, availableMemory: UInt64, runTime: Duration,
+ startTime: IggyTimestamp, readBytes: UInt64, writtenBytes: UInt64, messagesSizeBytes: UInt64, streamsCount: UInt32, topicsCount: UInt32,
+ partitionsCount: UInt32, segmentsCount: UInt32, messagesCount: UInt64, clientsCount: UInt32, consumerGroupsCount: UInt32, hostname: String,
+ osName: String, osVersion: String, kernelVersion: String, iggyServerVersion: String, iggyServerSemver: UInt32?, cacheMetrics: [CacheMetrics],
+ threadsCount: UInt32, freeDiskSpace: UInt64, totalDiskSpace: UInt64
+ ) {
+ self.processID = processID
+ self.cpuUsage = cpuUsage
+ self.totalCPUUsage = totalCPUUsage
+ self.memoryUsage = memoryUsage
+ self.totalMemory = totalMemory
+ self.availableMemory = availableMemory
+ self.runTime = runTime
+ self.startTime = startTime
+ self.readBytes = readBytes
+ self.writtenBytes = writtenBytes
+ self.messagesSizeBytes = messagesSizeBytes
+ self.streamsCount = streamsCount
+ self.topicsCount = topicsCount
+ self.partitionsCount = partitionsCount
+ self.segmentsCount = segmentsCount
+ self.messagesCount = messagesCount
+ self.clientsCount = clientsCount
+ self.consumerGroupsCount = consumerGroupsCount
+ self.hostname = hostname
+ self.osName = osName
+ self.osVersion = osVersion
+ self.kernelVersion = kernelVersion
+ self.iggyServerVersion = iggyServerVersion
+ self.iggyServerSemver = iggyServerSemver
+ self.cacheMetrics = cacheMetrics
+ self.threadsCount = threadsCount
+ self.freeDiskSpace = freeDiskSpace
+ self.totalDiskSpace = totalDiskSpace
+ }
+}
+
+/// Role of a node in a cluster.
+public enum ClusterNodeRole: UInt8, Sendable, Hashable {
+ case leader = 0
+ case follower = 1
+}
+
+/// Health of a node in a cluster.
+public enum ClusterNodeStatus: UInt8, Sendable, Hashable {
+ case healthy = 0
+ case starting = 1
+ case stopping = 2
+ case unreachable = 3
+ case maintenance = 4
+}
+
+/// Ports a cluster node listens on; zero when a transport is disabled.
+public struct TransportEndpoints: Sendable, Hashable {
+ public var tcp: UInt16
+ public var quic: UInt16
+ public var http: UInt16
+ public var websocket: UInt16
+
+ public init(tcp: UInt16, quic: UInt16, http: UInt16, websocket: UInt16) {
+ self.tcp = tcp
+ self.quic = quic
+ self.http = http
+ self.websocket = websocket
+ }
+}
+
+/// One node of the cluster roster.
+public struct ClusterNode: Sendable, Hashable {
+ public var name: String
+ public var ip: String
+ public var endpoints: TransportEndpoints
+ public var role: ClusterNodeRole
+ public var status: ClusterNodeStatus
+
+ public init(name: String, ip: String, endpoints: TransportEndpoints, role: ClusterNodeRole, status: ClusterNodeStatus) {
+ self.name = name
+ self.ip = ip
+ self.endpoints = endpoints
+ self.role = role
+ self.status = status
+ }
+
+ /// The `host:port` of the node for the TCP transport, bracketing an IPv6
+ /// literal so the spelling can be dialed.
+ public var tcpAddress: String {
+ if ip.contains(":") && !ip.hasPrefix("[") {
+ return "[\(ip)]:\(endpoints.tcp)"
+ }
+ return "\(ip):\(endpoints.tcp)"
+ }
+}
+
+/// The cluster roster.
+public struct ClusterMetadata: Sendable, Hashable {
+ public var name: String
+ public var nodes: [ClusterNode]
+
+ public init(name: String, nodes: [ClusterNode]) {
+ self.name = name
+ self.nodes = nodes
+ }
+}
+
+/// What a snapshot captures.
+public enum SystemSnapshotType: UInt8, Sendable, Hashable {
+ case filesystemOverview = 1
+ case processList = 2
+ case resourceUsage = 3
+ case test = 4
+ case serverLogs = 5
+ case serverConfig = 6
+ case all = 100
+}
+
+/// How a snapshot archive is compressed.
+public enum SnapshotCompression: UInt8, Sendable, Hashable {
+ case stored = 1
+ case deflated = 2
+ case bzip2 = 3
+ case zstd = 4
+ case lzma = 5
+ case xz = 6
+}
+
+/// A snapshot archive.
+public struct Snapshot: Sendable, Hashable {
+ public var data: [UInt8]
+
+ public init(data: [UInt8]) {
+ self.data = data
+ }
+}
+
+extension Duration {
+ /// Whole microseconds, or nil when the duration is negative or does not
+ /// fit a `UInt64`.
+ var wholeMicroseconds: UInt64? {
+ let (seconds, attoseconds) = components
+ guard seconds >= 0 else {
+ return nil
+ }
+ let (scaled, overflow) = UInt64(seconds).multipliedReportingOverflow(by: 1_000_000)
+ if overflow {
+ return nil
+ }
+ let (total, carry) = scaled.addingReportingOverflow(UInt64(attoseconds / 1_000_000_000_000))
+ return carry ? nil : total
+ }
+}
diff --git a/foreign/swift/Sources/Iggy/Models/TopicOptions.swift b/foreign/swift/Sources/Iggy/Models/TopicOptions.swift
new file mode 100644
index 000000000..fd9578ab7
--- /dev/null
+++ b/foreign/swift/Sources/Iggy/Models/TopicOptions.swift
@@ -0,0 +1,206 @@
+// 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.
+
+/// Keys of the topic option catalog.
+public enum TopicOptionKey {
+ public static let compressionAlgorithm = "compression_algorithm"
+ public static let messageExpiry = "message_expiry"
+ public static let maxTopicSize = "max_topic_size"
+ public static let segmentSize = "segment_size"
+ public static let durability = "durability"
+ public static let consumerOffsetDurability = "consumer_offset_durability"
+ public static let messagesRequiredToSave = "messages_required_to_save"
+ public static let sizeOfMessagesRequiredToSave = "size_of_messages_required_to_save"
+ public static let preallocateSegments = "preallocate_segments"
+}
+
+/// How far a write has to get before the server completes it.
+public enum Durability: String, Sendable, Hashable, CaseIterable {
+ /// Quorum commit without an additional stable-storage barrier.
+ case replicated
+ /// Quorum commit backed by recoverable stable-storage copies.
+ case persisted
+
+ /// Parses the wire spelling case-insensitively, as the server does.
+ init?(parsing text: String) {
+ self.init(rawValue: text.lowercased())
+ }
+}
+
+/// Options of a topic being created. Every knob is optional; an absent key
+/// resolves against the server's defaults at admission.
+///
+/// `raw` carries keys this SDK has no typed field for, as strings the server
+/// parses, so a newer server's option costs no SDK change.
+public struct TopicCreateOptions: Sendable, Hashable {
+ /// Number of partitions the topic starts with. A field of the create
+ /// command rather than an option key; nil means 1, the server's
+ /// `DEFAULT_PARTITIONS_COUNT`.
+ public var partitionsCount: UInt32?
+ public var compressionAlgorithm: CompressionAlgorithm?
+ public var messageExpiry: IggyExpiry?
+ public var maxTopicSize: MaxTopicSize?
+ public var segmentSizeBytes: UInt64?
+ /// Completion policy for messages. Always sent, like the Rust client does,
+ /// so the topic records the policy it was created with.
+ public var durability: Durability
+ /// Completion policy for stored consumer offsets, independent of `durability`.
+ public var consumerOffsetDurability: Durability
+ public var messagesRequiredToSave: UInt32?
+ public var sizeOfMessagesRequiredToSaveBytes: UInt64?
+ public var preallocateSegments: Bool?
+ public var raw: [String: String]
+
+ public init(
+ partitionsCount: UInt32? = nil, compressionAlgorithm: CompressionAlgorithm? = nil, messageExpiry: IggyExpiry? = nil,
+ maxTopicSize: MaxTopicSize? = nil, segmentSizeBytes: UInt64? = nil, durability: Durability = .replicated,
+ consumerOffsetDurability: Durability = .replicated, messagesRequiredToSave: UInt32? = nil,
+ sizeOfMessagesRequiredToSaveBytes: UInt64? = nil, preallocateSegments: Bool? = nil, raw: [String: String] = [:]
+ ) {
+ self.partitionsCount = partitionsCount
+ self.compressionAlgorithm = compressionAlgorithm
+ self.messageExpiry = messageExpiry
+ self.maxTopicSize = maxTopicSize
+ self.segmentSizeBytes = segmentSizeBytes
+ self.durability = durability
+ self.consumerOffsetDurability = consumerOffsetDurability
+ self.messagesRequiredToSave = messagesRequiredToSave
+ self.sizeOfMessagesRequiredToSaveBytes = sizeOfMessagesRequiredToSaveBytes
+ self.preallocateSegments = preallocateSegments
+ self.raw = raw
+ }
+
+ func toResourceOptions() throws -> ResourceOptions {
+ var options = try TopicUpdateOptions(
+ compressionAlgorithm: compressionAlgorithm, messageExpiry: messageExpiry, maxTopicSize: maxTopicSize, raw: raw
+ ).toResourceOptions()
+ if let segmentSizeBytes {
+ options[TopicOptionKey.segmentSize] = .explicit(.uint64(segmentSizeBytes))
+ }
+ for (key, policy) in [(TopicOptionKey.durability, durability), (TopicOptionKey.consumerOffsetDurability, consumerOffsetDurability)] {
+ // A raw entry may strengthen the default; it may not weaken a
+ // typed `persisted`, which would make the two fields disagree.
+ // Whatever spelling it used, the canonical one is sent.
+ var effective = policy
+ if let rawValue = raw[key] {
+ guard let parsed = Durability(parsing: rawValue) else {
+ throw IggyError(.invalidOptionValue, context: key)
+ }
+ if policy == .persisted, parsed != policy {
+ throw IggyError(.invalidOptionValue, context: key)
+ }
+ effective = parsed
+ }
+ options[key] = .explicit(try .string(effective.rawValue))
+ }
+ if let messagesRequiredToSave {
+ options[TopicOptionKey.messagesRequiredToSave] = .explicit(.uint32(messagesRequiredToSave))
+ }
+ if let sizeOfMessagesRequiredToSaveBytes {
+ options[TopicOptionKey.sizeOfMessagesRequiredToSave] = .explicit(.uint64(sizeOfMessagesRequiredToSaveBytes))
+ }
+ if let preallocateSegments {
+ options[TopicOptionKey.preallocateSegments] = .explicit(.bool(preallocateSegments))
+ }
+ return options
+ }
+
+ func encode() throws -> [UInt8] {
+ try OptionsBlock.encode(try toResourceOptions())
+ }
+}
+
+/// Options of a topic being updated.
+public struct TopicUpdateOptions: Sendable, Hashable {
+ public var compressionAlgorithm: CompressionAlgorithm?
+ public var messageExpiry: IggyExpiry?
+ public var maxTopicSize: MaxTopicSize?
+ public var raw: [String: String]
+
+ public init(
+ compressionAlgorithm: CompressionAlgorithm? = nil, messageExpiry: IggyExpiry? = nil, maxTopicSize: MaxTopicSize? = nil, raw: [String: String] = [:]
+ ) {
+ self.compressionAlgorithm = compressionAlgorithm
+ self.messageExpiry = messageExpiry
+ self.maxTopicSize = maxTopicSize
+ self.raw = raw
+ }
+
+ func toResourceOptions() throws -> ResourceOptions {
+ var options = try rawOptions(raw)
+ if let compressionAlgorithm {
+ options[TopicOptionKey.compressionAlgorithm] = .explicit(try .string(compressionAlgorithm.description))
+ }
+ if let messageExpiry, let microseconds = try messageExpiry.wireMicroseconds() {
+ options[TopicOptionKey.messageExpiry] = .explicit(.uint64(microseconds))
+ }
+ if let maxTopicSize {
+ options[TopicOptionKey.maxTopicSize] = .explicit(.uint64(maxTopicSize.wireValue))
+ }
+ return options
+ }
+
+ func encode() throws -> [UInt8] {
+ try OptionsBlock.encode(try toResourceOptions())
+ }
+}
+
+/// Options of a stream being updated. The catalog has no stream keys today;
+/// `raw` is forwarded as-is for forward compatibility.
+public struct StreamUpdateOptions: Sendable, Hashable {
+ public var raw: [String: String]
+
+ public init(raw: [String: String] = [:]) {
+ self.raw = raw
+ }
+
+ func encode() throws -> [UInt8] {
+ try OptionsBlock.encode(try rawOptions(raw))
+ }
+}
+
+/// Options of a user being updated. The catalog has no user keys today;
+/// `raw` is forwarded as-is for forward compatibility.
+public struct UserUpdateOptions: Sendable, Hashable {
+ public var raw: [String: String]
+
+ public init(raw: [String: String] = [:]) {
+ self.raw = raw
+ }
+
+ func encode() throws -> [UInt8] {
+ try OptionsBlock.encode(try rawOptions(raw))
+ }
+}
+
+/// Characters of an offending key an error message shows,
+/// `ERROR_KEY_PREVIEW_LEN` in the Rust crate.
+private let errorKeyPreviewLength = 64
+
+private func rawOptions(_ raw: [String: String]) throws -> ResourceOptions {
+ var options: ResourceOptions = [:]
+ for (key, value) in raw {
+ guard !key.isEmpty, key.utf8.count <= HeaderTLV.maxFieldLength else {
+ throw IggyError(.unsupportedOptionKey, context: String(key.prefix(errorKeyPreviewLength)))
+ }
+ guard let headerValue = try? HeaderValue.string(value) else {
+ throw IggyError(.invalidOptionValue, context: String(key.prefix(errorKeyPreviewLength)))
+ }
+ options[key] = .explicit(headerValue)
+ }
+ return options
+}
diff --git a/foreign/swift/Sources/Iggy/Models/UserHeaders.swift b/foreign/swift/Sources/Iggy/Models/UserHeaders.swift
new file mode 100644
index 000000000..2c46c5955
--- /dev/null
+++ b/foreign/swift/Sources/Iggy/Models/UserHeaders.swift
@@ -0,0 +1,427 @@
+// 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.
+
+/// Type tag of a user header key or value.
+public enum HeaderKind: UInt8, Sendable, Hashable, CaseIterable, CustomStringConvertible {
+ case raw = 1
+ case string = 2
+ case bool = 3
+ case int8 = 4
+ case int16 = 5
+ case int32 = 6
+ case int64 = 7
+ case int128 = 8
+ case uint8 = 9
+ case uint16 = 10
+ case uint32 = 11
+ case uint64 = 12
+ case uint128 = 13
+ case float32 = 14
+ case float64 = 15
+
+ /// Byte length of the fixed-size kinds; nil for `raw` and `string`.
+ var expectedSize: Int? {
+ switch self {
+ case .raw, .string: nil
+ case .bool, .int8, .uint8: 1
+ case .int16, .uint16: 2
+ case .int32, .uint32, .float32: 4
+ case .int64, .uint64, .float64: 8
+ case .int128, .uint128: 16
+ }
+ }
+
+ public var description: String {
+ switch self {
+ case .raw: "raw"
+ case .string: "string"
+ case .bool: "bool"
+ case .int8: "int8"
+ case .int16: "int16"
+ case .int32: "int32"
+ case .int64: "int64"
+ case .int128: "int128"
+ case .uint8: "uint8"
+ case .uint16: "uint16"
+ case .uint32: "uint32"
+ case .uint64: "uint64"
+ case .uint128: "uint128"
+ case .float32: "float32"
+ case .float64: "float64"
+ }
+ }
+
+ public init?(name: String) {
+ guard let kind = HeaderKind.allCases.first(where: { $0.description == name }) else {
+ return nil
+ }
+ self = kind
+ }
+}
+
+/// A typed value carried in a user header. The same shape describes option
+/// values in resource metadata.
+public struct HeaderValue: Sendable, Hashable, CustomStringConvertible {
+ public let kind: HeaderKind
+ public let bytes: [UInt8]
+
+ /// Wraps bytes already laid out for `kind`, validating fixed-size kinds.
+ /// The wire allows 1 to 255 bytes per field, so an empty value is
+ /// refused here rather than by the server.
+ public init(kind: HeaderKind, bytes: [UInt8]) throws {
+ if let expected = kind.expectedSize, bytes.count != expected {
+ throw IggyError(.invalidHeaderValue, context: "\(kind) needs \(expected) bytes, got \(bytes.count)")
+ }
+ if bytes.isEmpty || bytes.count > HeaderTLV.maxFieldLength {
+ throw IggyError(.invalidHeaderValue, context: "value must be 1-\(HeaderTLV.maxFieldLength) bytes, got \(bytes.count)")
+ }
+ self.kind = kind
+ self.bytes = bytes
+ }
+
+ init(uncheckedKind kind: HeaderKind, bytes: [UInt8]) {
+ self.kind = kind
+ self.bytes = bytes
+ }
+
+ public static func raw(_ bytes: [UInt8]) throws -> HeaderValue {
+ try HeaderValue(kind: .raw, bytes: bytes)
+ }
+
+ public static func string(_ value: String) throws -> HeaderValue {
+ try HeaderValue(kind: .string, bytes: Array(value.utf8))
+ }
+
+ public static func bool(_ value: Bool) -> HeaderValue {
+ HeaderValue(uncheckedKind: .bool, bytes: [value ? 1 : 0])
+ }
+
+ public static func int8(_ value: Int8) -> HeaderValue {
+ HeaderValue(uncheckedKind: .int8, bytes: [UInt8(bitPattern: value)])
+ }
+
+ public static func int16(_ value: Int16) -> HeaderValue {
+ HeaderValue(uncheckedKind: .int16, bytes: UInt16(bitPattern: value).littleEndianBytes)
+ }
+
+ public static func int32(_ value: Int32) -> HeaderValue {
+ HeaderValue(uncheckedKind: .int32, bytes: UInt32(bitPattern: value).littleEndianBytes)
+ }
+
+ public static func int64(_ value: Int64) -> HeaderValue {
+ HeaderValue(uncheckedKind: .int64, bytes: UInt64(bitPattern: value).littleEndianBytes)
+ }
+
+ /// A two's-complement 128-bit integer given as its unsigned bit pattern.
+ public static func int128(bitPattern value: UInt128Value) -> HeaderValue {
+ HeaderValue(uncheckedKind: .int128, bytes: value.littleEndianBytes)
+ }
+
+ public static func uint8(_ value: UInt8) -> HeaderValue {
+ HeaderValue(uncheckedKind: .uint8, bytes: [value])
+ }
+
+ public static func uint16(_ value: UInt16) -> HeaderValue {
+ HeaderValue(uncheckedKind: .uint16, bytes: value.littleEndianBytes)
+ }
+
+ public static func uint32(_ value: UInt32) -> HeaderValue {
+ HeaderValue(uncheckedKind: .uint32, bytes: value.littleEndianBytes)
+ }
+
+ public static func uint64(_ value: UInt64) -> HeaderValue {
+ HeaderValue(uncheckedKind: .uint64, bytes: value.littleEndianBytes)
+ }
+
+ public static func uint128(_ value: UInt128Value) -> HeaderValue {
+ HeaderValue(uncheckedKind: .uint128, bytes: value.littleEndianBytes)
+ }
+
+ public static func float32(_ value: Float) -> HeaderValue {
+ HeaderValue(uncheckedKind: .float32, bytes: value.bitPattern.littleEndianBytes)
+ }
+
+ public static func float64(_ value: Double) -> HeaderValue {
+ HeaderValue(uncheckedKind: .float64, bytes: value.bitPattern.littleEndianBytes)
+ }
+
+ public var stringValue: String? {
+ kind == .string ? String(decoding: bytes, as: UTF8.self) : nil
+ }
+
+ public var boolValue: Bool? {
+ guard kind == .bool, let byte = bytes.first else { return nil }
+ return byte != 0
+ }
+
+ public var int8Value: Int8? {
+ guard kind == .int8, let byte = bytes.first else { return nil }
+ return Int8(bitPattern: byte)
+ }
+
+ public var int16Value: Int16? {
+ kind == .int16 ? Int16(bitPattern: UInt16(littleEndianBytes: bytes[...])) : nil
+ }
+
+ public var int32Value: Int32? {
+ kind == .int32 ? Int32(bitPattern: UInt32(littleEndianBytes: bytes[...])) : nil
+ }
+
+ public var int64Value: Int64? {
+ kind == .int64 ? Int64(bitPattern: UInt64(littleEndianBytes: bytes[...])) : nil
+ }
+
+ public var int128BitPattern: UInt128Value? {
+ kind == .int128 ? UInt128Value(littleEndianBytes: bytes[...]) : nil
+ }
+
+ public var uint8Value: UInt8? {
+ kind == .uint8 ? bytes.first : nil
+ }
+
+ public var uint16Value: UInt16? {
+ kind == .uint16 ? UInt16(littleEndianBytes: bytes[...]) : nil
+ }
+
+ public var uint32Value: UInt32? {
+ kind == .uint32 ? UInt32(littleEndianBytes: bytes[...]) : nil
+ }
+
+ public var uint64Value: UInt64? {
+ kind == .uint64 ? UInt64(littleEndianBytes: bytes[...]) : nil
+ }
+
+ public var uint128Value: UInt128Value? {
+ kind == .uint128 ? UInt128Value(littleEndianBytes: bytes[...]) : nil
+ }
+
+ public var float32Value: Float? {
+ kind == .float32 ? Float(bitPattern: UInt32(littleEndianBytes: bytes[...])) : nil
+ }
+
+ public var float64Value: Double? {
+ kind == .float64 ? Double(bitPattern: UInt64(littleEndianBytes: bytes[...])) : nil
+ }
+
+ /// A readable rendering of the value regardless of its kind.
+ public var description: String {
+ switch kind {
+ case .raw: bytes.map { String($0, radix: 16).leftPadded(to: 2) }.joined()
+ case .string: stringValue ?? ""
+ case .bool: String(boolValue ?? false)
+ case .int8: String(int8Value ?? 0)
+ case .int16: String(int16Value ?? 0)
+ case .int32: String(int32Value ?? 0)
+ case .int64: String(int64Value ?? 0)
+ case .int128: (int128BitPattern ?? .zero).description
+ case .uint8: String(uint8Value ?? 0)
+ case .uint16: String(uint16Value ?? 0)
+ case .uint32: String(uint32Value ?? 0)
+ case .uint64: String(uint64Value ?? 0)
+ case .uint128: (uint128Value ?? .zero).description
+ case .float32: String(float32Value ?? 0)
+ case .float64: String(float64Value ?? 0)
+ }
+ }
+}
+
+extension HeaderValue: ExpressibleByStringLiteral {
+ /// A string literal must be 1 to 255 bytes of UTF-8; the literal form
+ /// traps otherwise. Use ``string(_:)`` to validate a runtime value.
+ public init(stringLiteral value: String) {
+ do {
+ self = try .string(value)
+ } catch {
+ preconditionFailure("invalid header value literal: \(error)")
+ }
+ }
+}
+
+extension HeaderValue: ExpressibleByBooleanLiteral {
+ public init(booleanLiteral value: Bool) {
+ self = .bool(value)
+ }
+}
+
+extension HeaderValue: ExpressibleByIntegerLiteral {
+ public init(integerLiteral value: Int64) {
+ self = .int64(value)
+ }
+}
+
+extension HeaderValue: ExpressibleByFloatLiteral {
+ public init(floatLiteral value: Double) {
+ self = .float64(value)
+ }
+}
+
+/// Key of a user header. Almost always a string, but the wire allows any
+/// kind, so the key carries its own tag.
+public struct HeaderKey: Sendable, Hashable, CustomStringConvertible {
+ public let kind: HeaderKind
+ public let bytes: [UInt8]
+
+ public init(kind: HeaderKind, bytes: [UInt8]) throws {
+ if bytes.isEmpty || bytes.count > HeaderTLV.maxFieldLength {
+ throw IggyError(.invalidHeaderKey, context: "key must be 1-\(HeaderTLV.maxFieldLength) bytes, got \(bytes.count)")
+ }
+ if let expected = kind.expectedSize, bytes.count != expected {
+ throw IggyError(.invalidHeaderKey, context: "\(kind) needs \(expected) bytes, got \(bytes.count)")
+ }
+ self.kind = kind
+ self.bytes = bytes
+ }
+
+ init(uncheckedKind kind: HeaderKind, bytes: [UInt8]) {
+ self.kind = kind
+ self.bytes = bytes
+ }
+
+ public init(_ name: String) throws {
+ try self.init(kind: .string, bytes: Array(name.utf8))
+ }
+
+ public var stringValue: String? {
+ kind == .string ? String(decoding: bytes, as: UTF8.self) : nil
+ }
+
+ public var description: String {
+ stringValue ?? bytes.map { String($0, radix: 16).leftPadded(to: 2) }.joined()
+ }
+}
+
+extension HeaderKey: ExpressibleByStringLiteral {
+ /// A string literal must be 1 to 255 bytes of UTF-8; the literal form
+ /// traps otherwise. Use ``init(_:)`` to validate a runtime value.
+ public init(stringLiteral value: String) {
+ do {
+ try self.init(value)
+ } catch {
+ preconditionFailure("invalid header key literal: \(error)")
+ }
+ }
+}
+
+extension HeaderKey: Comparable {
+ /// Ordered like the Rust `BTreeMap` the reference SDK keeps headers in,
+ /// so an encoded block is byte-identical across SDKs.
+ public static func < (lhs: HeaderKey, rhs: HeaderKey) -> Bool {
+ if lhs.kind != rhs.kind {
+ return lhs.kind.rawValue < rhs.kind.rawValue
+ }
+ return lhs.bytes.lexicographicallyPrecedes(rhs.bytes)
+ }
+}
+
+/// User headers attached to a message.
+public typealias UserHeaders = [HeaderKey: HeaderValue]
+
+extension String {
+ fileprivate func leftPadded(to width: Int) -> String {
+ count >= width ? self : String(repeating: "0", count: width - count) + self
+ }
+}
+
+/// The TLV encoding shared by user headers and resource options:
+/// `[key_kind:u8][key_len:u32][key][value_kind:u8][value_len:u32][value]`.
+enum HeaderTLV {
+ /// Longest key or value field the wire carries.
+ static let maxFieldLength = 255
+ /// Bytes each entry spends on its two kind tags and two `u32` lengths,
+ /// `OPTION_ENTRY_OVERHEAD` in the Rust crate.
+ static let entryOverhead = 2 * (1 + 4)
+
+ static func encodedSize(_ headers: UserHeaders) -> Int {
+ headers.reduce(0) { $0 + entryOverhead + $1.key.bytes.count + $1.value.bytes.count }
+ }
+
+ /// Encodes in key order, the deterministic layout the Rust SDK produces.
+ static func encode(_ headers: UserHeaders) -> [UInt8] {
+ var writer = ByteWriter(capacity: encodedSize(headers))
+ for (key, value) in headers.sorted(by: { $0.key < $1.key }) {
+ writer.write(key.kind.rawValue)
+ writer.write(UInt32(key.bytes.count))
+ writer.write(key.bytes)
+ writer.write(value.kind.rawValue)
+ writer.write(UInt32(value.bytes.count))
+ writer.write(value.bytes)
+ }
+ return writer.bytes
+ }
+
+ /// One structurally validated TLV entry. Kind codes are not interpreted so
+ /// unknown kinds stay forwardable.
+ struct Entry {
+ let keyKind: UInt8
+ let key: ArraySlice<UInt8>
+ let valueKind: UInt8
+ let value: ArraySlice<UInt8>
+ }
+
+ /// Walks the TLV pairs, checking every kind byte is non-zero, every length
+ /// is in `1...255`, nothing overruns, nothing trails, and entries pair up.
+ static func validate(_ bytes: ArraySlice<UInt8>) throws -> [Entry] {
+ var reader = ByteReader(bytes)
+ var entries: [Entry] = []
+ while !reader.isAtEnd {
+ let (keyKind, key) = try readField(from: &reader)
+ if reader.isAtEnd {
+ throw WireError.validation("odd number of TLV entries (\(2 * entries.count + 1)), expected key-value pairs")
+ }
+ let (valueKind, value) = try readField(from: &reader)
+ entries.append(Entry(keyKind: keyKind, key: key, valueKind: valueKind, value: value))
+ }
+ return entries
+ }
+
+ /// `[kind:u8][length:u32][bytes]`. The length is bounded as a `UInt32`
+ /// before it becomes an `Int`, so a hostile value never traps on a
+ /// 32-bit platform.
+ private static func readField(from reader: inout ByteReader) throws -> (UInt8, ArraySlice<UInt8>) {
+ let kind = try reader.readUInt8()
+ if kind == 0 {
+ throw WireError.validation("header kind is 0 (reserved) at offset \(reader.position - 1)")
+ }
+ let length = try reader.readUInt32()
+ guard (1...UInt32(maxFieldLength)).contains(length) else {
+ throw WireError.validation("header field length \(length) out of range 1...\(maxFieldLength)")
+ }
+ return (kind, try reader.readBytes(Int(length)))
+ }
+
+ /// Decodes into typed headers. An unknown kind code fails the block; the
+ /// options path does its own skipping of unknown value kinds.
+ static func decode(_ bytes: ArraySlice<UInt8>) throws -> UserHeaders {
+ var headers: UserHeaders = [:]
+ for entry in try validate(bytes) {
+ guard let keyKind = HeaderKind(rawValue: entry.keyKind) else {
+ throw IggyError(.invalidHeaderKind, context: "unknown key kind \(entry.keyKind)")
+ }
+ guard let valueKind = HeaderKind(rawValue: entry.valueKind) else {
+ throw IggyError(.invalidHeaderKind, context: "unknown value kind \(entry.valueKind)")
+ }
+ if let expected = keyKind.expectedSize, entry.key.count != expected {
+ throw IggyError(.invalidHeaderKey)
+ }
+ if let expected = valueKind.expectedSize, entry.value.count != expected {
+ throw IggyError(.invalidHeaderValue)
+ }
+ headers[HeaderKey(uncheckedKind: keyKind, bytes: Array(entry.key))] = HeaderValue(uncheckedKind: valueKind, bytes: Array(entry.value))
+ }
+ return headers
+ }
+}
diff --git a/foreign/swift/Sources/Iggy/Wire/ByteCodec.swift b/foreign/swift/Sources/Iggy/Wire/ByteCodec.swift
index 6233b1cef..a0c51996e 100644
--- a/foreign/swift/Sources/Iggy/Wire/ByteCodec.swift
+++ b/foreign/swift/Sources/Iggy/Wire/ByteCodec.swift
@@ -216,3 +216,10 @@ extension UInt64 {
self = bytes.withUnsafeBytes { UInt64(littleEndian: $0.loadUnaligned(as: UInt64.self)) }
}
}
+
+extension FixedWidthInteger {
+ /// Little-endian wire bytes, the layout user header values carry.
+ var littleEndianBytes: [UInt8] {
+ withUnsafeBytes(of: littleEndian) { Array($0) }
+ }
+}
diff --git a/foreign/swift/Tests/IggyTests/Fixtures/golden.json b/foreign/swift/Tests/IggyTests/Fixtures/golden.json
index 5ce2ef942..0cb0e5413 100644
--- a/foreign/swift/Tests/IggyTests/Fixtures/golden.json
+++ b/foreign/swift/Tests/IggyTests/Fixtures/golden.json
@@ -1,6 +1,8 @@
{
"protocol_version": 11264,
"protocol_version_min": 11264,
+ "no_assigned_partition": 4294967294,
+ "resync_required_partition": 4294967295,
"xxh3_64": {
"0": "2d06800538d394c2",
"1": "4c5cca45d0f4811f",
diff --git a/foreign/swift/Tests/IggyTests/ModelTests.swift b/foreign/swift/Tests/IggyTests/ModelTests.swift
new file mode 100644
index 000000000..72d778fa9
--- /dev/null
+++ b/foreign/swift/Tests/IggyTests/ModelTests.swift
@@ -0,0 +1,379 @@
+// 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 Testing
+
+@testable import Iggy
+
+@Suite("User headers")
+struct UserHeadersTests {
+ @Test func typedValuesRoundTrip() throws {
+ let headers: UserHeaders = [
+ "raw": try .raw([1, 2, 3]),
+ "string": "text",
+ "bool": true,
+ "int8": .int8(-8),
+ "int16": .int16(-16),
+ "int32": .int32(-32),
+ "int64": -64,
+ "int128": .int128(bitPattern: UInt128Value(low: 1, high: 2)),
+ "uint8": .uint8(8),
+ "uint16": .uint16(16),
+ "uint32": .uint32(32),
+ "uint64": .uint64(64),
+ "uint128": .uint128(UInt128Value(low: 3, high: 4)),
+ "float32": .float32(1.25),
+ "float64": 2.5,
+ ]
+ let encoded = HeaderTLV.encode(headers)
+ let decoded = try HeaderTLV.decode(encoded[...])
+ #expect(decoded == headers)
+ #expect(decoded["raw"]?.bytes == [1, 2, 3])
+ #expect(decoded["string"]?.stringValue == "text")
+ #expect(decoded["bool"]?.boolValue == true)
+ #expect(decoded["int8"]?.int8Value == -8)
+ #expect(decoded["int16"]?.int16Value == -16)
+ #expect(decoded["int32"]?.int32Value == -32)
+ #expect(decoded["int64"]?.int64Value == -64)
+ #expect(decoded["int128"]?.int128BitPattern == UInt128Value(low: 1, high: 2))
+ #expect(decoded["uint8"]?.uint8Value == 8)
+ #expect(decoded["uint16"]?.uint16Value == 16)
+ #expect(decoded["uint32"]?.uint32Value == 32)
+ #expect(decoded["uint64"]?.uint64Value == 64)
+ #expect(decoded["uint128"]?.uint128Value == UInt128Value(low: 3, high: 4))
+ #expect(decoded["float32"]?.float32Value == 1.25)
+ #expect(decoded["float64"]?.float64Value == 2.5)
+ #expect(decoded["float64"]?.stringValue == nil)
+ #expect(HeaderTLV.encodedSize(headers) == encoded.count)
+ }
+
+ @Test func encodingIsSortedByKey() {
+ let encoded = HeaderTLV.encode(["b": "2", "a": "1"])
+ #expect(encoded == HeaderTLV.encode(["a": "1", "b": "2"]))
+ #expect(encoded[5] == UInt8(ascii: "a"))
+ }
+
+ @Test func structuralValidation() throws {
+ #expect(throws: WireError.self) { try HeaderTLV.validate([0, 1, 0, 0, 0, 42]) }
+ #expect(throws: WireError.self) { try HeaderTLV.validate([1, 0, 0, 0, 0]) }
+ #expect(throws: WireError.self) { try HeaderTLV.validate([1, 2, 0, 0, 0, 97, 98]) }
+ var trailing = HeaderTLV.encode(["k": "v"])
+ trailing.append(0xFF)
+ #expect(throws: WireError.self) { try HeaderTLV.validate(trailing[...]) }
+ #expect(try HeaderTLV.validate([]).isEmpty)
+ }
+
+ @Test func unknownKindsNameTheCode() throws {
+ let unknownValue: [UInt8] = [2, 1, 0, 0, 0, UInt8(ascii: "k"), 200, 1, 0, 0, 0, 1]
+ #expect(throws: IggyError(.invalidHeaderKind)) { try HeaderTLV.decode(unknownValue[...]) }
+ do {
+ _ = try HeaderTLV.decode(unknownValue[...])
+ } catch let error as IggyError {
+ #expect(error.context == "unknown value kind 200")
+ }
+ let unknownKey: [UInt8] = [201, 1, 0, 0, 0, UInt8(ascii: "k"), 2, 1, 0, 0, 0, UInt8(ascii: "v")]
+ do {
+ _ = try HeaderTLV.decode(unknownKey[...])
+ } catch let error as IggyError {
+ #expect(error.context == "unknown key kind 201")
+ }
+ }
+
+ /// A length above 255 is refused as a `UInt32`, before it could trap a
+ /// 32-bit `Int` conversion.
+ @Test func hugeFieldLengthsAreValidationErrors() {
+ let huge: [UInt8] = [2, 0x00, 0x00, 0x00, 0x80]
+ #expect(throws: WireError.validation("header field length 2147483648 out of range 1...255")) { try HeaderTLV.validate(huge[...]) }
+ let oddPair: [UInt8] = [2, 1, 0, 0, 0, UInt8(ascii: "k")]
+ #expect(throws: WireError.validation("odd number of TLV entries (1), expected key-value pairs")) { try HeaderTLV.validate(oddPair[...]) }
+ }
+
+ @Test func emptyValuesAreRejectedLikeTheWireDoes() {
+ #expect(throws: IggyError(.invalidHeaderValue)) { try HeaderValue.string("") }
+ #expect(throws: IggyError(.invalidHeaderValue)) { try HeaderValue.raw([]) }
+ #expect(throws: IggyError(.invalidHeaderValue)) { try HeaderValue.raw([UInt8](repeating: 1, count: 256)) }
+ #expect(throws: Never.self) { try HeaderValue.raw([UInt8](repeating: 1, count: 255)) }
+ }
+
+ @Test func fixedSizeKindsAreChecked() {
+ #expect(throws: IggyError.self) { try HeaderValue(kind: .uint32, bytes: [1, 2, 3]) }
+ #expect(throws: IggyError.self) { try HeaderKey(kind: .string, bytes: []) }
+ #expect(throws: IggyError.self) { try HeaderValue.string(String(repeating: "a", count: 256)) }
+ }
+}
+
+@Suite("Options block")
+struct OptionsBlockTests {
+ /// A newer server's option kind is skipped rather than failing the read.
+ @Test func unknownValueKindsAreSkippedOnDecode() throws {
+ var block = HeaderTLV.encode(["known": "v"])
+ block += [2, 7, 0, 0, 0] + Array("unknown".utf8) + [200, 1, 0, 0, 0, 9]
+ let options = try OptionsBlock.decode(block[...], explicit: true)
+ #expect(options == ["known": .explicit("v")])
+ }
+
+ /// A prefixed block whose declared length exceeds the buffer is reported
+ /// as truncation, even when the length would not fit a 32-bit `Int`.
+ @Test func prefixedBlocksAreBoundsChecked() {
+ var reader = ByteReader([0xFF, 0xFF, 0xFF, 0xFF, 1, 2])
+ #expect(throws: WireError.truncated(offset: 4, need: 4_294_967_295, have: 2)) { try OptionsBlock.readPrefixed(from: &reader) }
+ }
+
+ @Test func rejectsNonStringAndDuplicateKeys() throws {
+ let numericKey: [UInt8] = [12, 8, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 2, 1, 0, 0, 0, 118]
+ #expect(throws: WireError.self) { try OptionsBlock.validate(numericKey[...]) }
+ let entry = HeaderTLV.encode(["segment_size": "1 GiB"])
+ #expect(throws: WireError.self) { try OptionsBlock.validate((entry + entry)[...]) }
+ #expect(try OptionsBlock.validate(entry[...]).count == 1)
+ }
+
+ @Test func prefixedBlocksRoundTrip() throws {
+ let options: ResourceOptions = ["key": .explicit("value")]
+ let block = try OptionsBlock.encode(options)
+ var writer = ByteWriter()
+ writer.write(UInt32(block.count))
+ writer.write(block)
+ var reader = ByteReader(writer.bytes)
+ let decoded = try OptionsBlock.decode(try OptionsBlock.readPrefixed(from: &reader), explicit: true)
+ #expect(decoded == options)
+ #expect(reader.isAtEnd)
+ var truncated = ByteReader(Array(writer.bytes.dropLast()))
+ #expect(throws: WireError.self) { try OptionsBlock.readPrefixed(from: &truncated) }
+ }
+
+ @Test func derivedEntriesAreNotSentAndExplicitWinsOnMerge() throws {
+ let options: ResourceOptions = ["a": .explicit("1"), "b": .derived("2")]
+ let encoded = try OptionsBlock.encode(options)
+ #expect(try OptionsBlock.decode(encoded[...], explicit: true) == ["a": .explicit("1")])
+ let merged = try OptionsBlock.decodeSplit(explicit: encoded[...], derived: try OptionsBlock.encode(["a": .explicit("9"), "c": .explicit("3")])[...])
+ #expect(merged == ["a": .explicit("1"), "c": .derived("3")])
+ }
+
+ @Test func limitsAreEnforced() throws {
+ var many: ResourceOptions = [:]
+ for index in 0...OptionsBlock.maxOptions {
+ many["key_\(index)"] = .explicit("v")
+ }
+ #expect(throws: IggyError(.optionsBlockTooLarge, context: "\(OptionsBlock.maxOptions + 1) entries, maximum \(OptionsBlock.maxOptions)")) {
+ try OptionsBlock.encode(many)
+ }
+ var big: ResourceOptions = [:]
+ let value = try HeaderValue.string(String(repeating: "v", count: 255))
+ for index in 0..<400 {
+ big[String(repeating: "k", count: 250) + String(format: "%05d", index)] = .explicit(value)
+ }
+ #expect(throws: IggyError.self) { try OptionsBlock.encode(big) }
+ }
+}
+
+@Suite("Identifiers")
+struct IdentifierTests {
+ @Test func literalsAndParsing() throws {
+ let numeric: Identifier = 7
+ let named: Identifier = "orders"
+ #expect(numeric.numericValue == 7)
+ #expect(named.name == "orders")
+ #expect(try Identifier(parsing: "12").numericValue == 12)
+ #expect(try Identifier(parsing: "abc").name == "abc")
+ #expect(throws: IggyError.self) { try Identifier(named: "") }
+ #expect(throws: IggyError.self) { try Identifier(named: String(repeating: "x", count: 256)) }
+ #expect("\(numeric) \(named)" == "7 orders")
+ }
+
+ /// A multi-byte name is length-prefixed by its UTF-8 byte count, not its
+ /// character count.
+ @Test func namesArePrefixedByByteCount() throws {
+ var writer = ByteWriter()
+ let named: Identifier = "café"
+ try named.encode(into: &writer)
+ #expect(writer.bytes == [2, 5, 0x63, 0x61, 0x66, 0xC3, 0xA9])
+ writer = ByteWriter()
+ try Identifier(numeric: UInt32.max).encode(into: &writer)
+ #expect(writer.bytes == [1, 4, 0xFF, 0xFF, 0xFF, 0xFF])
+ }
+}
+
+@Suite("Messages")
+struct IggyMessageTests {
+ struct Order: Codable, Equatable {
+ let id: Int
+ let item: String
+ }
+
+ @Test func textAndJSONPayloads() throws {
+ let text = try IggyMessage("hello", id: MessageID(low: 1, high: 0))
+ #expect(text.payloadString == "hello")
+ #expect(text.id == MessageID(low: 1, high: 0))
+ #expect(text.userHeaders == nil && text.rawUserHeaders == nil)
+
+ let order = Order(id: 7, item: "book")
+ let json = try IggyMessage(encoding: order)
+ #expect(try json.decode(Order.self) == order)
+ #expect(try IggyMessage(payload: [0xFF]).payloadString == "\u{FFFD}")
+ }
+
+ /// A received message keeps headers it cannot interpret as raw bytes and
+ /// reports no decoded headers, so one unknown kind does not fail a poll.
+ @Test func undecodableHeadersStayRaw() {
+ let unknownKind: [UInt8] = [2, 1, 0, 0, 0, UInt8(ascii: "k"), 200, 1, 0, 0, 0, 1]
+ let received = IggyMessage(id: .zero, offset: 3, timestamp: .zero, originTimestamp: .zero, checksum: 0, payload: [1], rawUserHeaders: unknownKind)
+ #expect(received.rawUserHeaders == unknownKind)
+ #expect(received.userHeaders == nil)
+ #expect(received[header: "k"] == nil)
+ let known = IggyMessage(
+ id: .zero, offset: 3, timestamp: .zero, originTimestamp: .zero, checksum: 0, payload: [1], rawUserHeaders: HeaderTLV.encode(["k": "v"]))
+ #expect(known.userHeaders == ["k": "v"])
+ #expect(known[header: "k"] == "v")
+ let empty = IggyMessage(id: .zero, offset: 0, timestamp: .zero, originTimestamp: .zero, checksum: 0, payload: [1], rawUserHeaders: [])
+ #expect(empty.rawUserHeaders == nil)
+ }
+
+ @Test func descriptionShowsABoundedPreview() throws {
+ let short = try IggyMessage("hello")
+ #expect(short.description == "[0] ID:0 'hello'")
+ // 46 ASCII bytes then a two-byte "é" straddling the 47-byte cut.
+ let straddling = try IggyMessage(String(repeating: "a", count: 46) + "éé" + String(repeating: "b", count: 20))
+ #expect(straddling.description == "[0] ID:0 '\(String(repeating: "a", count: 46))... (70B)'")
+ let binary = try IggyMessage(payload: [UInt8](repeating: 0xFF, count: 100))
+ #expect(binary.description == "[0] ID:0 <binary 100B>")
+ }
+
+ @Test func headersTravelEncodedAndDecodeBack() throws {
+ let headers: UserHeaders = ["trace-id": "abc", "attempt": .uint32(3)]
+ let message = try IggyMessage("hello", userHeaders: headers)
+ #expect(message.rawUserHeaders == HeaderTLV.encode(headers))
+ #expect(message.userHeaders == headers)
+ #expect(try IggyMessage("hello", userHeaders: [:]).rawUserHeaders == nil)
+ }
+
+ @Test func limitsAreEnforced() {
+ #expect(throws: IggyError(.invalidMessagePayloadLength)) { try IggyMessage(payload: []) }
+ #expect(throws: IggyError(.tooBigMessagePayload)) {
+ try IggyMessage(payload: [UInt8](repeating: 0, count: IggyMessage.maxPayloadSize + 1))
+ }
+ }
+}
+
+@Suite("Partitioning")
+struct PartitioningTests {
+ func encode(_ partitioning: Partitioning) throws -> [UInt8] {
+ var writer = ByteWriter()
+ try partitioning.encode(into: &writer)
+ return writer.bytes
+ }
+
+ @Test func messagesKeyMustFitTheLengthByte() throws {
+ #expect(try encode(.messagesKey([UInt8](repeating: 7, count: 255))).count == 257)
+ #expect(throws: IggyError(.invalidKeyValueLength)) { try encode(.messagesKey([])) }
+ #expect(throws: IggyError(.invalidKeyValueLength)) { try encode(.messagesKey([UInt8](repeating: 7, count: 256))) }
+ #expect(throws: IggyError(.invalidKeyValueLength)) { try encode(.messagesKey(String(repeating: "k", count: 300))) }
+ }
+
+ /// The key width is part of the label because it is part of the key.
+ @Test func integerKeysCarryTheirWidth() throws {
+ #expect(try encode(.messagesKey(uint32: 42)) == [3, 4, 42, 0, 0, 0])
+ #expect(try encode(.messagesKey(uint64: 42)) == [3, 8, 42, 0, 0, 0, 0, 0, 0, 0])
+ #expect(try encode(.messagesKey(uint128: UInt128Value(low: 42, high: 0))).count == 18)
+ #expect(try encode(.balanced) == [1, 0])
+ #expect(try encode(.partition(9)) == [2, 4, 9, 0, 0, 0])
+ }
+
+ /// The poll sentinel is compared against the value the Rust generator
+ /// dumped from `core/common`, so the golden lane catches drift.
+ @Test func noAssignedPartitionMatchesTheServerConstant() {
+ #expect(PolledMessages.noAssignedPartition == GoldenFixture.shared.noAssignedPartition)
+ #expect(GoldenFixture.shared.resyncRequiredPartition != GoldenFixture.shared.noAssignedPartition)
+ }
+
+ /// Offsets are shared with the Rust, Go, Node and C# defaults only if the
+ /// default consumer id matches theirs.
+ @Test func defaultConsumerIsIdZero() throws {
+ #expect(Consumer.default == .consumer(0))
+ var writer = ByteWriter()
+ try Consumer.default.encode(into: &writer)
+ #expect(writer.bytes == [1, 1, 4, 0, 0, 0, 0])
+ }
+
+ @Test func timestampsFromDatesClamp() {
+ #expect(IggyTimestamp(date: Date(timeIntervalSince1970: -5)).microseconds == 0)
+ #expect(IggyTimestamp(date: Date(timeIntervalSince1970: 1.5)).microseconds == 1_500_000)
+ #expect(IggyTimestamp(date: Date(timeIntervalSince1970: 1e15)).microseconds == UInt64.max)
+ #expect(IggyTimestamp(date: Date(timeIntervalSince1970: .infinity)).microseconds == UInt64.max)
+ #expect(IggyTimestamp(date: Date(timeIntervalSince1970: .nan)).microseconds == 0)
+ }
+}
+
+@Suite("Permissions")
+struct PermissionsTests {
+ @Test func roundTripWithNestedTopics() throws {
+ let permissions = Permissions(
+ global: .all,
+ streams: [
+ 3: StreamPermissions(readStream: true, topics: [1: TopicPermissions(readTopic: true), 2: TopicPermissions(sendMessages: true)]),
+ 1: StreamPermissions(manageStream: true),
+ ])
+ let encoded = permissions.encode()
+ var reader = ByteReader(encoded)
+ #expect(try Permissions.decode(from: &reader) == permissions)
+ #expect(reader.isAtEnd)
+ for cut in 0..<encoded.count {
+ var truncated = ByteReader(Array(encoded[..<cut]))
+ #expect(throws: WireError.self) { try Permissions.decode(from: &truncated) }
+ }
+ }
+
+ @Test func globalOnlyIsElevenBytes() {
+ #expect(Permissions(global: .all).encode().count == 11)
+ }
+}
+
+@Suite("Topic options")
+struct TopicOptionsTests {
+ @Test func durabilityIsAlwaysSentAndRawMayStrengthenIt() throws {
+ let defaults = try TopicCreateOptions().toResourceOptions()
+ #expect(defaults[TopicOptionKey.durability] == .explicit("replicated"))
+ #expect(defaults[TopicOptionKey.consumerOffsetDurability] == .explicit("replicated"))
+ let raw = try TopicCreateOptions(raw: [TopicOptionKey.durability: "persisted"]).toResourceOptions()
+ #expect(raw[TopicOptionKey.durability] == .explicit("persisted"))
+ // The server parses the policy case-insensitively; the canonical
+ // spelling is what travels.
+ let spelled = try TopicCreateOptions(raw: [TopicOptionKey.durability: "Persisted"]).toResourceOptions()
+ #expect(spelled[TopicOptionKey.durability] == .explicit("persisted"))
+ #expect(throws: IggyError.self) {
+ try TopicCreateOptions(durability: .persisted, raw: [TopicOptionKey.durability: "replicated"]).toResourceOptions()
+ }
+ #expect(throws: IggyError.self) {
+ try TopicCreateOptions(raw: [TopicOptionKey.durability: "sometimes"]).toResourceOptions()
+ }
+ }
+
+ /// Zero and negative expiries would travel as 0, which the server reads
+ /// as its default retention, so they are refused before encoding.
+ @Test func expiryMustBePositive() throws {
+ #expect(throws: IggyError(.invalidOptionValue)) { try TopicCreateOptions(messageExpiry: .after(.seconds(-5))).encode() }
+ #expect(throws: IggyError(.invalidOptionValue)) { try TopicCreateOptions(messageExpiry: .after(.zero)).encode() }
+ #expect(throws: IggyError(.invalidOptionValue)) { try TopicUpdateOptions(messageExpiry: .after(.milliseconds(-1))).encode() }
+ #expect(throws: IggyError(.invalidOptionValue)) { try TopicCreateOptions(messageExpiry: .after(.nanoseconds(500))).encode() }
+ #expect(throws: IggyError(.invalidOptionValue)) { try TopicCreateOptions(messageExpiry: .after(.seconds(1 << 58))).encode() }
+ let options = try TopicCreateOptions(messageExpiry: .after(.seconds(1))).toResourceOptions()
+ #expect(options[TopicOptionKey.messageExpiry] == .explicit(.uint64(1_000_000)))
+ #expect(try TopicCreateOptions(messageExpiry: .after(.microseconds(1))).toResourceOptions()[TopicOptionKey.messageExpiry] == .explicit(.uint64(1)))
+ #expect(try TopicCreateOptions(messageExpiry: .never).toResourceOptions()[TopicOptionKey.messageExpiry] == .explicit(.uint64(UInt64.max)))
+ #expect(try TopicCreateOptions(messageExpiry: .serverDefault).toResourceOptions()[TopicOptionKey.messageExpiry] == nil)
+ }
+}
diff --git a/foreign/swift/Tests/IggyTests/ModelVectorTests.swift b/foreign/swift/Tests/IggyTests/ModelVectorTests.swift
new file mode 100644
index 000000000..5b2496b99
--- /dev/null
+++ b/foreign/swift/Tests/IggyTests/ModelVectorTests.swift
@@ -0,0 +1,116 @@
+// 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
+
+/// Every model encoder is compared against bytes the Rust protocol crate
+/// produced for the same values, so the Swift SDK is checked against the
+/// reference rather than against itself.
+@Suite("Model encoders match the Rust golden vectors")
+struct ModelVectorTests {
+ let golden = GoldenFixture.shared
+
+ func encode(_ id: Identifier) throws -> [UInt8] {
+ var writer = ByteWriter()
+ try id.encode(into: &writer)
+ #expect(writer.bytes.count == id.encodedSize)
+ return writer.bytes
+ }
+
+ func encode(_ consumer: Consumer) throws -> [UInt8] {
+ var writer = ByteWriter()
+ try consumer.encode(into: &writer)
+ return writer.bytes
+ }
+
+ func encode(_ partitioning: Partitioning) throws -> [UInt8] {
+ var writer = ByteWriter()
+ try partitioning.encode(into: &writer)
+ #expect(writer.bytes.count == partitioning.encodedSize)
+ return writer.bytes
+ }
+
+ func encode(_ strategy: PollingStrategy) -> [UInt8] {
+ var writer = ByteWriter()
+ strategy.encode(into: &writer)
+ return writer.bytes
+ }
+
+ @Test func identifiers() throws {
+ #expect(try encode(1) == golden.bytes("identifier.numeric.1"))
+ #expect(try encode("my-stream") == golden.bytes("identifier.named.my-stream"))
+ }
+
+ @Test func consumers() throws {
+ #expect(try encode(.consumer(42)) == golden.bytes("consumer.numeric.42"))
+ #expect(try encode(.group("my-group")) == golden.bytes("consumer_group.named.my-group"))
+ }
+
+ @Test func partitioning() throws {
+ #expect(try encode(.balanced) == golden.bytes("partitioning.balanced"))
+ #expect(try encode(.partition(7)) == golden.bytes("partitioning.partition_id.7"))
+ #expect(try encode(.messagesKey("user-123")) == golden.bytes("partitioning.messages_key.user-123"))
+ }
+
+ @Test func pollingStrategies() {
+ #expect(encode(.offset(100)) == golden.bytes("polling.offset.100"))
+ #expect(encode(.timestamp(IggyTimestamp(microseconds: 1_700_000_000_000))) == golden.bytes("polling.timestamp.1700000000000"))
+ #expect(encode(.first) == golden.bytes("polling.first"))
+ #expect(encode(.last) == golden.bytes("polling.last"))
+ #expect(encode(.next) == golden.bytes("polling.next"))
+ }
+
+ @Test func userHeaders() {
+ let headers: UserHeaders = ["trace-id": "abc", "attempt": .uint32(3)]
+ #expect(HeaderTLV.encode(headers) == golden.bytes("user_headers.sample"))
+ }
+
+ @Test func resourceOptions() throws {
+ let options: ResourceOptions = ["durability": .explicit("persisted"), "segment_size": .explicit(.uint64(1_073_741_824))]
+ #expect(try OptionsBlock.encode(options) == golden.bytes("options.golden"))
+ }
+
+ @Test func topicOptions() throws {
+ let create = TopicCreateOptions(
+ partitionsCount: 3, compressionAlgorithm: .gzip, messageExpiry: .after(.seconds(604_800)),
+ maxTopicSize: .bytes(1_073_741_824), segmentSizeBytes: 1_048_576, durability: .persisted,
+ consumerOffsetDurability: .replicated, messagesRequiredToSave: 7, sizeOfMessagesRequiredToSaveBytes: 4096,
+ preallocateSegments: false)
+ #expect(try create.encode() == golden.bytes("options.topic_create.full"))
+ let update = TopicUpdateOptions(compressionAlgorithm: CompressionAlgorithm.none, messageExpiry: .never, maxTopicSize: .unlimited)
+ #expect(try update.encode() == golden.bytes("options.topic_update.full"))
+ let raw = TopicCreateOptions(raw: ["durability": "persisted"])
+ #expect(try raw.encode() == golden.bytes("options.topic_create.raw"))
+ }
+
+ @Test func permissions() {
+ let permissions = Permissions(
+ global: GlobalPermissions(manageServers: true, readServers: true, readUsers: true, readStreams: true, readTopics: true, pollMessages: true),
+ streams: [
+ 1: StreamPermissions(
+ manageStream: true, readStream: true, readTopics: true, pollMessages: true,
+ topics: [
+ 10: TopicPermissions(manageTopic: true, readTopic: true, sendMessages: true),
+ 20: TopicPermissions(readTopic: true, pollMessages: true, sendMessages: true),
+ ]),
+ 2: StreamPermissions(readStream: true, manageTopics: true, sendMessages: true),
+ ])
+ #expect(permissions.encode() == golden.bytes("permissions.sample"))
+ }
+}
diff --git a/foreign/swift/Tests/IggyTests/Support/GoldenFixture.swift b/foreign/swift/Tests/IggyTests/Support/GoldenFixture.swift
index a68de70eb..898d2e069 100644
--- a/foreign/swift/Tests/IggyTests/Support/GoldenFixture.swift
+++ b/foreign/swift/Tests/IggyTests/Support/GoldenFixture.swift
@@ -35,9 +35,15 @@ struct GoldenFixture: Decodable {
let name: String
}
+ /// `NO_ASSIGNED_PARTITION` and `RESYNC_REQUIRED_PARTITION_SENTINEL` from
+ /// `core/common`.
+ let noAssignedPartition: UInt32
+ let resyncRequiredPartition: UInt32
private let xxh3_64Table: [String: String]
private let xxh32Table: [String: String]
private let errorTable: [String: String]
+ /// Encoded values and request and reply bodies by name, as hex.
+ let vectors: [String: String]
/// The XXH3-64 vectors, by input length.
var xxh3_64: [HashVector] { Self.hashVectors(xxh3_64Table) }
@@ -55,9 +61,12 @@ struct GoldenFixture: Decodable {
}
enum CodingKeys: String, CodingKey {
+ case noAssignedPartition = "no_assigned_partition"
+ case resyncRequiredPartition = "resync_required_partition"
case xxh3_64Table = "xxh3_64"
case xxh32Table = "xxh32"
case errorTable = "errors"
+ case vectors
}
static let shared: GoldenFixture = {
@@ -70,4 +79,27 @@ struct GoldenFixture: Decodable {
static func pattern(_ length: Int) -> [UInt8] {
(0..<length).map { UInt8(truncatingIfNeeded: $0) &* 31 &+ 7 }
}
+
+ /// Bytes of a named vector; fails the test when the name is unknown.
+ func bytes(_ name: String) -> [UInt8] {
+ guard let hex = vectors[name] else {
+ Issue.record("missing golden vector \(name)")
+ return []
+ }
+ return [UInt8](hex: hex)
+ }
+}
+
+extension [UInt8] {
+ init(hex: String) {
+ var bytes: [UInt8] = []
+ bytes.reserveCapacity(hex.count / 2)
+ var index = hex.startIndex
+ while index < hex.endIndex {
+ let next = hex.index(index, offsetBy: 2)
+ bytes.append(UInt8(hex[index..<next], radix: 16)!)
+ index = next
+ }
+ self = bytes
+ }
}
diff --git a/foreign/swift/Tools/golden-vectors/src/main.rs b/foreign/swift/Tools/golden-vectors/src/main.rs
index b599ae9d4..baec2d104 100644
--- a/foreign/swift/Tools/golden-vectors/src/main.rs
+++ b/foreign/swift/Tools/golden-vectors/src/main.rs
@@ -90,7 +90,8 @@ use iggy_binary_protocol::{
};
use iggy_common::{
CompressionAlgorithm, Durability, IggyByteSize, IggyDuration, IggyError, IggyExpiry,
- MaxTopicSize, TopicCreateOptions, TopicUpdateOptions,
+ MaxTopicSize, NO_ASSIGNED_PARTITION, RESYNC_REQUIRED_PARTITION_SENTINEL, TopicCreateOptions,
+ TopicUpdateOptions,
};
use secrecy::SecretString;
use serde::Serialize;
@@ -103,6 +104,8 @@ use twox_hash::{XxHash3_64, XxHash32};
struct Golden {
protocol_version: u32,
protocol_version_min: u32,
+ no_assigned_partition: u32,
+ resync_required_partition: u32,
xxh3_64: BTreeMap<usize, String>,
xxh32: BTreeMap<usize, String>,
errors: BTreeMap<u32, String>,
@@ -1414,6 +1417,8 @@ fn main() {
let golden = Golden {
protocol_version: IGGY_PROTOCOL_VERSION,
protocol_version_min: IGGY_PROTOCOL_VERSION_MIN,
+ no_assigned_partition: NO_ASSIGNED_PARTITION,
+ resync_required_partition: RESYNC_REQUIRED_PARTITION_SENTINEL,
xxh3_64,
xxh32,
errors,