improvement(kafkax): add close for wrapper

xinpianchang · Aug 30, 2021 · aeb3a9f · aeb3a9f
1 parent a172447
commit aeb3a9f
Showing 1 changed file with 11 additions and 0 deletions.
diff --git a/pkg/kafkax/kafka.go b/pkg/kafkax/kafka.go
@@ -41,6 +41,9 @@ type Client interface {
 	// Get get kafka client
 	Get() sarama.Client
 
+	// Close close kafka client
+	Close() error
+
 	// SendMessage send message to kafka
 	SendMessage(ctx context.Context, message *sarama.ProducerMessage) error
 
@@ -102,6 +105,14 @@ func (t *defaultKafka) Get() sarama.Client {
 	return t.client
 }
 
+// Close close kafka client
+func (t *defaultKafka) Close() error {
+	if !t.client.Closed() {
+		return t.client.Close()
+	}
+	return nil
+}
+
 // SendMessage send message to kafka
 func (t *defaultKafka) SendMessage(ctx context.Context, message *sarama.ProducerMessage) (err error) {
 	if err := t.initializeProducer(ctx); err != nil {