---
title: "Go Middleware and Concurrency Patterns"
url: "https://vanducng.dev/2024/04/05/Go-Middleware-and-Concurrency-Patterns/"
date: "2024-04-05"
series: "go-backend"
categories: ["Backend"]
tags: ["go","api"]
---

# Go Middleware and Concurrency Patterns

Middleware and concurrency are two powerful features that make Go excellent for backend development. Middleware enables cross-cutting concerns like logging and authentication, while Go's goroutines and channels provide elegant solutions for concurrent processing. This post explores both patterns in depth.

## Understanding Middleware

Middleware intercepts requests before they reach handlers, enabling common functionality across all routes.

```mermaid
flowchart LR
    R[Request] --> M1[Logger]
    M1 --> M2[Auth]
    M2 --> M3[CORS]
    M3 --> H[Handler]
    H --> M3
    M3 --> M2
    M2 --> M1
    M1 --> Res[Response]

    classDef blueClass fill:#4A90E2,stroke:#333,stroke-width:2px,color:#fff
    classDef orangeClass fill:#F39C12,stroke:#333,stroke-width:2px,color:#fff
    classDef greenClass fill:#27AE60,stroke:#333,stroke-width:2px,color:#fff

    class M1 blueClass
    class M2 orangeClass
    class M3 greenClass
```

### Basic Middleware Structure

```go
func MyMiddleware() gin.HandlerFunc {
    return func(c *gin.Context) {
        // Before request
        start := time.Now()

        // Process request
        c.Next()

        // After request
        duration := time.Since(start)
        log.Printf("Request took %v", duration)
    }
}
```

## Common Middleware Patterns

### Logging Middleware

```go
func LoggerMiddleware() gin.HandlerFunc {
    return func(c *gin.Context) {
        start := time.Now()
        path := c.Request.URL.Path

        // Process request
        c.Next()

        // Log after response
        latency := time.Since(start)
        status := c.Writer.Status()
        clientIP := c.ClientIP()
        method := c.Request.Method

        log.Printf("[%s] %s %s %d %v",
            method, path, clientIP, status, latency)
    }
}
```

### Request ID Middleware

```go
func RequestIDMiddleware() gin.HandlerFunc {
    return func(c *gin.Context) {
        requestID := c.GetHeader("X-Request-ID")
        if requestID == "" {
            requestID = uuid.New().String()
        }

        c.Set("request_id", requestID)
        c.Header("X-Request-ID", requestID)

        c.Next()
    }
}

// Access in handlers
func handler(c *gin.Context) {
    requestID := c.GetString("request_id")
    log.Printf("[%s] Processing request", requestID)
}
```

### Recovery Middleware

```go
func RecoveryMiddleware() gin.HandlerFunc {
    return func(c *gin.Context) {
        defer func() {
            if err := recover(); err != nil {
                // Log stack trace
                log.Printf("Panic recovered: %v\n%s", err, debug.Stack())

                c.JSON(http.StatusInternalServerError, gin.H{
                    "error": "Internal server error",
                })
                c.Abort()
            }
        }()
        c.Next()
    }
}
```

### CORS Middleware

```go
func CORSMiddleware() gin.HandlerFunc {
    return func(c *gin.Context) {
        c.Header("Access-Control-Allow-Origin", "*")
        c.Header("Access-Control-Allow-Methods", "GET, POST, PUT, PATCH, DELETE, OPTIONS")
        c.Header("Access-Control-Allow-Headers", "Origin, Content-Type, Authorization")
        c.Header("Access-Control-Max-Age", "86400")

        if c.Request.Method == "OPTIONS" {
            c.AbortWithStatus(http.StatusNoContent)
            return
        }

        c.Next()
    }
}
```

### Timeout Middleware

```go
func TimeoutMiddleware(timeout time.Duration) gin.HandlerFunc {
    return func(c *gin.Context) {
        ctx, cancel := context.WithTimeout(c.Request.Context(), timeout)
        defer cancel()

        c.Request = c.Request.WithContext(ctx)

        finished := make(chan struct{})

        go func() {
            c.Next()
            close(finished)
        }()

        select {
        case <-finished:
            // Request completed normally
        case <-ctx.Done():
            c.JSON(http.StatusGatewayTimeout, gin.H{
                "error": "Request timeout",
            })
            c.Abort()
        }
    }
}
```

### Applying Middleware

```go
func main() {
    router := gin.New() // Start without default middleware

    // Global middleware
    router.Use(RecoveryMiddleware())
    router.Use(LoggerMiddleware())
    router.Use(CORSMiddleware())

    // Group-specific middleware
    api := router.Group("/api")
    api.Use(RequestIDMiddleware())
    {
        api.GET("/public", publicHandler)

        // Protected routes
        protected := api.Group("/")
        protected.Use(AuthMiddleware())
        {
            protected.GET("/users", listUsers)
            protected.POST("/users", createUser)
        }
    }

    router.Run(":8080")
}
```

## Concurrency in Go

### Goroutines

Goroutines are lightweight threads managed by Go runtime:

