-
Notifications
You must be signed in to change notification settings - Fork 7
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #38 from ahmetfurkankavraz/main
- Loading branch information
Showing
11 changed files
with
257 additions
and
87 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,96 @@ | ||
| using Confluent.Kafka; | ||
|
|
||
| namespace KafkaRetry.Job.Helpers.KafkaConfigs; | ||
|
|
||
| public static class ClientConfigBuilder | ||
| { | ||
| public static ClientConfig WithBootstrapServers(this ClientConfig config, string bootstrapServers) | ||
| { | ||
| if (!string.IsNullOrWhiteSpace(bootstrapServers)) | ||
| { | ||
| config.BootstrapServers = bootstrapServers; | ||
| } | ||
| return config; | ||
| } | ||
|
|
||
| public static ClientConfig WithSaslUsername(this ClientConfig config, string username) | ||
| { | ||
| if (!string.IsNullOrWhiteSpace(username)) | ||
| { | ||
| config.SaslUsername = username; | ||
| } | ||
| return config; | ||
| } | ||
|
|
||
| public static ClientConfig WithSaslPassword(this ClientConfig config, string password) | ||
| { | ||
| if (!string.IsNullOrWhiteSpace(password)) | ||
| { | ||
| config.SaslPassword = password; | ||
| } | ||
| return config; | ||
| } | ||
|
|
||
| public static ClientConfig WithSslCaLocation(this ClientConfig config, string sslCaLocation) | ||
| { | ||
| if (!string.IsNullOrWhiteSpace(sslCaLocation)) | ||
| { | ||
| config.SslCaLocation = sslCaLocation; | ||
| } | ||
| return config; | ||
| } | ||
|
|
||
| public static ClientConfig WithSaslMechanism(this ClientConfig config, SaslMechanism? saslMechanism) | ||
| { | ||
| if (saslMechanism is not null) | ||
| { | ||
| config.SaslMechanism = saslMechanism; | ||
| } | ||
| return config; | ||
| } | ||
|
|
||
| public static ClientConfig WithSecurityProtocol(this ClientConfig config, SecurityProtocol? securityProtocol) | ||
| { | ||
| if (securityProtocol is not null) | ||
| { | ||
| config.SecurityProtocol = securityProtocol; | ||
| } | ||
| return config; | ||
| } | ||
|
|
||
| public static ClientConfig WithSslKeystorePassword(this ClientConfig config, string sslKeystorePassword) | ||
| { | ||
| if (!string.IsNullOrWhiteSpace(sslKeystorePassword)) | ||
| { | ||
| config.SslKeystorePassword = sslKeystorePassword; | ||
| } | ||
| return config; | ||
| } | ||
|
|
||
| public static ClientConfig WithClientId(this ClientConfig config, string clientId) | ||
| { | ||
| if (!string.IsNullOrWhiteSpace(clientId)) | ||
| { | ||
| config.ClientId = clientId; | ||
| } | ||
| return config; | ||
| } | ||
|
|
||
| public static ClientConfig WithMessageMaxBytes(this ClientConfig config, int? messageMaxBytes) | ||
| { | ||
| if (messageMaxBytes is not null) | ||
| { | ||
| config.MessageMaxBytes = messageMaxBytes; | ||
| } | ||
| return config; | ||
| } | ||
|
|
||
| public static ClientConfig WithAcks(this ClientConfig config, Acks? acks) | ||
| { | ||
| if (acks is not null) | ||
| { | ||
| config.Acks = acks; | ||
| } | ||
| return config; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,42 @@ | ||
| using Confluent.Kafka; | ||
|
|
||
| namespace KafkaRetry.Job.Helpers.KafkaConfigs; | ||
|
|
||
| public static class ConsumerConfigBuilder | ||
| { | ||
| public static ConsumerConfig WithAutoOffsetReset(this ConsumerConfig config, AutoOffsetReset? autoOffsetReset) | ||
| { | ||
| if (autoOffsetReset is not null) | ||
| { | ||
| config.AutoOffsetReset = autoOffsetReset; | ||
| } | ||
| return config; | ||
| } | ||
|
|
||
| public static ConsumerConfig WithGroupId(this ConsumerConfig config, string groupId) | ||
| { | ||
| if (!string.IsNullOrWhiteSpace(groupId)) | ||
| { | ||
| config.GroupId = groupId; | ||
| } | ||
| return config; | ||
| } | ||
|
|
||
| public static ConsumerConfig WithEnableAutoCommit(this ConsumerConfig config, bool? autoCommit) | ||
| { | ||
| if (autoCommit is not null) | ||
| { | ||
| config.EnableAutoCommit = autoCommit; | ||
| } | ||
| return config; | ||
| } | ||
|
|
||
| public static ConsumerConfig WithEnableAutoOffsetStore(this ConsumerConfig config, bool? autoOffsetStore) | ||
| { | ||
| if (autoOffsetStore is not null) | ||
| { | ||
| config.EnableAutoOffsetStore = autoOffsetStore; | ||
| } | ||
| return config; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,48 @@ | ||
| using Confluent.Kafka; | ||
|
|
||
| namespace KafkaRetry.Job.Helpers.KafkaConfigs; | ||
|
|
||
| public static class ProducerConfigBuilder | ||
| { | ||
| public static ProducerConfig WithEnableIdempotence(this ProducerConfig config, bool? idempotence) | ||
| { | ||
| if (idempotence is not null) | ||
| { | ||
| config.EnableIdempotence = idempotence; | ||
| } | ||
| return config; | ||
| } | ||
|
|
||
| public static ProducerConfig WithBatchSize(this ProducerConfig config, int? batchSize) | ||
| { | ||
| if (batchSize is not null) | ||
| { | ||
| config.BatchSize = batchSize; | ||
| } | ||
| return config; | ||
| } | ||
|
|
||
| public static ProducerConfig WithLingerMs(this ProducerConfig config, double? lingerMs) | ||
| { | ||
| if (lingerMs is not null) | ||
| { | ||
| config.LingerMs = lingerMs; | ||
| } | ||
| return config; | ||
| } | ||
|
|
||
| public static ProducerConfig WithMessageTimeoutMs(this ProducerConfig config, int? messageTimeoutMs) | ||
| { | ||
| if (messageTimeoutMs is not null) | ||
| { | ||
| config.MessageTimeoutMs = messageTimeoutMs; | ||
| } | ||
| return config; | ||
| } | ||
|
|
||
| public static ProducerConfig WithRequestTimeoutMs(this ProducerConfig config, int? requestTimeoutMs) | ||
| { | ||
| config.RequestTimeoutMs = requestTimeoutMs; | ||
| return config; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.