feat(swift): add the consumer

foreign/swift/README.md

@@ -14,8 +14,9 @@ The SDK is built up in stages. The client is a TCP connection with optional TLS
 speaks the consensus protocol, signs in, keeps the session alive with a heartbeat,
 reconnects, and follows the cluster leader. It manages streams, topics, partitions,
 segments, users, personal access tokens, and consumer groups, sends and polls messages,
-stores consumer offsets, and produces with a batching producer. The consumer, examples,
-and BDD scenarios follow in later changes.
+stores consumer offsets, produces with a batching producer, and consumes with a
+consumer that tracks and commits its position. Examples and BDD scenarios follow in
+later changes.
 
 Swift supports the consensus wire protocol and checks commitment proof during
 automatic retries. It does not yet expose retained-operation resolve/abandon APIs,
@@ -322,6 +323,40 @@ written in order. A topic addressed by name and by numeric id counts as two, and
 failed request does not hold back the ones after it, so a resend from `errorHandler`
 lands behind them.
 
+### Consumer
+
+`IggyConsumer` reads one topic and hands its messages over one at a time. A standalone
+consumer reads one partition; a member of a consumer group reads the partitions the
+server assigns, joining the group in `initialize()`:
+
+```swift
+let consumer = try client.consumerGroup(name: "billing", streamID: "orders", topicID: "created")
+try await consumer.initialize()
+for try await received in consumer {
+    print(received.partitionID, received.message.offset, received.message.payloadString)
+}
+```
+
+A poll refills the buffer once it runs empty. `pollingStrategy` says where reading a
+partition starts, and each later poll of it continues after the last message handed over
+from it. The exception is `.next` with an `autoCommit` that commits with every poll, as in
+the default configuration. That poll starts after the offset that the previous poll
+stored. A failed poll is thrown after the retry
+interval and leaves the consumer usable, so a caller of `next()` can call it again, and
+`consumeMessages(_:)` logs it and keeps reading until the client cannot come back on its
+own. Without `allowReplay` a message at or below the last one handed over from the same
+partition history is skipped.
+
+`autoCommit` decides when the position is stored. The default commits with every poll,
+before the batch is handed over, and on a one-second interval; `.disabled` leaves it to
+`storePosition(_:)` with the position of a `ReceivedMessage`, the only setting under which a
+message whose handling failed comes back on the next run. `shutdown()` sends the queued
+commits, stores the reading position of every partition read unless auto-commit is
+disabled, and leaves the group.
+
+Unlike the Rust consumer this one has no encryptor, and it does not resolve a group create
+or join whose reply was lost, since the Swift client has no operation lanes yet.
+
 ### Raw requests
 
 `sendRawRequest(code:payload:)` sends a command code with an encoded payload and returns the raw reply body. `sendRawRequest(code:payload:replica:)` sends it to one node of a cluster, named by the replica id that cluster metadata returns.

foreign/swift/Sources/Iggy/Client/ClientCore.swift

@@ -383,6 +383,19 @@ actor ClientCore {
         try VSRFrame.encodeRequest(session: &session, code: code, payload: payload, partitionContext: context)
     }
 