```go
func main() {
    // Start goroutine
    go processTask("task1")

    // Multiple goroutines
    for i := 0; i < 10; i++ {
        go func(id int) {
            fmt.Printf("Processing task %d\n", id)
        }(i) // Pass i as argument to avoid closure issues
    }

    time.Sleep(time.Second) // Wait for goroutines
}
```

### Channels

Channels enable safe communication between goroutines:

```go
func main() {
    // Unbuffered channel
    ch := make(chan string)

    go func() {
        ch <- "Hello from goroutine"
    }()

    message := <-ch
    fmt.Println(message)

    // Buffered channel
    buffered := make(chan int, 3)
    buffered <- 1
    buffered <- 2
    buffered <- 3
    // buffered <- 4 // Would block!

    fmt.Println(<-buffered) // 1
}
```

```mermaid
flowchart LR
    subgraph Goroutine1["Goroutine 1"]
        G1[Producer]
    end

    subgraph Channel["Channel"]
        C[Buffer]
    end

    subgraph Goroutine2["Goroutine 2"]
        G2[Consumer]
    end

    G1 -->|Send| C
    C -->|Receive| G2

    classDef blueClass fill:#4A90E2,stroke:#333,stroke-width:2px,color:#fff

    class Channel blueClass
```

### Select Statement

Handle multiple channels:

```go
func processWithTimeout(data string) (string, error) {
    resultCh := make(chan string)
    errCh := make(chan error)

    go func() {
        // Simulate work
        time.Sleep(100 * time.Millisecond)
        resultCh <- "Processed: " + data
    }()

    select {
    case result := <-resultCh:
        return result, nil
    case err := <-errCh:
        return "", err
    case <-time.After(500 * time.Millisecond):
        return "", errors.New("timeout")
    }
}
```

## Concurrency Patterns for APIs

### Background Processing

```go
func (h *Handler) CreateOrder(c *gin.Context) {
    var order Order
    if err := c.ShouldBindJSON(&order); err != nil {
        c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
        return
    }

    // Save order
    if err := h.repo.Create(&order); err != nil {
        c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to create order"})
        return
    }

    // Background processing (non-blocking)
    go func() {
        // Send confirmation email
        h.emailService.SendOrderConfirmation(order)

        // Update inventory
        h.inventoryService.DeductStock(order.Items)

        // Notify warehouse
        h.notificationService.NotifyWarehouse(order)
    }()

    // Return immediately
    c.JSON(http.StatusCreated, gin.H{
        "message": "Order created",
        "order":   order,
    })
}
```

### Parallel Data Fetching

```go
type DashboardData struct {
    User     *User
    Orders   []Order
    Stats    *Statistics
    Notifications []Notification
}

func (h *Handler) GetDashboard(c *gin.Context) {
    userID := c.GetUint("user_id")
    ctx := c.Request.Context()

    var (
        user          *User
        orders        []Order
        stats         *Statistics
        notifications []Notification
        wg            sync.WaitGroup
        errCh         = make(chan error, 4)
    )

    wg.Add(4)

    // Fetch user
    go func() {
        defer wg.Done()
        var err error
        user, err = h.userRepo.FindByID(ctx, userID)
        if err != nil {
            errCh <- err
        }
    }()

    // Fetch orders
    go func() {
        defer wg.Done()
        var err error
        orders, err = h.orderRepo.FindByUserID(ctx, userID)
        if err != nil {
            errCh <- err
        }
    }()

    // Fetch stats
    go func() {
        defer wg.Done()
        var err error
        stats, err = h.statsService.GetUserStats(ctx, userID)
        if err != nil {
            errCh <- err
        }
    }()

    // Fetch notifications
    go func() {
        defer wg.Done()
        var err error
        notifications, err = h.notificationRepo.FindUnread(ctx, userID)
        if err != nil {
            errCh <- err
        }
    }()

    // Wait for all goroutines
    wg.Wait()
    close(errCh)

    // Check for errors
    for err := range errCh {
        if err != nil {
            c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to load dashboard"})
            return
        }
    }

    c.JSON(http.StatusOK, DashboardData{
        User:          user,
        Orders:        orders,
        Stats:         stats,
        Notifications: notifications,
    })
}
```

### Worker Pool Pattern

```go
type Job struct {
    ID   int
    Data string
}

type Result struct {
    JobID  int
    Output string
    Error  error
}

func worker(id int, jobs <-chan Job, results chan<- Result) {
    for job := range jobs {
        // Process job
        output, err := processJob(job)
        results <- Result{
            JobID:  job.ID,
            Output: output,
            Error:  err,
        }
    }
}

func ProcessBatch(items []string, workerCount int) []Result {
    jobs := make(chan Job, len(items))
    results := make(chan Result, len(items))

    // Start workers
    for w := 1; w <= workerCount; w++ {
        go worker(w, jobs, results)
    }

    // Send jobs
    for i, item := range items {
        jobs <- Job{ID: i, Data: item}
    }
    close(jobs)

    // Collect results
    var output []Result
    for i := 0; i < len(items); i++ {
        output = append(output, <-results)
    }

    return output
}
```

