chore(PubSub): Move pool to its own package

Author: Neurostep

pkg/pubsub/kafka/partitionconsumer.go

@@ -7,10 +7,11 @@ import (
 
 	sdkkafka "github.com/scribd/go-sdk/pkg/instrumentation/kafka"
 	sdklogger "github.com/scribd/go-sdk/pkg/logger"
+	"github.com/scribd/go-sdk/pkg/pubsub/pool"
 )
 
 type pconsumer struct {
-	pool *pool
+	pool *pool.Pool
 
 	quit chan struct{}
 	done chan struct{}

pkg/pubsub/kafka/subscriber.go

@@ -12,6 +12,7 @@ import (
 
 	sdkkafka "github.com/scribd/go-sdk/pkg/instrumentation/kafka"
 	sdklogger "github.com/scribd/go-sdk/pkg/logger"
+	"github.com/scribd/go-sdk/pkg/pubsub/pool"
 )
 
 type (
@@ -152,7 +153,7 @@ func (s *Subscriber) assigned(_ context.Context, cl *kgo.Client, assigned map[st
 			pc := pconsumer{
 				quit: make(chan struct{}),
 				recs: make(chan *sdkkafka.FetchPartition),
-				pool: newPool(s.numWorkers),
+				pool: pool.New(s.numWorkers),
 				done: make(chan struct{}),
 			}
 			s.consumers[topic][partition] = pc

pkg/pubsub/kafka/pool.go renamed to pkg/pubsub/pool/pool.go

@@ -1,20 +1,20 @@
-package kafka
+package pool
 
-type pool struct {
+type Pool struct {
 	sem  chan struct{}
 	work chan func()
 }
 
-func newPool(size int) *pool {
-	p := &pool{
+func New(size int) *Pool {
+	p := &Pool{
 		sem:  make(chan struct{}, size),
 		work: make(chan func()),
 	}
 
 	return p
 }
 
-func (p *pool) Schedule(task func()) {
+func (p *Pool) Schedule(task func()) {
 	select {
 	case p.work <- task:
 		return
@@ -23,7 +23,7 @@ func (p *pool) Schedule(task func()) {
 	}
 }
 
-func (p *pool) worker(task func()) {
+func (p *Pool) worker(task func()) {
 	defer func() { <-p.sem }()
 
 	for {