68 lines
1.0 KiB
Go
68 lines
1.0 KiB
Go
package kafka
|
|
|
|
import (
|
|
"context"
|
|
"crypto/tls"
|
|
"fmt"
|
|
"sync"
|
|
"time"
|
|
|
|
"ultone/internal/interfaces"
|
|
|
|
"github.com/loveuer/nf/nft/log"
|
|
kfkgo "github.com/segmentio/kafka-go"
|
|
"github.com/segmentio/kafka-go/sasl"
|
|
)
|
|
|
|
var Client *client
|
|
|
|
type client struct {
|
|
sync.Mutex
|
|
ctx context.Context
|
|
d *kfkgo.Dialer
|
|
topic string
|
|
partition int
|
|
reconnection bool
|
|
mechanism sasl.Mechanism
|
|
address string
|
|
logger interfaces.Logger
|
|
writer *kfkgo.Writer
|
|
tls *tls.Config
|
|
}
|
|
|
|
func Init(address string, opts ...OptionFn) error {
|
|
c, err := New(address, opts...)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
Client = c
|
|
|
|
return nil
|
|
}
|
|
|
|
func New(address string, opts ...OptionFn) (*client, error) {
|
|
c := &client{
|
|
ctx: context.Background(),
|
|
d: &kfkgo.Dialer{
|
|
Timeout: 30 * time.Second,
|
|
DualStack: false,
|
|
},
|
|
tls: nil,
|
|
mechanism: nil,
|
|
logger: log.New(),
|
|
}
|
|
|
|
if address == "" {
|
|
return nil, fmt.Errorf("address required")
|
|
}
|
|
|
|
for _, fn := range opts {
|
|
fn(c)
|
|
}
|
|
|
|
c.address = address
|
|
|
|
return c, nil
|
|
}
|