```mermaid
flowchart TD
    subgraph Input["Job Queue"]
        J1[Job 1]
        J2[Job 2]
        J3[Job 3]
        J4[Job 4]
    end

    subgraph Workers["Worker Pool"]
        W1[Worker 1]
        W2[Worker 2]
        W3[Worker 3]
    end

    subgraph Output["Results"]
        R1[Result 1]
        R2[Result 2]
        R3[Result 3]
        R4[Result 4]
    end

    Input --> Workers
    Workers --> Output

    classDef blueClass fill:#4A90E2,stroke:#333,stroke-width:2px,color:#fff

    class Workers blueClass
```

### Rate Limiting with Channels

```go
type RateLimiter struct {
    ticker *time.Ticker
    tokens chan struct{}
}

func NewRateLimiter(rate int, burst int) *RateLimiter {
    rl := &RateLimiter{
        ticker: time.NewTicker(time.Second / time.Duration(rate)),
        tokens: make(chan struct{}, burst),
    }

    // Fill initial burst
    for i := 0; i < burst; i++ {
        rl.tokens <- struct{}{}
    }

    // Refill tokens
    go func() {
        for range rl.ticker.C {
            select {
            case rl.tokens <- struct{}{}:
            default: // Bucket full
            }
        }
    }()

    return rl
}

func (rl *RateLimiter) Wait() {
    <-rl.tokens
}

func RateLimitMiddleware(limiter *RateLimiter) gin.HandlerFunc {
    return func(c *gin.Context) {
        limiter.Wait()
        c.Next()
    }
}
```

## Context Propagation

### Passing Context Through Middleware

```go
func ContextMiddleware() gin.HandlerFunc {
    return func(c *gin.Context) {
        // Add values to context
        ctx := context.WithValue(c.Request.Context(), "request_time", time.Now())

        // Replace request context
        c.Request = c.Request.WithContext(ctx)

        c.Next()
    }
}

// Use in handlers
func handler(c *gin.Context) {
    ctx := c.Request.Context()

    // Pass context to services
    result, err := service.Process(ctx, data)

    // Context for database operations
    db.WithContext(ctx).Find(&users)
}
```

### Handling Cancellation

```go
func (h *Handler) LongRunningTask(c *gin.Context) {
    ctx := c.Request.Context()

    resultCh := make(chan string)

    go func() {
        result := h.service.ProcessLongTask(ctx)
        resultCh <- result
    }()

    select {
    case result := <-resultCh:
        c.JSON(http.StatusOK, gin.H{"result": result})
    case <-ctx.Done():
        // Client disconnected or timeout
        log.Println("Request cancelled:", ctx.Err())
        return
    }
}
```

## Error Handling in Concurrent Code

```go
type SafeResult struct {
    mu     sync.Mutex
    errors []error
}

func (sr *SafeResult) AddError(err error) {
    sr.mu.Lock()
    defer sr.mu.Unlock()
    sr.errors = append(sr.errors, err)
}

func (sr *SafeResult) HasErrors() bool {
    sr.mu.Lock()
    defer sr.mu.Unlock()
    return len(sr.errors) > 0
}

// Usage with errgroup (recommended)
import "golang.org/x/sync/errgroup"

func FetchAll(ctx context.Context) (*Data, error) {
    g, ctx := errgroup.WithContext(ctx)

    var users []User
    var orders []Order

    g.Go(func() error {
        var err error
        users, err = fetchUsers(ctx)
        return err
    })

    g.Go(func() error {
        var err error
        orders, err = fetchOrders(ctx)
        return err
    })

    if err := g.Wait(); err != nil {
        return nil, err
    }

    return &Data{Users: users, Orders: orders}, nil
}
```

## Summary

| Pattern | Use Case |
|---------|----------|
| **Middleware** | Cross-cutting concerns (logging, auth, CORS) |
| **Goroutines** | Lightweight concurrent execution |
| **Channels** | Safe communication between goroutines |
| **Select** | Multi-channel operations, timeouts |
| **Worker Pool** | Controlled parallelism for batch processing |
| **Context** | Cancellation, deadlines, request-scoped values |

```mermaid
flowchart TD
    subgraph Middleware["Middleware Chain"]
        M1[Logger] --> M2[Recovery]
        M2 --> M3[Auth]
        M3 --> M4[RateLimit]
    end

    subgraph Concurrency["Concurrent Processing"]
        G1[Goroutine] --> C[Channel]
        G2[Goroutine] --> C
        C --> H[Handler]
    end

    M4 --> Concurrency

    classDef blueClass fill:#4A90E2,stroke:#333,stroke-width:2px,color:#fff
    classDef greenClass fill:#27AE60,stroke:#333,stroke-width:2px,color:#fff

    class Middleware blueClass
    class Concurrency greenClass
```

---

*Next post: [Scaling Go APIs](https://vanducng.dev/2024/04/12/Scaling-Go-APIs-Pagination-Caching-Rate-Limiting/) - Pagination, caching, and performance optimization.*