Dedicated KafkaProducerMessageID type (#71)

* Dedicated `KafkaProducerMessageID` type

> **Warning**: this can potentially be a breaking change as message ids
> will lose their chronological order in the public API after this
> change.

Motivation:

We don't want to expose `UInt` for `KafkaProducer`'s message ids.

Modifications:

* create a new `struct` `KafkaProducerMessageID` that is internally
  backed by `UInt` (can also be `UUID` or similar in the future
* update tests

* Review Franz

Modifications:

* `KafkaProducerMessageID`: add `Comparable` conformance

Author: felixschlegel

Sources/SwiftKafka/KafkaAcknowledgedMessage.swift

@@ -20,7 +20,7 @@ public struct KafkaAcknowledgedMessage: Hashable {
     /// The unique identifier assigned by the ``KafkaProducer`` when the message was send to Kafka.
     /// The same identifier is returned by ``KafkaProducer/sendAsync(_:)`` and can be used to correlate
     /// a sent message and an acknowledged message.
-    public var id: UInt
+    public var id: KafkaProducerMessageID
     /// The topic that the message was sent to.
     public var topic: String
     /// The partition that the message was sent to.
@@ -34,7 +34,7 @@ public struct KafkaAcknowledgedMessage: Hashable {
 
     /// Initialize ``KafkaAcknowledgedMessage`` from `rd_kafka_message_t` pointer.
     /// - Throws: A ``KafkaAcknowledgedMessageError`` for failed acknowledgements or malformed messages.
-    init(messagePointer: UnsafePointer<rd_kafka_message_t>, id: UInt) throws {
+    init(messagePointer: UnsafePointer<rd_kafka_message_t>, id: KafkaProducerMessageID) throws {
         self.id = id
 
         let rdKafkaMessage = messagePointer.pointee

Sources/SwiftKafka/KafkaAcknowledgedMessageError.swift

@@ -20,11 +20,11 @@ import Crdkafka
 /// ``KafkaError/code``.
 public struct KafkaAcknowledgedMessageError: Error, CustomStringConvertible {
     /// Identifier of the message that caused the error.
-    public var messageID: UInt
+    public var messageID: KafkaProducerMessageID
     /// The underlying ``KafkaError``.
     public let error: KafkaError
 
-    init(messageID: UInt, error: KafkaError) {
+    init(messageID: KafkaProducerMessageID, error: KafkaError) {
         self.messageID = messageID
         self.error = error
     }
@@ -34,7 +34,7 @@ public struct KafkaAcknowledgedMessageError: Error, CustomStringConvertible {
     }
 
     static func fromRDKafkaError(
-        messageID: UInt,
+        messageID: KafkaProducerMessageID,
         error: rd_kafka_resp_err_t,
         file: String = #fileID,
         line: UInt = #line
@@ -50,7 +50,7 @@ public struct KafkaAcknowledgedMessageError: Error, CustomStringConvertible {
     }
 
     static func fromMessage(
-        messageID: UInt,
+        messageID: KafkaProducerMessageID,
         message: String,
         file: String = #fileID,
         line: UInt = #line

Sources/SwiftKafka/KafkaProducer.swift

@@ -217,10 +217,11 @@ public actor KafkaProducer {
     /// Send messages to the Kafka cluster asynchronously, aka "fire and forget".
     /// This function is non-blocking.
     /// - Parameter message: The ``KafkaProducerMessage`` that is sent to the KafkaCluster.
-    /// - Returns: Unique message identifier matching the `id` property of the corresponding ``KafkaAcknowledgedMessage``
+    /// - Returns: Unique ``KafkaProducerMessageID``matching the ``KafkaAcknowledgedMessage/id`` property
+    /// of the corresponding ``KafkaAcknowledgedMessage``.
     /// - Throws: A ``KafkaError`` if sending the message failed.
     @discardableResult
-    public func sendAsync(_ message: KafkaProducerMessage) throws -> UInt {
+    public func sendAsync(_ message: KafkaProducerMessage) throws -> KafkaProducerMessageID {
         switch self.state {
         case .started:
             return try self._sendAsync(message)
@@ -229,7 +230,7 @@ public actor KafkaProducer {
         }
     }
 
-    private func _sendAsync(_ message: KafkaProducerMessage) throws -> UInt {
+    private func _sendAsync(_ message: KafkaProducerMessage) throws -> KafkaProducerMessageID {
         let topicHandle = try self.createTopicHandleIfNeeded(topic: message.topic)
 
         let keyBytes: [UInt8]?
@@ -261,7 +262,7 @@ public actor KafkaProducer {
             throw KafkaError.rdKafkaError(wrapping: rd_kafka_last_error())
         }
 
-        return self.messageIDCounter
+        return KafkaProducerMessageID(rawValue: self.messageIDCounter)
     }
 
     /// Check `topicHandles` for a handle matching the topic name and create a new handle if needed.

Sources/SwiftKafka/KafkaProducerMessageID.swift

@@ -0,0 +1,44 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the swift-kafka-gsoc open source project
+//
+// Copyright (c) 2022 Apple Inc. and the swift-kafka-gsoc project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of swift-kafka-gsoc project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+/// ID of message produced by the ``KafkaProducer``.
+/// The ``KafkaProducerMessageID`` is relates incoming ``KafkaAcknowledgedMessage``'s
+/// with their corresponding ``KafkaProducer/send(_:)`` invocation.
+public struct KafkaProducerMessageID {
+    internal var rawValue: UInt
+
+    internal init(rawValue: UInt) {
+        self.rawValue = rawValue
+    }
+}
+
+// MARK: - KafkaProducerMessageID + CustomStringConvertible
+
+extension KafkaProducerMessageID: CustomStringConvertible {
+    public var description: String {
+        return String(self.rawValue)
+    }
+}
+
+// MARK: - KafkaProducerMessageID + Hashable
+
+extension KafkaProducerMessageID: Hashable {}
+
+// MARK: - KafkaProducerMessageID + Comparable
+
+extension KafkaProducerMessageID: Comparable {
+    public static func < (lhs: KafkaProducerMessageID, rhs: KafkaProducerMessageID) -> Bool {
+        lhs.rawValue < rhs.rawValue
+    }
+}

Sources/SwiftKafka/RDKafka/RDKafkaConfig.swift

@@ -115,7 +115,7 @@ struct RDKafkaConfig {
             return nil
         }
 
-        let messageID = UInt(bitPattern: messagePointer.pointee._private)
+        let messageID = KafkaProducerMessageID(rawValue: UInt(bitPattern: messagePointer.pointee._private))
 
         let messageResult: KafkaAcknowledgementResult
         do {

Tests/IntegrationTests/SwiftKafkaTests.swift

@@ -270,7 +270,7 @@ final class SwiftKafkaTests: XCTestCase {
         acknowledgements: KafkaMessageAcknowledgements,
         messages: [KafkaProducerMessage]
     ) async throws {
-        var messageIDs = Set<UInt>()
+        var messageIDs = Set<KafkaProducerMessageID>()
 
         for message in messages {
             messageIDs.insert(try await producer.sendAsync(message))
@@ -292,7 +292,7 @@ final class SwiftKafkaTests: XCTestCase {
         }
 
         XCTAssertEqual(messages.count, acknowledgedMessages.count)
-        XCTAssertEqual(acknowledgedMessages.map(\.id).sorted(), messageIDs.sorted())
+        XCTAssertEqual(Set(acknowledgedMessages.map(\.id)), messageIDs)
 
         for message in messages {
             XCTAssertTrue(acknowledgedMessages.contains(where: { $0.topic == message.topic }))

Tests/SwiftKafkaTests/KafkaProducerTests.swift

@@ -150,7 +150,7 @@ final class KafkaProducerTests: XCTestCase {
                     value: "Hello, London!"
                 )
 
-                var messageIDs = Set<UInt>()
+                var messageIDs = Set<KafkaProducerMessageID>()
 
                 messageIDs.insert(try await producer.sendAsync(message1))
                 messageIDs.insert(try await producer.sendAsync(message2))
@@ -171,7 +171,7 @@ final class KafkaProducerTests: XCTestCase {
                 }
 
                 XCTAssertEqual(2, acknowledgedMessages.count)
-                XCTAssertEqual(acknowledgedMessages.map(\.id).sorted(), messageIDs.sorted())
+                XCTAssertEqual(Set(acknowledgedMessages.map(\.id)), messageIDs)
                 XCTAssertTrue(acknowledgedMessages.contains(where: { $0.topic == message1.topic }))
                 XCTAssertTrue(acknowledgedMessages.contains(where: { $0.topic == message2.topic }))
                 XCTAssertTrue(acknowledgedMessages.contains(where: { $0.key == message1.key }))