Add non blocking channel operation exercise.

Author: kawpii

internal/exercises/catalog.yaml

@@ -241,6 +241,12 @@ concepts:
   test_regex: ".*"
   hints:
     - Use `for i := range ch {...}` syntax to range over a channels. 
+- slug: 46_non_blocking_channel_operations
+  title: Non-blocking channel operations
+  test_regex: ".*"
+  hints:
+    - Use `default` case in select to handle non-blocking channel operations.
+    - Safely increment `dropped` using mutex `Lock()` and `Unlock()`.
 
 projects:
 - slug: 101_text_analyzer

internal/exercises/solutions/46_non_blocking_channel_operations/non_blocking_channel_operations.go

@@ -0,0 +1,66 @@
+package non_blocking_channel_operations
+
+import (
+	"sync"
+	"time"
+)
+
+// Assume case where you have high throughput system
+// where you have to drop logs
+// if they cannot be processed in time
+// in order to keep system throughput high.
+
+type Logger struct {
+	ch        chan string
+	wg        sync.WaitGroup
+	processed int
+	dropped   int
+
+	mu sync.Mutex
+}
+
+func NewLogger(bufferSize int) *Logger {
+	l := &Logger{
+		ch: make(chan string, bufferSize),
+	}
+	l.wg.Add(1)
+	go l.run()
+	return l
+}
+
+var processingDelay = 200 * time.Microsecond
+
+// slow log processor
+func (l *Logger) run() {
+	defer l.wg.Done()
+	for msg := range l.ch {
+		_ = msg
+		time.Sleep(processingDelay)
+
+		l.mu.Lock()
+		l.processed++
+		l.mu.Unlock()
+	}
+}
+
+func (l *Logger) Log(msg string) {
+	select {
+	case l.ch <- msg: // queued
+	default:
+		l.mu.Lock()
+		l.dropped++
+		l.mu.Unlock()
+	}
+}
+
+// graceful shutdown
+func (l *Logger) Close() {
+	close(l.ch)
+	l.wg.Wait()
+}
+
+func (l *Logger) Stats() (processed, dropped int) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	return l.processed, l.dropped
+}

internal/exercises/templates/46_non_blocking_channel_operations/non_blocking_channel_operations.go

@@ -0,0 +1,63 @@
+package non_blocking_channel_operations
+
+import (
+	"sync"
+	"time"
+)
+
+// Assume case where you have high throughput system
+// where you have to drop logs
+// if they cannot be processed in time,
+// in order to keep system throughput high.
+
+type Logger struct {
+	ch        chan string
+	wg        sync.WaitGroup
+	processed int
+	dropped   int
+
+	mu sync.Mutex
+}
+
+func NewLogger(bufferSize int) *Logger {
+	l := &Logger{
+		ch: make(chan string, bufferSize),
+	}
+	l.wg.Add(1)
+	go l.run()
+	return l
+}
+
+var processingDelay = 200 * time.Microsecond
+
+// slow log processor
+func (l *Logger) run() {
+	defer l.wg.Done()
+	for msg := range l.ch {
+		_ = msg
+		time.Sleep(processingDelay)
+
+		l.mu.Lock()
+		l.processed++
+		l.mu.Unlock()
+	}
+}
+
+// TODO: Implement logger function in a way
+// 1. incoming messages are queued to the channel
+// 2. if queue is full, increment dropped
+func (l *Logger) Log(msg string) {
+
+}
+
+// graceful shutdown
+func (l *Logger) Close() {
+	close(l.ch)
+	l.wg.Wait()
+}
+
+func (l *Logger) Stats() (processed, dropped int) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	return l.processed, l.dropped
+}

internal/exercises/templates/46_non_blocking_channel_operations/non_blocking_channel_operations_test.go

@@ -0,0 +1,49 @@
+package non_blocking_channel_operations
+
+import (
+	"testing"
+	"time"
+)
+
+func TestNonBlockingChannelOperations(t *testing.T) {
+
+	t.Run("Logs are processed without dropping", func(t *testing.T) {
+
+		processingDelay = 20 * time.Microsecond
+
+		logger := NewLogger(10)
+
+		for range 10 {
+			logger.Log("msg")
+		}
+		logger.Close()
+
+		processed, dropped := logger.Stats()
+
+		if dropped != 0 {
+			t.Fatalf("Expected no dropped logs, got %d", dropped)
+		}
+		if processed != 10 {
+			t.Fatalf("Expected 10 processed logs, got %d", processed)
+		}
+	})
+
+	t.Run("Logs are not dropped", func(t *testing.T) {
+		logger := NewLogger(5)
+
+		for range 10 {
+			logger.Log("msg")
+		}
+
+		logger.Close()
+
+		processed, dropped := logger.Stats()
+
+		if dropped != 5 {
+			t.Fatalf("Expected no dropped logs, got %d", dropped)
+		}
+		if processed != 5 {
+			t.Fatalf("Expected 10 processed logs, got %d", processed)
+		}
+	})
+}