fix: continue leader failed should close leader component (#56)

* fix: continue leader failed should close leader component

* test: add case for LeaderChangeHandler

Author: yannan

fastflow.go

@@ -152,7 +152,7 @@ func (l *LeaderChangedHandler) Handle(cxt context.Context, e goevent.Event) {
 		log.Println("leader initial")
 	}
 	// continue leader failed
-	if !lcEvent.IsLeader && len(l.leaderCloser) == 0 {
+	if !lcEvent.IsLeader && len(l.leaderCloser) != 0 {
 		l.Close()
 	}
 }

fastflow_test.go

@@ -1,11 +1,13 @@
 package fastflow
 
 import (
+	"context"
 	"fmt"
 	"testing"
 	"time"
 
 	"github.com/shiningrush/fastflow/pkg/entity"
+	"github.com/shiningrush/fastflow/pkg/event"
 	"github.com/shiningrush/fastflow/pkg/mod"
 	"github.com/shiningrush/fastflow/pkg/utils"
 	"github.com/stretchr/testify/assert"
@@ -190,3 +192,48 @@ tasks:
 		})
 	}
 }
+
+func Test_LeaderChangeHandler(t *testing.T) {
+	tests := []struct {
+		isLeader      bool
+		calledEnsured []bool
+		closerLenth   int
+	}{
+		{
+			isLeader:      true,
+			calledEnsured: []bool{},
+			closerLenth:   1,
+		},
+		{
+			isLeader:      false,
+			calledEnsured: []bool{true},
+			closerLenth:   0,
+		},
+	}
+
+	for _, tc := range tests {
+		called := []bool{}
+		keeper := &mod.MockKeeper{}
+		handler := &LeaderChangedHandler{
+			opt: &InitialOption{
+				Keeper: keeper,
+			},
+			leaderCloser: []mod.Closer{},
+		}
+
+		closer := &mod.MockCloser{}
+		closer.On("Close").Run(func(args mock.Arguments) {
+			called = append(called, true)
+		}).Return(nil)
+		handler.leaderCloser = append(handler.leaderCloser, closer)
+
+		mockEvent := &event.LeaderChanged{IsLeader: tc.isLeader, WorkerKey: "worker-key"}
+		ctx := context.Background()
+
+		// Simulate the event that leads to leadership
+		handler.Handle(ctx, mockEvent)
+
+		assert.Equal(t, tc.calledEnsured, called)
+		assert.Equal(t, tc.closerLenth, len(handler.leaderCloser))
+	}
+}