|
| 1 | +//===----------------------------------------------------------------------===// |
| 2 | +// |
| 3 | +// This source file is part of the swift-kafka-gsoc open source project |
| 4 | +// |
| 5 | +// Copyright (c) 2022 Apple Inc. and the swift-kafka-gsoc project authors |
| 6 | +// Licensed under Apache License v2.0 |
| 7 | +// |
| 8 | +// See LICENSE.txt for license information |
| 9 | +// See CONTRIBUTORS.txt for the list of swift-kafka-gsoc project authors |
| 10 | +// |
| 11 | +// SPDX-License-Identifier: Apache-2.0 |
| 12 | +// |
| 13 | +//===----------------------------------------------------------------------===// |
| 14 | + |
| 15 | +internal protocol KafkaProducerSharedProperties: Sendable, Hashable { |
| 16 | + // MARK: - SwiftKafka-specific Config properties |
| 17 | + |
| 18 | + /// The time between two consecutive polls. |
| 19 | + /// Effectively controls the rate at which incoming events are consumed. |
| 20 | + /// Default: `.milliseconds(100)` |
| 21 | + var pollInterval: Duration { get } |
| 22 | + |
| 23 | + /// Maximum timeout for flushing outstanding produce requests when the ``KakfaProducer`` is shutting down. |
| 24 | + /// Default: `10000` |
| 25 | + var flushTimeoutMilliseconds: Int { get } |
| 26 | + |
| 27 | + // MARK: - Producer-specific Config Properties |
| 28 | + |
| 29 | + /// When set to true, the producer will ensure that messages are successfully produced exactly once and in the original produce order. The following configuration properties are adjusted automatically (if not modified by the user) when idempotence is enabled: max.in.flight.requests.per.connection=5 (must be less than or equal to 5), retries=INT32_MAX (must be greater than 0), acks=all, queuing.strategy=fifo. Producer instantation will fail if user-supplied configuration is incompatible. |
| 30 | + /// Default: `false` |
| 31 | + var enableIdempotence: Bool { get } |
| 32 | + |
| 33 | + /// Producer queue options. |
| 34 | + var queue: KafkaConfiguration.QueueOptions { get } |
| 35 | + |
| 36 | + /// How many times to retry sending a failing Message. Note: retrying may cause reordering unless enable.idempotence is set to true. |
| 37 | + /// Default: `2_147_483_647` |
| 38 | + var messageSendMaxRetries: Int { get } |
| 39 | + |
| 40 | + /// Allow automatic topic creation on the broker when producing to non-existent topics. |
| 41 | + /// The broker must also be configured with auto.create.topics.enable=true for this configuration to take effect. |
| 42 | + /// Default: `true` |
| 43 | + var allowAutoCreateTopics: Bool { get } |
| 44 | + |
| 45 | + // MARK: - Common Client Config Properties |
| 46 | + |
| 47 | + /// Client identifier. |
| 48 | + /// Default: `"rdkafka"` |
| 49 | + var clientID: String { get } |
| 50 | + |
| 51 | + /// Initial list of brokers. |
| 52 | + /// Default: `[]` |
| 53 | + var bootstrapServers: [KafkaConfiguration.Broker] { get } |
| 54 | + |
| 55 | + /// Message options. |
| 56 | + var message: KafkaConfiguration.MessageOptions { get } |
| 57 | + |
| 58 | + /// Maximum Kafka protocol response message size. This serves as a safety precaution to avoid memory exhaustion in case of protocol hickups. This value must be at least fetch.max.bytes + 512 to allow for protocol overhead; the value is adjusted automatically unless the configuration property is explicitly set. |
| 59 | + /// Default: `100_000_000` |
| 60 | + var receiveMessageMaxBytes: Int { get } |
| 61 | + |
| 62 | + /// Maximum number of in-flight requests per broker connection. This is a generic property applied to all broker communication, however it is primarily relevant to produce requests. In particular, note that other mechanisms limit the number of outstanding consumer fetch request per broker to one. |
| 63 | + /// Default: `1_000_000` |
| 64 | + var maxInFlightRequestsPerConnection: Int { get } |
| 65 | + |
| 66 | + /// Metadata cache max age. |
| 67 | + /// Default: `900_000` |
| 68 | + var metadataMaxAgeMilliseconds: Int { get } |
| 69 | + |
| 70 | + /// Topic metadata options. |
| 71 | + var topicMetadata: KafkaConfiguration.TopicMetadataOptions { get } |
| 72 | + |
| 73 | + /// Topic denylist. |
| 74 | + /// Default: `[]` |
| 75 | + var topicDenylist: [String] { get } |
| 76 | + |
| 77 | + /// Debug options. |
| 78 | + /// Default: `[]` |
| 79 | + var debug: [KafkaConfiguration.DebugOption] { get } |
| 80 | + |
| 81 | + /// Socket options. |
| 82 | + var socket: KafkaConfiguration.SocketOptions { get } |
| 83 | + |
| 84 | + /// Broker options. |
| 85 | + var broker: KafkaConfiguration.BrokerOptions { get } |
| 86 | + |
| 87 | + /// Reconnect options. |
| 88 | + var reconnect: KafkaConfiguration.ReconnectOptions { get } |
| 89 | + |
| 90 | + /// Security protocol to use (plaintext, ssl, sasl_plaintext, sasl_ssl). |
| 91 | + /// Default: `.plaintext` |
| 92 | + var securityProtocol: KafkaConfiguration.SecurityProtocol { get } |
| 93 | + |
| 94 | + var dictionary: [String: String] { get } |
| 95 | +} |
| 96 | + |
| 97 | +extension KafkaProducerSharedProperties { |
| 98 | + internal var sharedPropsDictionary: [String: String] { |
| 99 | + var resultDict: [String: String] = [:] |
| 100 | + |
| 101 | + resultDict["enable.idempotence"] = String(self.enableIdempotence) |
| 102 | + resultDict["queue.buffering.max.messages"] = String(self.queue.bufferingMaxMessages) |
| 103 | + resultDict["queue.buffering.max.kbytes"] = String(self.queue.bufferingMaxKBytes) |
| 104 | + resultDict["queue.buffering.max.ms"] = String(self.queue.bufferingMaxMilliseconds) |
| 105 | + resultDict["message.send.max.retries"] = String(self.messageSendMaxRetries) |
| 106 | + resultDict["allow.auto.create.topics"] = String(self.allowAutoCreateTopics) |
| 107 | + |
| 108 | + resultDict["client.id"] = self.clientID |
| 109 | + resultDict["bootstrap.servers"] = self.bootstrapServers.map(\.description).joined(separator: ",") |
| 110 | + resultDict["message.max.bytes"] = String(self.message.maxBytes) |
| 111 | + resultDict["message.copy.max.bytes"] = String(self.message.copyMaxBytes) |
| 112 | + resultDict["receive.message.max.bytes"] = String(self.receiveMessageMaxBytes) |
| 113 | + resultDict["max.in.flight.requests.per.connection"] = String(self.maxInFlightRequestsPerConnection) |
| 114 | + resultDict["metadata.max.age.ms"] = String(self.metadataMaxAgeMilliseconds) |
| 115 | + resultDict["topic.metadata.refresh.interval.ms"] = String(self.topicMetadata.refreshIntervalMilliseconds) |
| 116 | + resultDict["topic.metadata.refresh.fast.interval.ms"] = String(self.topicMetadata.refreshFastIntervalMilliseconds) |
| 117 | + resultDict["topic.metadata.refresh.sparse"] = String(self.topicMetadata.refreshSparse) |
| 118 | + resultDict["topic.metadata.propagation.max.ms"] = String(self.topicMetadata.propagationMaxMilliseconds) |
| 119 | + resultDict["topic.blacklist"] = self.topicDenylist.joined(separator: ",") |
| 120 | + if !self.debug.isEmpty { |
| 121 | + resultDict["debug"] = self.debug.map(\.description).joined(separator: ",") |
| 122 | + } |
| 123 | + resultDict["socket.timeout.ms"] = String(self.socket.timeoutMilliseconds) |
| 124 | + resultDict["socket.send.buffer.bytes"] = String(self.socket.sendBufferBytes) |
| 125 | + resultDict["socket.receive.buffer.bytes"] = String(self.socket.receiveBufferBytes) |
| 126 | + resultDict["socket.keepalive.enable"] = String(self.socket.keepaliveEnable) |
| 127 | + resultDict["socket.nagle.disable"] = String(self.socket.nagleDisable) |
| 128 | + resultDict["socket.max.fails"] = String(self.socket.maxFails) |
| 129 | + resultDict["socket.connection.setup.timeout.ms"] = String(self.socket.connectionSetupTimeoutMilliseconds) |
| 130 | + resultDict["broker.address.ttl"] = String(self.broker.addressTTL) |
| 131 | + resultDict["broker.address.family"] = self.broker.addressFamily.description |
| 132 | + resultDict["reconnect.backoff.ms"] = String(self.reconnect.backoffMilliseconds) |
| 133 | + resultDict["reconnect.backoff.max.ms"] = String(self.reconnect.backoffMaxMilliseconds) |
| 134 | + |
| 135 | + // Merge with SecurityProtocol configuration dictionary |
| 136 | + resultDict.merge(self.securityProtocol.dictionary) { _, _ in |
| 137 | + fatalError("securityProtocol and \(#file) should not have duplicate keys") |
| 138 | + } |
| 139 | + |
| 140 | + return resultDict |
| 141 | + } |
| 142 | +} |
0 commit comments