Go redis queue worker

Author: Nikhil

.idea/.gitignore

@@ -0,0 +1,67 @@
+# Redis Queue Implementation in Go
+
+This repository contains an implementation of a Redis queue in Go. It includes separate files for the Redis client, worker, and an API to push data into the queue.
+
+## Prerequisites
+
+- Go 1.16 or higher installed
+- Redis installed and running on your local machine
+
+## Getting Started
+
+1. Clone the repository:
+
+   ```
+   git clone <repository_url>
+   ```
+
+2. Install the required dependencies:
+
+   ```
+   go get github.com/go-redis/redis/v8
+   ```
+
+3. Start the Redis server on your local machine. If Redis is not installed, you can install it using Homebrew on macOS:
+
+   ```
+   brew install redis
+   ```
+
+4. Start the Redis server:
+
+   ```
+   brew services start redis
+   ```
+
+5. Build and run the Go program:
+
+   ```
+   go build
+   ./<binary_file_name>
+   ```
+
+   The server and worker will start running. The worker will listen for tasks in the Redis queue, and the API server will be available on port 8080.
+
+## Pushing Tasks to the Queue
+
+To push tasks to the Redis queue, you can make HTTP requests to the `/push` endpoint of the API server. The tasks will be added to the Redis queue and processed by the worker.
+
+Send a POST request to `http://localhost:8080/push?task=<your_task_here>` to push a task to the queue. Replace `<your_task_here>` with the task you want to add.
+
+Example using cURL:
+
+```
+curl -X POST "http://localhost:8080/push?task=example_task"
+```
+
+## Stopping the Program
+
+To stop the program gracefully, send a termination signal to the process. You can do this by pressing Ctrl+C in the terminal where the program is running.
+
+## Customization
+
+Feel free to customize the code according to your requirements. You can add error handling, authentication, or enhance the worker's processing logic.
+
+## License
+
+This project is licensed under the MIT License. See the [LICENSE](LICENSE) file for details.

.idea/go-redis-queue.iml

@@ -0,0 +1,9 @@
+module go-redis-queue
+
+go 1.20
+
+require (
+	github.com/cespare/xxhash/v2 v2.1.2 // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
+	github.com/go-redis/redis/v8 v8.11.5 // indirect
+)

.idea/modules.xml

@@ -0,0 +1,6 @@
+github.com/cespare/xxhash/v2 v2.1.2 h1:YRXhKfTDauu4ajMg1TPgFO5jnlC2HCbmLXMcTG5cbYE=
+github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
+github.com/go-redis/redis/v8 v8.11.5 h1:AcZZR7igkdvfVmQTPnu9WE37LRrO/YrBH5zWyjDC0oI=
+github.com/go-redis/redis/v8 v8.11.5/go.mod h1:gREzHqY1hg6oD9ngVRbLStwAWKhA0FEgq8Jd4h5lpwo=

README.md

@@ -0,0 +1,82 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"net/http"
+	"os"
+	"os/signal"
+	"sync"
+	"syscall"
+
+	"github.com/go-redis/redis/v8"
+)
+
+var (
+	ctx    context.Context
+	cancel context.CancelFunc
+)
+
+func main() {
+	// Create a new Redis client
+	redisClient := NewRedisClient()
+
+	// Create a context to handle cancellation
+	ctx, cancel = context.WithCancel(context.Background())
+
+	// Create a wait group to wait for all workers to finish
+	var wg sync.WaitGroup
+
+	// Create a worker
+	worker := NewWorker(redisClient)
+
+	// Start the worker goroutine
+	wg.Add(1)
+	go worker.Start(&wg)
+
+	// Start the API server
+	go startAPIServer(redisClient)
+
+	fmt.Println("Server started.")
+
+	// Wait for termination signal
+	waitForTerminationSignal()
+
+	// Stop the worker
+	cancel()
+	wg.Wait()
+
+	fmt.Println("Server stopped.")
+}
+
+func startAPIServer(redisClient *redis.Client) {
+	http.HandleFunc("/push", func(w http.ResponseWriter, r *http.Request) {
+		task := r.URL.Query().Get("task")
+		if task == "" {
+			http.Error(w, "Missing task parameter", http.StatusBadRequest)
+			return
+		}
+
+		// Push the task to the Redis queue
+		err := redisClient.RPush(ctx, "myqueue", task).Err()
+		if err != nil {
+			log.Println("Failed to push task to Redis queue:", err)
+			http.Error(w, "Failed to push task to Redis queue", http.StatusInternalServerError)
+			return
+		}
+
+		fmt.Fprintf(w, "Task %s pushed to the Redis queue", task)
+	})
+
+	err := http.ListenAndServe(":8080", nil)
+	if err != nil {
+		log.Fatal("Failed to start API server:", err)
+	}
+}
+
+func waitForTerminationSignal() {
+	signals := make(chan os.Signal, 1)
+	signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)
+	<-signals
+}

go-redis-queue

@@ -0,0 +1,13 @@
+package main
+
+import "github.com/go-redis/redis/v8"
+
+// NewRedisClient creates a new Redis client instance
+func NewRedisClient() *redis.Client {
+	client := redis.NewClient(&redis.Options{
+		Addr:     "localhost:6379", // Redis server address
+		Password: "",               // Redis password
+		DB:       0,                // Redis database index
+	})
+	return client
+}

go.mod

@@ -0,0 +1,47 @@
+package main
+
+import (
+	"fmt"
+	"log"
+	"sync"
+	"time"
+
+	"github.com/go-redis/redis/v8"
+)
+
+// Worker represents a worker that processes tasks from the Redis queue
+type Worker struct {
+	client *redis.Client
+}
+
+// NewWorker creates a new Worker instance
+func NewWorker(client *redis.Client) *Worker {
+	return &Worker{
+		client: client,
+	}
+}
+
+// Start starts the worker and listens for tasks in the Redis queue
+func (w *Worker) Start(wg *sync.WaitGroup) {
+	defer wg.Done()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			// Retrieve a task from the Redis queue
+			task, err := w.client.BLPop(ctx, 0, "myqueue").Result()
+			if err != nil {
+				log.Println("Failed to retrieve task:", err)
+				continue
+			}
+
+			// Process the task
+			fmt.Println("Processing task:", task[1])
+
+			// Simulate some work
+			time.Sleep(1 * time.Second)
+		}
+	}
+}