Initial commit

Author: avalkov

Diff for: .env

@@ -0,0 +1,6 @@
+AMQP_HOST="rabbitmq"
+AMQP_PORT=5672
+AMQP_USER="user"
+AMQP_PASS="password"
+AMQP_QUEUE_NAME="commands_queue"
+PROCESSING_WORKERS_COUNT=3

Diff for: .gitignore

@@ -0,0 +1,23 @@
+# If you prefer the allow list template instead of the deny list, see community template:
+# https://github.com/github/gitignore/blob/main/community/Golang/Go.AllowList.gitignore
+#
+# Binaries for programs and plugins
+*.exe
+*.exe~
+*.dll
+*.so
+*.dylib
+
+# Test binary, built with `go test -c`
+*.test
+
+# Output of the go coverage tool, specifically when used with LiteIDE
+*.out
+
+# Dependency directories (remove the comment below to include it)
+# vendor/
+
+# Go workspace file
+go.work
+
+.netrc

Diff for: README.md

@@ -0,0 +1,14 @@
+# SCS - Simple Client-Server via RabbitMQ
+
+You can run it with docker-compose and the default .env:
+```
+docker-compose up -d
+```
+
+Uses CSP (Common Sequantial Processes) pattern to distribute the work between logically independant routines.
+After messages are received from AMQP, they are being sent to router that will use consistient hashing to distribute them 
+between set of pre-configured workers so all operations for same ```key``` are serialized to execute in same worker to avoid potential race conditions.
+
+The ```commands``` directory contains multiple files with commands. Which ones to be loaded by the client during test is specified in ```client/Dockerfile```. Each file will be loaded and executed concurrently to simulate multiple clients.
+
+After client finishes, it won't exit so we can attach to the running container and inspect the saved files.

Diff for: apps/client/main.go

@@ -0,0 +1,157 @@
+package main
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"log"
+	"os"
+	"regexp"
+	"time"
+
+	"github.com/avalkov/SCS/internal/configuration"
+	"github.com/joho/godotenv"
+	"github.com/sethvargo/go-envconfig"
+	"github.com/streadway/amqp"
+)
+
+type Message struct {
+	Body          string
+	ReplyTo       string
+	CorrelationId string
+}
+
+const (
+	replyQueue = "reply_queue"
+)
+
+var jsonArrayRegex = regexp.MustCompile(`^\s*\[\s*.*\s*\]\s*$`)
+
+func failOnError(err error, msg string) {
+	if err != nil {
+		log.Fatalf("%s: %s", msg, err)
+	}
+}
+
+func connectToRabbitMQ(user, password, host string, port int) (*amqp.Connection, *amqp.Channel) {
+	conn, err := amqp.Dial(fmt.Sprintf("amqp://%s:%s@%s:%d/", user, password, host, port))
+	failOnError(err, "Failed to connect to RabbitMQ")
+
+	ch, err := conn.Channel()
+	failOnError(err, "Failed to open a channel")
+
+	return conn, ch
+}
+
+func declareQueue(ch *amqp.Channel, name string) amqp.Queue {
+	q, err := ch.QueueDeclare(
+		name,  // name
+		false, // durable
+		false, // delete when unused
+		false, // exclusive
+		false, // no-wait
+		nil,   // arguments
+	)
+	failOnError(err, fmt.Sprintf("Failed to declare queue: %s", name))
+	return q
+}
+
+func loadCommandsFromFile(filename string) ([]string, error) {
+	file, err := os.Open(filename)
+	if err != nil {
+		return nil, err
+	}
+	defer file.Close()
+
+	var commands []string
+	scanner := bufio.NewScanner(file)
+	for scanner.Scan() {
+		commands = append(commands, scanner.Text())
+	}
+
+	if err := scanner.Err(); err != nil {
+		return nil, err
+	}
+
+	return commands, nil
+}
+
+func sendCommand(ch *amqp.Channel, command string, requestQueue, replyQueue string) {
+	corrId := fmt.Sprintf("%d", time.Now().UnixNano())
+
+	err := ch.Publish(
+		"",           // exchange
+		requestQueue, // routing key
+		false,        // mandatory
+		false,        // immediate
+		amqp.Publishing{
+			ContentType:   "text/plain",
+			CorrelationId: corrId,
+			ReplyTo:       replyQueue,
+			Body:          []byte(command),
+		})
+	failOnError(err, "Failed to publish a message")
+}
+
+func receiveReplies(ch *amqp.Channel, replyQueue string, clientID int) {
+	msgs, err := ch.Consume(
+		replyQueue, // queue
+		"",         // consumer
+		true,       // auto-ack
+		false,      // exclusive
+		false,      // no-local
+		false,      // no-wait
+		nil,        // args
+	)
+	failOnError(err, "Failed to register a consumer")
+
+	for d := range msgs {
+		fmt.Printf("Client %d received reply: %s\n", clientID, d.Body)
+		if jsonArrayRegex.Match(d.Body) {
+			saveToFile(fmt.Sprintf("getAllItemsResponse_client_%d.json", clientID), d.Body)
+		}
+	}
+}
+
+func saveToFile(filename string, data []byte) {
+	file, err := os.Create(filename)
+	failOnError(err, fmt.Sprintf("Failed to create file: %s", filename))
+	defer file.Close()
+
+	_, err = file.Write(data)
+	failOnError(err, fmt.Sprintf("Failed to write to file: %s", filename))
+}
+
+func main() {
+	if len(os.Args) < 2 {
+		log.Fatalf("Usage: %s <commands_file_1> <commands_file_2> ... <commands_file_n>", os.Args[0])
+	}
+
+	godotenv.Load(".env")
+
+	var config configuration.Config
+	if err := envconfig.Process(context.Background(), &config); err != nil {
+		log.Fatalf("Failed to process env var: %v", err)
+	}
+
+	conn, ch := connectToRabbitMQ(config.AMQP.User, config.AMQP.Pass, config.AMQP.Host, config.AMQP.Port)
+	defer conn.Close()
+
+	replyQueue := declareQueue(ch, replyQueue)
+
+	for i, filename := range os.Args[1:] {
+		log.Printf("Processing commands from file: %s", filename)
+		commands, err := loadCommandsFromFile(filename)
+		failOnError(err, fmt.Sprintf("Failed to load commands from file %s", filename))
+
+		go func(clientID int, commands []string) {
+			go receiveReplies(ch, replyQueue.Name, clientID)
+			for _, command := range commands {
+				sendCommand(ch, command, config.AMQP.QueueName, replyQueue.Name)
+			}
+		}(i, commands)
+	}
+
+	ctx := context.Background()
+	<-ctx.Done()
+}

