Merge pull request #7 from ksysoev/perfomance_improvments

Improve stability of server test

Author: ksysoev

server.go

@@ -27,7 +27,7 @@ type Server struct {
 	handlers     map[string]RequestHandler
 	handlersLock *sync.RWMutex
 	cancel       context.CancelFunc
-	sem          chan struct{}
+	sem          chan token
 	wg           *sync.WaitGroup
 	stream       string
 	group        string
@@ -54,7 +54,7 @@ func NewServer(redisClient *redis.Client, stream, group, consumer string, opts .
 		ctx:          ctx,
 		cancel:       cancel,
 		consumer:     consumer,
-		sem:          make(chan struct{}, DefaultConcurency),
+		sem:          make(chan token, DefaultConcurency),
 		wg:           &sync.WaitGroup{},
 	}
 

server_test.go

@@ -244,20 +244,11 @@ func TestServer_Close(t *testing.T) {
 	group := "myGroup"
 	consumer := "myConsumer"
 
-	xReadArgs := &redis.XReadGroupArgs{
-		Group:    group,
-		Consumer: consumer,
-		Streams:  []string{stream, ">"},
-		Block:    DefaultBlockInterval,
-		Count:    DefaultConcurency,
-		NoAck:    false,
-	}
-
 	mock.ExpectXGroupCreateMkStream(stream, group, "$").SetVal("OK")
 	mock.ExpectXGroupCreateConsumer(stream, group, consumer).SetVal(1)
-	mock.ExpectXReadGroup(xReadArgs).SetErr(redis.Nil)
 
 	server := NewServer(redisClient, stream, group, consumer)
+	server.sem = make(chan token)
 
 	done := make(chan struct{})
 	go func() {