mirror of
https://github.com/hanzo-fi/go-libs.git
synced 2026-10-30 20:53:44 +00:00
de61635b81
* refactor: move all packages under pkg/ for v5 Move all 13 top-level Go packages into pkg/ for a cleaner repository structure. All internal import paths updated from github.com/formancehq/go-libs/v5/<pkg> to github.com/formancehq/go-libs/v5/pkg/<pkg>. Also includes cleanup: - Fix .golangci.yml stale v4 CircuitBreaker path - Replace deprecated str.Contains with stdlib slices.Contains - Update TODO(v4) markers to TODO - Delete stale coverage.txt - Fix pgtesting timeout not being propagated to docker pool - Update docs/ARCHITECTURE.md to reflect new structure Co-Authored-By: Claude Opus 4.6 <[email protected]> * chore: remove compatibility aliases, unused params, and stub files - Remove unused type aliases (Authenticator, KeySet) from authnfx - Remove unused `debug` parameter from S3ModuleFromFlags - Delete comment-only stub files in messaging/publish/circuit/ Co-Authored-By: Claude Opus 4.6 <[email protected]> * chore: rename cli.go to flags.go for consistency Standardize naming across all packages: cli.go -> flags.go since these files all serve the same purpose (flag definitions). Co-Authored-By: Claude Opus 4.6 <[email protected]> * chore: rename * chore(deps): update all dependencies Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: resolve package name mismatches in generated mocks Generated mock files had stale package names from before the pkg/ move, causing import cycles and build failures with --build-tags it. Co-Authored-By: Claude Opus 4.6 <[email protected]> --------- Co-authored-by: Claude Opus 4.6 <[email protected]>
114 lines
2.7 KiB
Go
114 lines
2.7 KiB
Go
package publish
|
|
|
|
import (
|
|
"crypto/tls"
|
|
|
|
"github.com/IBM/sarama"
|
|
"github.com/ThreeDotsLabs/watermill"
|
|
"github.com/ThreeDotsLabs/watermill-kafka/v3/pkg/kafka"
|
|
)
|
|
|
|
type SaramaOption interface {
|
|
Apply(config *sarama.Config)
|
|
}
|
|
type SaramaOptionFn func(config *sarama.Config)
|
|
|
|
func (fn SaramaOptionFn) Apply(config *sarama.Config) {
|
|
fn(config)
|
|
}
|
|
|
|
func WithConsumerOffsetsInitial(v int64) SaramaOptionFn {
|
|
return func(config *sarama.Config) {
|
|
config.Consumer.Offsets.Initial = v
|
|
}
|
|
}
|
|
|
|
func WithConsumerReturnErrors() SaramaOptionFn {
|
|
return func(config *sarama.Config) {
|
|
config.Consumer.Return.Errors = true
|
|
}
|
|
}
|
|
|
|
func WithProducerReturnSuccess() SaramaOptionFn {
|
|
return func(config *sarama.Config) {
|
|
config.Producer.Return.Successes = true
|
|
}
|
|
}
|
|
|
|
func WithSASLEnabled() SaramaOptionFn {
|
|
return func(config *sarama.Config) {
|
|
config.Net.SASL.Enable = true
|
|
}
|
|
}
|
|
|
|
func WithSASLMechanism(mechanism sarama.SASLMechanism) SaramaOptionFn {
|
|
return func(config *sarama.Config) {
|
|
config.Net.SASL.Mechanism = mechanism
|
|
}
|
|
}
|
|
|
|
func WithSASLScramClient(fn func() sarama.SCRAMClient) SaramaOptionFn {
|
|
return func(config *sarama.Config) {
|
|
config.Net.SASL.SCRAMClientGeneratorFunc = fn
|
|
}
|
|
}
|
|
|
|
func WithSASLCredentials(user, pwd string) SaramaOptionFn {
|
|
return func(config *sarama.Config) {
|
|
config.Net.SASL.User = user
|
|
config.Net.SASL.Password = pwd
|
|
}
|
|
}
|
|
|
|
func WithTokenProvider(provider sarama.AccessTokenProvider) SaramaOptionFn {
|
|
return func(config *sarama.Config) {
|
|
config.Net.SASL.TokenProvider = provider
|
|
}
|
|
}
|
|
|
|
func WithTLS() SaramaOptionFn {
|
|
return func(config *sarama.Config) {
|
|
config.Net.TLS = struct {
|
|
Enable bool
|
|
Config *tls.Config
|
|
}{
|
|
Enable: true,
|
|
Config: &tls.Config{},
|
|
}
|
|
}
|
|
}
|
|
|
|
type ClientID string
|
|
|
|
func NewSaramaConfig(clientId ClientID, version sarama.KafkaVersion, options ...SaramaOption) *sarama.Config {
|
|
config := sarama.NewConfig()
|
|
config.ClientID = string(clientId)
|
|
config.Version = version
|
|
|
|
for _, opt := range options {
|
|
opt.Apply(config)
|
|
}
|
|
|
|
return config
|
|
}
|
|
|
|
func NewKafkaPublisher(logger watermill.LoggerAdapter, config *sarama.Config, marshaller kafka.Marshaler, brokers ...string) (*kafka.Publisher, error) {
|
|
return kafka.NewPublisher(kafka.PublisherConfig{
|
|
Brokers: brokers,
|
|
Marshaler: marshaller,
|
|
OverwriteSaramaConfig: config,
|
|
OTELEnabled: true,
|
|
}, logger)
|
|
}
|
|
|
|
func NewKafkaSubscriber(logger watermill.LoggerAdapter, config *sarama.Config,
|
|
unmarshaler kafka.Unmarshaler, consumerGroup string, brokers ...string) (*kafka.Subscriber, error) {
|
|
return kafka.NewSubscriber(kafka.SubscriberConfig{
|
|
Brokers: brokers,
|
|
OverwriteSaramaConfig: config,
|
|
Unmarshaler: unmarshaler,
|
|
OTELEnabled: true,
|
|
ConsumerGroup: consumerGroup,
|
|
}, logger)
|
|
}
|