@@ -93,6 +93,88 @@ final class KafkaClient {
9393 }
9494 }
9595
96+ /// Wraps a Swift closure inside of a class to be able to pass it to `librdkafka` as an `OpaquePointer`.
97+ /// This is specifically used to pass a Swift closure as a commit callback for the ``KafkaConsumer``.
98+ final class CapturedCommitCallback {
99+ typealias Closure = ( Result < Void , KafkaError > ) -> Void
100+ let closure : Closure
101+
102+ init ( _ closure: @escaping Closure ) {
103+ self . closure = closure
104+ }
105+ }
106+
107+ /// Non-blocking commit of a the `message`'s offset to Kafka.
108+ ///
109+ /// - Parameter message: Last received message that shall be marked as read.
110+ func commitSync( _ message: KafkaConsumerMessage ) async throws {
111+ // Declare captured closure outside of withCheckedContinuation.
112+ // We do that because do an unretained pass of the captured closure to
113+ // librdkafka which means we have to keep a reference to the closure
114+ // ourselves to make sure it does not get deallocated before
115+ // commitSync returns.
116+ var capturedClosure : CapturedCommitCallback !
117+ try await withCheckedThrowingContinuation { continuation in
118+ capturedClosure = CapturedCommitCallback { result in
119+ continuation. resume ( with: result)
120+ }
121+
122+ let changesList = rd_kafka_topic_partition_list_new ( 1 )
123+ defer { rd_kafka_topic_partition_list_destroy ( changesList) }
124+ guard let partitionPointer = rd_kafka_topic_partition_list_add (
125+ changesList,
126+ message. topic,
127+ message. partition. rawValue
128+ ) else {
129+ fatalError ( " rd_kafka_topic_partition_list_add returned invalid pointer " )
130+ }
131+
132+ // Unretained pass because the reference that librdkafka holds to capturedClosure
133+ // should not be counted in ARC as this can lead to memory leaks.
134+ let opaquePointer : UnsafeMutableRawPointer ? = Unmanaged . passUnretained ( capturedClosure) . toOpaque ( )
135+
136+ // The offset committed is always the offset of the next requested message.
137+ // Thus, we increase the offset of the current message by one before committing it.
138+ // See: https://github.com/edenhill/librdkafka/issues/2745#issuecomment-598067945
139+ partitionPointer. pointee. offset = Int64 ( message. offset + 1 )
140+
141+ let consumerQueue = rd_kafka_queue_get_consumer ( self . kafkaHandle)
142+
143+ // Create a C closure that calls the captured closure
144+ let callbackWrapper : (
145+ @convention ( c) (
146+ OpaquePointer ? ,
147+ rd_kafka_resp_err_t ,
148+ UnsafeMutablePointer < rd_kafka_topic_partition_list_t > ? ,
149+ UnsafeMutableRawPointer ?
150+ ) -> Void
151+ ) = { _, error, _, opaquePointer in
152+
153+ guard let opaquePointer = opaquePointer else {
154+ fatalError ( " Could not resolve reference to catpured Swift callback instance " )
155+ }
156+ let opaque = Unmanaged < CapturedCommitCallback > . fromOpaque ( opaquePointer) . takeUnretainedValue ( )
157+
158+ let actualCallback = opaque. closure
159+
160+ if error == RD_KAFKA_RESP_ERR_NO_ERROR {
161+ actualCallback ( . success( ( ) ) )
162+ } else {
163+ let kafkaError = KafkaError . rdKafkaError ( wrapping: error)
164+ actualCallback ( . failure( kafkaError) )
165+ }
166+ }
167+
168+ rd_kafka_commit_queue (
169+ self . kafkaHandle,
170+ changesList,
171+ consumerQueue,
172+ callbackWrapper,
173+ opaquePointer
174+ )
175+ }
176+ }
177+
96178 /// Scoped accessor that enables safe access to the pointer of the client's Kafka handle.
97179 /// - Warning: Do not escape the pointer from the closure for later use.
98180 /// - Parameter body: The closure will use the Kafka handle pointer.
0 commit comments