Diff for: apps/server/main.go

@@ -0,0 +1,77 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"os"
+
+	"github.com/avalkov/SCS/internal/configuration"
+	ds "github.com/avalkov/SCS/internal/datastructures"
+	commandsParser "github.com/avalkov/SCS/internal/domain/commands_parser"
+	commandsProcessor "github.com/avalkov/SCS/internal/domain/commands_processor"
+	"github.com/avalkov/SCS/internal/multiworker"
+	"github.com/avalkov/SCS/internal/queueservice"
+	"github.com/avalkov/SCS/internal/queueservice/amqp"
+	"github.com/joho/godotenv"
+	"github.com/sethvargo/go-envconfig"
+)
+
+func main() {
+	if err := run(); err != nil {
+		fmt.Fprintf(os.Stderr, "%v\n", err)
+		os.Exit(1)
+	}
+}
+
+func run() error {
+	ctx, cancelFunc := context.WithCancel(context.Background())
+	defer cancelFunc()
+
+	godotenv.Load(".env")
+
+	var config configuration.Config
+	if err := envconfig.Process(ctx, &config); err != nil {
+		return err
+	}
+
+	amqpWorker := amqp.NewAmqpWorker(amqp.AmqpConfig{
+		Host:      config.AMQP.Host,
+		Port:      config.AMQP.Port,
+		User:      config.AMQP.User,
+		Pass:      config.AMQP.Pass,
+		QueueName: config.AMQP.QueueName,
+	})
+	if err := amqpWorker.Initialize(); err != nil {
+		return fmt.Errorf("failed to initialize AMQP worker: %v", err)
+	}
+
+	defer amqpWorker.Close()
+
+	requestsChans := make(chan queueservice.Message)
+	repliesChans := make(chan queueservice.Message)
+
+	defer close(requestsChans)
+	defer close(repliesChans)
+
+	if err := amqpWorker.Run(ctx, requestsChans, repliesChans); err != nil {
+		return fmt.Errorf("failed to run AMQP worker: %v", err)
+	}
+
+	parser := commandsParser.NewCommandsParser()
+	processor := commandsProcessor.NewCommandsProcessor(parser, ds.NewOrderedMap())
+
+	multiWorker := multiworker.NewMultiWorker(
+		config.PROCESSING_WORKERS_COUNT,
+		parser,
+		processor,
+	)
+
+	go multiWorker.Run(ctx, requestsChans, repliesChans)
+
+	log.Printf("Server started")
+
+	<-ctx.Done()
+
+	return nil
+}

Diff for: cleanup.sh

@@ -0,0 +1,29 @@
+#!/bin/bash
+
+# Function to delete a Docker container
+delete_container() {
+  container_name=$1
+  echo "Deleting container: $container_name"
+  docker rm -f $container_name
+}
+
+# Function to delete a Docker image
+delete_image() {
+  image_name=$1
+  echo "Deleting image: $image_name"
+  docker rmi $image_name
+}
+
+# Delete containers
+delete_container "scs_server"
+delete_container "scs_client"
+delete_container "rabbitmq"
+
+# Delete images
+delete_image "scs_server"
+delete_image "scs_client"
+
+# Delete volumes
+#docker volume rm scs_rabbitmq_data
+
+echo "Cleanup complete."

Diff for: client/Dockerfile

@@ -0,0 +1,20 @@
+FROM amd64/golang:1.22.3-bullseye AS client_builder
+
+WORKDIR /app
+
+COPY . .
+
+RUN go mod download
+
+RUN go build -o /app/scs_client /app/apps/client/main.go
+
+FROM debian:bullseye-slim
+
+WORKDIR /root/
+
+COPY --from=client_builder /app/scs_client /root/scs_client
+COPY ./commands /root/commands
+
+RUN chmod +x /root/scs_client
+
+ENTRYPOINT ["/root/scs_client", "/root/commands/commands1.txt", "/root/commands/commands2.txt", "/root/commands/commands3.txt"]

Diff for: commands/commands1.txt

@@ -0,0 +1,7 @@
+addItem('key1', 'val1')
+addItem('key2', 'val2')
+addItem('key3', 'val3')
+getItem('key1')
+deleteItem('key2')
+addItem('key4', 'val4')
+getAllItems()

Diff for: commands/commands2.txt

@@ -0,0 +1,7 @@
+deleteItem('key10')
+addItem('key10', 'val10')
+addItem('key20', 'val20')
+addItem('key30', 'val30')
+getItem('key10')
+addItem('key40', 'val40')
+getAllItems()

Diff for: commands/commands3.txt

@@ -0,0 +1,2 @@
+addItem('key33', 'val33')
+getAllItems()