chore: fix tests

Trendyol · Jan 14, 2024 · c75e4bf · c75e4bf
1 parent e63a263
commit c75e4bf
Show file tree

Hide file tree

Showing 5 changed files with 41 additions and 33 deletions.
diff --git a/batch_consumer_test.go b/batch_consumer_test.go
@@ -360,10 +360,10 @@ func Test_batchConsumer_Pause(t *testing.T) {
 	ctx, cancelFn := context.WithCancel(context.Background())
 	bc := batchConsumer{
 		base: &base{
-			logger:   NewZapLogger(LogLevelDebug),
-			context:  ctx,
-			pause:    make(chan struct{}),
-			cancelFn: cancelFn,
+			logger:  NewZapLogger(LogLevelDebug),
+			pause:   make(chan struct{}),
+			context: ctx, cancelFn: cancelFn,
+			consumerState: stateRunning,
 		},
 	}
 
@@ -382,12 +382,16 @@ func Test_batchConsumer_Pause(t *testing.T) {
 
 func Test_batchConsumer_Resume(t *testing.T) {
 	// Given
+	mc := mockReader{}
 	ctx, cancelFn := context.WithCancel(context.Background())
 	bc := batchConsumer{
 		base: &base{
-			logger:   NewZapLogger(LogLevelDebug),
-			context:  ctx,
-			cancelFn: cancelFn,
+			r:       &mc,
+			logger:  NewZapLogger(LogLevelDebug),
+			pause:   make(chan struct{}),
+			quit:    make(chan struct{}),
+			wg:      sync.WaitGroup{},
+			context: ctx, cancelFn: cancelFn,
 		},
 	}
 

diff --git a/consumer_base.go b/consumer_base.go
@@ -179,7 +179,6 @@ func (c *base) Pause() {
 	c.cancelFn()
 
 	c.pause <- struct{}{}
-	close(c.pause)
 
 	c.consumerState = statePaused
 }
@@ -200,15 +199,15 @@ func (c *base) WithLogger(logger LoggerInterface) {
 }
 
 func (c *base) Stop() error {
-	c.logger.Info("Stop called!")
+	c.logger.Info("Stop is called!")
 
 	var err error
 	c.once.Do(func() {
 		c.subprocesses.Stop()
 		c.cancelFn()
 
 		// In order to save cpu, we break startConsume loop in pause mode.
-		// If consumer is pause mode, and Stop called
+		// If consumer is pause mode and Stop is called
 		// We need to close incomingMessageStream, because c.wg.Wait() blocks indefinitely.
 		if c.consumerState == stateRunning {
 			c.quit <- struct{}{}

diff --git a/consumer_base_test.go b/consumer_base_test.go
@@ -3,24 +3,24 @@ package kafka
 import (
 	"context"
 	"errors"
+	"github.com/google/go-cmp/cmp"
+	"github.com/segmentio/kafka-go"
 	"sync"
 	"testing"
 	"time"
-
-	"github.com/google/go-cmp/cmp"
-
-	"github.com/segmentio/kafka-go"
 )
 
 func Test_base_startConsume(t *testing.T) {
 	t.Run("Return_When_Quit_Signal_Is_Came", func(t *testing.T) {
 		mc := mockReader{wantErr: true}
 		b := base{
-			wg: sync.WaitGroup{}, r: &mc,
+			wg:                    sync.WaitGroup{},
+			r:                     &mc,
 			incomingMessageStream: make(chan *IncomingMessage),
 			quit:                  make(chan struct{}),
 			pause:                 make(chan struct{}),
-			logger:                NewZapLogger(LogLevelInfo),
+			logger:                NewZapLogger(LogLevelError),
+			consumerState:         stateRunning,
 		}
 		b.context, b.cancelFn = context.WithCancel(context.Background())
 
@@ -65,6 +65,7 @@ func Test_base_Pause(t *testing.T) {
 		logger:  NewZapLogger(LogLevelDebug),
 		pause:   make(chan struct{}),
 		context: ctx, cancelFn: cancelFn,
+		consumerState: stateRunning,
 	}
 	go func() {
 		<-b.pause
@@ -86,10 +87,10 @@ func Test_base_Resume(t *testing.T) {
 	b := base{
 		r:       &mc,
 		logger:  NewZapLogger(LogLevelDebug),
-		quit:    make(chan struct{}),
 		pause:   make(chan struct{}),
+		quit:    make(chan struct{}),
+		wg:      sync.WaitGroup{},
 		context: ctx, cancelFn: cancelFn,
-		wg: sync.WaitGroup{},
 	}
 
 	// When

diff --git a/consumer_test.go b/consumer_test.go
@@ -3,6 +3,7 @@ package kafka
 import (
 	"context"
 	"errors"
+	"sync"
 	"testing"
 )
 
@@ -121,10 +122,10 @@ func Test_consumer_Pause(t *testing.T) {
 	ctx, cancelFn := context.WithCancel(context.Background())
 	c := consumer{
 		base: &base{
-			logger:   NewZapLogger(LogLevelDebug),
-			pause:    make(chan struct{}),
-			context:  ctx,
-			cancelFn: cancelFn,
+			logger:  NewZapLogger(LogLevelDebug),
+			pause:   make(chan struct{}),
+			context: ctx, cancelFn: cancelFn,
+			consumerState: stateRunning,
 		},
 	}
 	go func() {
@@ -142,12 +143,16 @@ func Test_consumer_Pause(t *testing.T) {
 
 func Test_consumer_Resume(t *testing.T) {
 	// Given
+	mc := mockReader{}
 	ctx, cancelFn := context.WithCancel(context.Background())
 	c := consumer{
 		base: &base{
-			logger:   NewZapLogger(LogLevelDebug),
-			context:  ctx,
-			cancelFn: cancelFn,
+			r:       &mc,
+			logger:  NewZapLogger(LogLevelDebug),
+			pause:   make(chan struct{}),
+			quit:    make(chan struct{}),
+			wg:      sync.WaitGroup{},
+			context: ctx, cancelFn: cancelFn,
 		},
 	}
 

diff --git a/examples/with-pause-resume-consumer/main.go b/examples/with-pause-resume-consumer/main.go
@@ -33,16 +33,15 @@ func main() {
 
 		time.Sleep(10 * time.Second)
 		consumer.Resume()
-		/*
-			time.Sleep(10 * time.Second)
-			consumer.Pause()
 
-				time.Sleep(10 * time.Second)
-				consumer.Resume()
+		time.Sleep(10 * time.Second)
+		consumer.Pause()
 
-				time.Sleep(10 * time.Second)
-				consumer.Pause()
-		*/
+		time.Sleep(10 * time.Second)
+		consumer.Resume()
+
+		time.Sleep(10 * time.Second)
+		consumer.Pause()
 	}()
 
 	c := make(chan os.Signal, 1)