refactor: add semaphore

Replaced channel-based concurrency control with a custom semaphore implementation
to improve resource management in fetchAllPages. Added semaphore package for
handling concurrent operations.

Author: Ajnasz

loggly.go

@@ -66,7 +66,7 @@ var count = flags.Bool("count", false, "")
 var concurrency = flags.Int("concurrency", 3, "")
 var versionQuery = flags.Bool("version", false, "")
 var account = flags.String("account", "", "")
-var maxPages = flags.Int("maxPages", 3, "")
+var maxPages = flags.Int64("maxPages", 3, "")
 var token = flags.String("token", "", "")
 var size = flags.Int("size", 100, "")
 var from = flags.String("from", "-24h", "")
@@ -155,7 +155,7 @@ func sendQuery(
 	size int,
 	from string,
 	to string,
-	maxPages int,
+	maxPages int64,
 	concurrency int,
 ) {
 	doneChan := make(chan error)

search/query.go

@@ -12,7 +12,7 @@ type Query struct {
 	until    string
 	order    string
 	size     int
-	maxPages int
+	maxPages int64
 }
 
 // Create a new query
@@ -50,7 +50,7 @@ func (q *Query) From(str string) *Query {
 }
 
 // MaxPage sets the max page
-func (q *Query) MaxPage(maxPages int) *Query {
+func (q *Query) MaxPage(maxPages int64) *Query {
 	q.maxPages = maxPages
 	return q
 }

search/search.go

@@ -10,6 +10,7 @@ import (
 	"sync/atomic"
 
 	"github.com/Ajnasz/go-loggly-cli/orderedbuffer"
+	"github.com/Ajnasz/go-loggly-cli/semaphore"
 	"github.com/bitly/go-simplejson"
 )
 
@@ -170,8 +171,8 @@ func (c *Client) fetchAllPages(ctx context.Context, q Query, resChan chan Respon
 		return
 	}
 
-	concurrent := min(q.maxPages, int(c.concurrency.Load()))
-	pool := make(chan struct{}, concurrent)
+	concurrent := min(q.maxPages, c.concurrency.Load())
+	sem := semaphore.New(concurrent)
 
 	var page atomic.Int64
 	page.Store(-1)
@@ -182,11 +183,15 @@ func (c *Client) fetchAllPages(ctx context.Context, q Query, resChan chan Respon
 	responsesStore := orderedbuffer.NewOrderedBuffer(resChan)
 
 	for {
-		pool <- struct{}{}
+		if err := sem.Acquire(ctx); err != nil {
+			errChan <- err
+			break
+		}
+
 		wg.Add(1)
 		go func(page int) {
 			defer wg.Done()
-			defer func() { <-pool }()
+			defer sem.Release()
 
 			if ctx.Err() != nil {
 				errChan <- ctx.Err()
@@ -204,7 +209,7 @@ func (c *Client) fetchAllPages(ctx context.Context, q Query, resChan chan Respon
 			}
 		}(int(page.Add(1)))
 
-		shouldBreak := int(page.Load()) >= q.maxPages || !hasMore.Load()
+		shouldBreak := page.Load() >= q.maxPages || !hasMore.Load()
 
 		if shouldBreak {
 			break

semaphore/semaphore.go

@@ -0,0 +1,26 @@
+package semaphore
+
+import "context"
+
+type Semaphore struct {
+	ch chan struct{}
+}
+
+func New(n int64) *Semaphore {
+	return &Semaphore{
+		ch: make(chan struct{}, n),
+	}
+}
+
+func (s *Semaphore) Acquire(ctx context.Context) error {
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case s.ch <- struct{}{}:
+		return nil
+	}
+}
+
+func (s *Semaphore) Release() {
+	<-s.ch
+}