clean up tests

pkg/util/logger.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"io"
+	"os"
 	"sync"
 	"time"
 
@@ -84,6 +85,7 @@ type AsyncWriter struct {
 	closeOnce     sync.Once
 	close         chan struct{}
 	done          chan error
+	closed        bool
 }
 
 func NewAsyncWriter(w io.Writer, bufSize, maxBuffers, maxWrites int, flushInterval time.Duration) *AsyncWriter {
@@ -108,6 +110,9 @@ func NewAsyncWriter(w io.Writer, bufSize, maxBuffers, maxWrites int, flushInterv
 func (aw *AsyncWriter) Write(p []byte) (int, error) {
 	aw.mu.Lock()
 	defer aw.mu.Unlock()
+	if aw.closed {
+		return 0, os.ErrClosed
+	}
 	if aw.overflows(len(p)) {
 		aw.enqueueFlush()
 	}
@@ -136,6 +141,7 @@ func (aw *AsyncWriter) Close() error {
 		for buf := range aw.flushQueue {
 			aw.flushSync(buf)
 		}
+		aw.closed = true
 	})
 	return nil
 }
@@ -147,7 +153,10 @@ func (aw *AsyncWriter) enqueueFlush() {
 	}
 	aw.buffer = nil
 	aw.writes = 0
-	aw.flushQueue <- buf
+	select {
+	case aw.flushQueue <- buf:
+	default:
+	}
 }
 
 func (aw *AsyncWriter) loop() {

pkg/util/logger_test.go

@@ -2,6 +2,7 @@ package util
 
 import (
 	"bytes"
+	"fmt"
 	"sync"
 	"testing"
 	"time"
@@ -12,63 +13,52 @@ import (
 
 func TestAsyncWriter_Write(t *testing.T) {
 	var buf bytes.Buffer
-	writer := NewAsyncWriter(&buf, 10, 2, 2, 100*time.Millisecond)
-	n, err := writer.Write([]byte("hello"))
+	w := NewAsyncWriter(&buf, 10, 2, 2, 100*time.Millisecond)
+	n, err := w.Write([]byte("hello"))
 	require.NoError(t, err)
 	assert.Equal(t, 5, n)
-	writer.Close()
+	assert.NoError(t, w.Close())
 	assert.Equal(t, "hello", buf.String())
 }
 
 func TestAsyncWriter_Empty(t *testing.T) {
 	var buf bytes.Buffer
-	writer := NewAsyncWriter(&buf, 10, 2, 2, 100*time.Millisecond)
-	writer.Close()
+	w := NewAsyncWriter(&buf, 10, 2, 2, 100*time.Millisecond)
+	assert.NoError(t, w.Close())
 	assert.EqualValues(t, 0, buf.Len())
 }
 
 func TestAsyncWriter_Overflow(t *testing.T) {
 	var buf bytes.Buffer
-	writer := NewAsyncWriter(&buf, 10, 2, 2, 100*time.Millisecond)
-	_, _ = writer.Write([]byte("hello"))
-	_, _ = writer.Write([]byte("world"))
-	writer.Close()
+	w := NewAsyncWriter(&buf, 10, 2, 2, 100*time.Millisecond)
+	_, _ = w.Write([]byte("hello"))
+	_, _ = w.Write([]byte("world"))
+	assert.NoError(t, w.Close())
 	assert.Equal(t, "helloworld", buf.String())
 }
 
-func TestAsyncWriter_FlushInterval(t *testing.T) {
-	var buf bytes.Buffer
-	writer := NewAsyncWriter(&buf, 10, 2, 2, 10*time.Millisecond)
-	defer writer.Close()
-	_, _ = writer.Write([]byte("hello"))
-	assert.Eventually(t,
-		func() bool { return assert.Equal(t, "hello", buf.String()) },
-		time.Second, 10*time.Millisecond,
-	)
-}
-
 func TestAsyncWriter_Close(t *testing.T) {
 	var buf bytes.Buffer
-	writer := NewAsyncWriter(&buf, 10, 2, 2, 100*time.Millisecond)
-	_, _ = writer.Write([]byte("hello"))
-	assert.NoError(t, writer.Close())
+	w := NewAsyncWriter(&buf, 10, 2, 2, 100*time.Millisecond)
+	_, _ = w.Write([]byte("hello"))
+	assert.NoError(t, w.Close())
 	assert.Equal(t, "hello", buf.String())
-	assert.NoError(t, writer.Close())
+	assert.NoError(t, w.Close())
 }
 
 func TestAsyncWriter_ConcurrentWrites(t *testing.T) {
 	var buf bytes.Buffer
-	writer := NewAsyncWriter(&buf, 10, 2, 2, 100*time.Millisecond)
+	w := NewAsyncWriter(&buf, 50, 2, 10, 100*time.Millisecond)
 	var wg sync.WaitGroup
 	for i := 0; i < 10; i++ {
 		wg.Add(1)
 		go func(i int) {
 			defer wg.Done()
-			_, _ = writer.Write([]byte("hello"))
+			_, _ = fmt.Fprintf(w, "hello %d\n", i)
 		}(i)
 	}
 	wg.Wait()
 
-	writer.Close()
-	assert.Equal(t, 50, buf.Len())
+	assert.NoError(t, w.Close())
+	assert.Equal(t, 80, buf.Len())
 }