+    /// Whether the client comes back on its own: it is signed in, or a
+    /// request or the heartbeat reconnects and signs it in again. A client
+    /// that is shut down, that the caller disconnected, or that has no
+    /// reconnection or no credentials only comes back when the caller acts.
+    var canRecover: Bool {
+        switch state {
+        case .shutdown: false
+        case .authenticated: true
+        case .disconnected, .connecting, .connected, .authenticating:
+            !disconnectedByCaller && configuration.reconnection.enabled && signInCredentials() != nil
+        }
+    }
+
     /// The refusal of a call that started under `captured` once the caller
     /// signed in or out since: nothing of it may be written any more.
     func lineageRefusal(_ captured: UInt64, code: UInt32? = nil) -> IggyError? {

foreign/swift/Sources/Iggy/Client/IggyClient.swift

@@ -267,6 +267,11 @@ public final class IggyClient: Sendable {
         return captured
     }
 
+    /// Whether the client comes back on its own after an outage.
+    var canRecover: Bool {
+        get async { await core.canRecover }
+    }
+
     /// The lineage a call that starts now runs under.
     var operationLineage: UInt64 {
         get async { await core.lineage }

foreign/swift/Sources/Iggy/Consumer/ConsumerConfiguration.swift

@@ -0,0 +1,201 @@
+// 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.
+
+/// When a consumer stores its offsets on the server.
+public enum AutoCommit: Sendable, Hashable {
+    /// Never, not even on ``IggyConsumer/shutdown()``; store with
+    /// ``IggyConsumer/storePosition(_:)``.
+    case disabled
+    /// On every tick, the reading position of every partition read so far.
+    case interval(Duration)
+    /// On every tick, and as `when` says while messages are handed over.
+    case intervalOrWhen(Duration, AutoCommitWhen)
+    /// On every tick, and as `after` says once a handler of
+    /// ``IggyConsumer/consumeMessages(_:)`` returned.
+    case intervalOrAfter(Duration, AutoCommitAfter)
+    /// As `when` says while messages are handed over.
+    case when(AutoCommitWhen)
+    /// As `after` says once a handler of ``IggyConsumer/consumeMessages(_:)``
+    /// returned. Reading with ``IggyConsumer/next()`` never triggers it.
+    case after(AutoCommitAfter)
+
+    var interval: Duration? {
+        switch self {
+        case .interval(let interval), .intervalOrWhen(let interval, _), .intervalOrAfter(let interval, _): interval
+        case .disabled, .when, .after: nil
+        }
+    }
+
+    var whenTrigger: AutoCommitWhen? {
+        switch self {
+        case .intervalOrWhen(_, let when), .when(let when): when
+        case .disabled, .interval, .intervalOrAfter, .after: nil
+        }
+    }
+
+    var afterTrigger: AutoCommitAfter? {
+        switch self {
+        case .intervalOrAfter(_, let after), .after(let after): after
+        case .disabled, .interval, .intervalOrWhen, .when: nil
+        }
+    }
+}
+
+/// A commit triggered while messages are handed over, before the code that
+/// reads them has seen them.
+public enum AutoCommitWhen: Sendable, Hashable {
+    /// With the poll request itself: the server stores the offset of the
+    /// last message it returns, before any of the batch is handed over.
+    case pollingMessages
+    /// Queued when the buffer of the current batch runs empty.
+    case consumingAllMessages
+    /// Queued just before each message is handed over. Commits queued faster
+    /// than they are sent collapse into the latest one per partition.
+    case consumingEachMessage
+    /// Queued just before a message whose offset divides by `n` is handed
+    /// over. It tests the offset, not a count of messages. `n` must not be
+    /// zero.
+    case consumingEveryNthMessage(UInt32)
+}
+
+/// A commit triggered once a handler of ``IggyConsumer/consumeMessages(_:)``
+/// returned, whether it succeeded or threw.
+public enum AutoCommitAfter: Sendable, Hashable {
+    /// After the message whose offset is the partition's newest at poll
+    /// time.
+    case consumingAllMessages
+    /// After each message.
+    case consumingEachMessage
+    /// After a message whose offset divides by `n`. `n` must not be zero.
+    case consumingEveryNthMessage(UInt32)
+}
+
+/// Options of an ``IggyConsumer``, with the Rust SDK's defaults.
+public struct ConsumerConfiguration: Sendable {
+    /// Where reading a partition starts. After the first batch each poll
+    /// continues after the last message handed over from that partition,
+    /// except a ``PollingStrategy/next`` poll that commits with
+    /// ``AutoCommitWhen/pollingMessages``: it starts after the offset the
+    /// previous poll stored.
+    public var pollingStrategy: PollingStrategy
+    /// Messages asked for per poll.
+    public var batchLength: UInt32
+    /// Minimum gap between the starts of two polls. Nil polls as soon as the
+    /// buffer is empty.
+    public var pollInterval: Duration?
+    public var autoCommit: AutoCommit
+    /// Joins the group during ``IggyConsumer/initialize()`` and again before
+    /// a poll once the membership was lost. Off, a poll without a membership
+    /// fails with ``IggyErrorCode/consumerGroupMemberNotFound``.
+    public var autoJoinConsumerGroup: Bool
+    /// Creates the group when it is missing before joining it.
+    public var createConsumerGroupIfNotExists: Bool
+    /// The pause after a failed poll or join, and between checks while the
+    /// client is signed out.
+    public var pollingRetryInterval: Duration
+    /// Lookups of a missing stream or topic during
+    /// ``IggyConsumer/initialize()`` before it fails. Nil or zero looks once.
+    public var initRetries: UInt32?
+    public var initRetryInterval: Duration
+    /// Hands over a message again when a poll returns it again, as after a
+    /// rebalance or a store that moved back. Off, a message at or below the
+    /// last one handed over from the same partition history is skipped.
+    public var allowReplay: Bool
+    /// How long ``IggyConsumer/shutdown()`` waits for each background commit
+    /// task before it stops it.
+    public var offsetDrainTimeout: Duration
+
+    public init(
+        pollingStrategy: PollingStrategy = .next, batchLength: UInt32 = 1000, pollInterval: Duration? = nil,
+        autoCommit: AutoCommit = .intervalOrWhen(.seconds(1), .pollingMessages), autoJoinConsumerGroup: Bool = true,
+        createConsumerGroupIfNotExists: Bool = true, pollingRetryInterval: Duration = .seconds(1), initRetries: UInt32? = nil,
+        initRetryInterval: Duration = .seconds(1), allowReplay: Bool = false, offsetDrainTimeout: Duration = .seconds(5)
+    ) {
+        self.pollingStrategy = pollingStrategy
+        self.batchLength = batchLength
+        self.pollInterval = pollInterval
+        self.autoCommit = autoCommit
+        self.autoJoinConsumerGroup = autoJoinConsumerGroup
+        self.createConsumerGroupIfNotExists = createConsumerGroupIfNotExists
+        self.pollingRetryInterval = pollingRetryInterval
+        self.initRetries = initRetries
+        self.initRetryInterval = initRetryInterval
+        self.allowReplay = allowReplay
+        self.offsetDrainTimeout = offsetDrainTimeout
+    }
+
+    /// Refuses what the Rust SDK's types cannot hold: a zero batch, a zero
+    /// or negative retry or commit interval (`NonZeroIggyDuration`), a zero
+    /// `n`, and any duration past `UInt64.max` microseconds, which would
+    /// also trap in `Task.sleep`.
+    func validate() throws {
+        if batchLength == 0 {
+            throw IggyError(.invalidConfiguration, context: "batchLength must be at least 1")
+        }
+        try Self.validatePositive(pollingRetryInterval, named: "pollingRetryInterval")
+        try Self.validatePositive(initRetryInterval, named: "initRetryInterval")
+        if let interval = autoCommit.interval {
+            try Self.validatePositive(interval, named: "the auto-commit interval")
+        }
+        if let pollInterval {
+            try ProducerConfiguration.validate(pollInterval, named: "pollInterval")
+        }
+        try ProducerConfiguration.validate(offsetDrainTimeout, named: "offsetDrainTimeout")
+        if case .consumingEveryNthMessage(0)? = autoCommit.whenTrigger {
+            throw IggyError(.invalidConfiguration, context: "consumingEveryNthMessage needs n of at least 1")
+        }
+        if case .consumingEveryNthMessage(0)? = autoCommit.afterTrigger {
+            throw IggyError(.invalidConfiguration, context: "consumingEveryNthMessage needs n of at least 1")
+        }
+    }
+
+    private static func validatePositive(_ duration: Duration, named name: String) throws {
+        guard duration > .zero else {
+            throw IggyError(.invalidConfiguration, context: "\(name) must be positive, got \(duration)")
+        }
+        try ProducerConfiguration.validate(duration, named: name)
+    }
+}
+
+/// One message handed over by an ``IggyConsumer``.
+public struct ReceivedMessage: Sendable {
+    public var message: IggyMessage
+    /// The newest offset the partition had when the batch was polled, the
+    /// same for every message of the batch. Compared with the message's
+    /// offset it shows how far the consumer lags.
+    public var currentOffset: UInt64
+    /// The partition the message was read from. A group member reads
+    /// several.
+    public var partitionID: UInt32
+    /// The partition incarnation and owner the poll ran under.
+    public var context: PartitionContext
+
+    public init(message: IggyMessage, currentOffset: UInt64, partitionID: UInt32, context: PartitionContext) {
+        self.message = message
+        self.currentOffset = currentOffset
+        self.partitionID = partitionID
+        self.context = context
+    }
+
+    /// The checkpoint to store for this message with
+    /// ``IggyConsumer/storePosition(_:)``. It keeps the context that
+    /// delivered the message, so a store after a recreation or a reassignment
+    /// is refused instead of landing on the newer partition.
+    public var position: ConsumerPosition {
+        ConsumerPosition(partitionID: partitionID, offset: message.offset, context: context)
+    }
+}