Skip to content

Commit

Permalink
feat: switch log level
Browse files Browse the repository at this point in the history
  • Loading branch information
franklinkim committed May 30, 2024
1 parent 738ddd9 commit 3c143f5
Show file tree
Hide file tree
Showing 4 changed files with 7 additions and 12 deletions.
2 changes: 1 addition & 1 deletion integration/loki/loki.go
Original file line number Diff line number Diff line change
Expand Up @@ -182,7 +182,7 @@ func (l *Loki) Stop() {
// ------------------------------------------------------------------------------------------------

func (l *Loki) process(entries []logproto.Entry) {
l.l.Info("processing entries batch", zap.Int("num", len(entries)))
l.l.Debug("processing entries batch", zap.Int("num", len(entries)))

labels := model.LabelSet{
"name": "events",
Expand Down
6 changes: 3 additions & 3 deletions integration/watermill/gtag/subscriber.go
Original file line number Diff line number Diff line change
Expand Up @@ -166,13 +166,13 @@ func (s *Subscriber) handle(l *zap.Logger, r *http.Request, payload *gtag.Payloa
// wait for ACK
select {
case <-msg.Acked():
l.Info("message acked")
l.Debug("message acked")
return nil
case <-msg.Nacked():
l.Info("message nacked")
l.Debug("message nacked")
return ErrMessageNacked
case <-r.Context().Done():
l.Info("message cancled")
l.Debug("message cancled")
return ErrContextCanceled
}
}
Expand Down
6 changes: 3 additions & 3 deletions integration/watermill/mpv2/subscriber.go
Original file line number Diff line number Diff line change
Expand Up @@ -139,13 +139,13 @@ func (s *Subscriber) handle(l *zap.Logger, r *http.Request, payload *mpv2.Payloa
// wait for ACK
select {
case <-msg.Acked():
l.Info("message acked")
l.Debug("message acked")
return nil
case <-msg.Nacked():
l.Info("message nacked")
l.Debug("message nacked")
return ErrMessageNacked
case <-r.Context().Done():
l.Info("message cancled")
l.Debug("message cancled")
return ErrContextCanceled
}
}
Expand Down
5 changes: 0 additions & 5 deletions pkg/utils/batch.go
Original file line number Diff line number Diff line change
Expand Up @@ -2,14 +2,12 @@ package utils

import (
"context"
"fmt"
)

// Batch reads from a channel and calls fn with a slice of batchSize.
func Batch[T any](ctx context.Context, ch <-chan T, batchSize int, fn func([]T)) {
if batchSize <= 1 { // sanity check,
for v := range ch {
fmt.Println("<< 1")
fn([]T{v})
}
return
Expand All @@ -21,20 +19,17 @@ func Batch[T any](ctx context.Context, ch <-chan T, batchSize int, fn func([]T))
select {
case <-ctx.Done():
if len(batch) > 0 {
fmt.Println("<< 2")
fn(batch)
}
return
case v, ok := <-ch:
if !ok { // closed
fmt.Println("<< 3")
fn(batch)
return
}

batch = append(batch, v)
if len(batch) == batchSize { // full
fmt.Println("<< 4")
fn(batch)
batch = make([]T, 0, batchSize) // reset
}
Expand Down

0 comments on commit 3c143f5

Please sign in to comment.