docs: Added AMQP transport benchmarks and updated documentation for v0.8.0

Author: Gerfey

README.md

@@ -7,14 +7,14 @@
 [![Go Reference](https://pkg.go.dev/badge/github.com/Gerfey/messenger.svg)](https://pkg.go.dev/github.com/Gerfey/messenger)
 [![License](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE)
 
-> `v0.7.0` is a pre-release version — feel free to test and report issues!
+> `v0.8.0` is a pre-release version — feel free to test and report issues!
 
 > Full documentation available in the [GitHub Wiki](https://github.com/Gerfey/messenger/wiki/Documentation)
 
 🇷🇺 [Русская версия](README.ru.md)
 
 ## Features
-- **Multiple Transports**: AMQP (RabbitMQ), In-Memory (sync)
+- **Multiple Transports**: AMQP (RabbitMQ), Kafka, Redis (Stream), In-Memory (sync)
 - **Middleware Chain**: Extensible middleware system for message processing
 - **Event-Driven**: Built-in event dispatcher for lifecycle hooks
 - **Retry Mechanism**: Configurable retry strategies with exponential backoff
@@ -25,7 +25,7 @@
 ## Installation
 > Requires Go 1.24+
 ```bash
-go get github.com/gerfey/messenger@v0.7.0
+go get github.com/gerfey/messenger@v0.8.0
 ```
 
 ## Quick Start
@@ -93,6 +93,10 @@ _, _ = bus.Dispatch(ctx, &HelloMessage{Text: "World"})
 
 > See [Usage Scenarios](https://github.com/Gerfey/messenger/wiki/Usage-Scenarios) for commands, queries, return values and advanced use-cases.
 
+## Efficiency
+
+- AMQP (RabbitMQ): [AMQP Benchmark Report](docs/benchmark/AMQP-Benchmark.md)
+
 ## Contributing
 
 1. Fork the repository
@@ -101,7 +105,7 @@ _, _ = bus.Dispatch(ctx, &HelloMessage{Text: "World"})
 4. Push to the branch (`git push origin feature/amazing-feature`)
 5. Open a Pull Request
 
-## ⚖License
+## License
 
 This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
 

README.ru.md

@@ -7,7 +7,7 @@
 [![Go Reference](https://pkg.go.dev/badge/github.com/Gerfey/messenger.svg)](https://pkg.go.dev/github.com/Gerfey/messenger)
 [![License](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE)
 
-> Версия `v0.7.0` — это пре-релиз. Тестируйте и сообщайте о багах!
+> Версия `v0.8.0` — это пре-релиз. Тестируйте и сообщайте о багах!
 
 > Полная документация доступна на [GitHub Wiki](https://github.com/Gerfey/messenger/wiki/Documentation)
 
@@ -16,7 +16,7 @@
 ---
 
 ## Возможности
-- **Множественные транспорты**: AMQP (RabbitMQ), In-Memory (`sync`)
+- **Множественные транспорты**: AMQP (RabbitMQ), Kafka, Redis (Stream), In-Memory (sync)
 - **Цепочка middleware**: Расширяемая система промежуточной обработки
 - **Событийный движок**: Встроенный dispatcher событий жизненного цикла
 - **Механизм повторов**: Настраиваемые стратегии ретраев с поддержкой DLQ
@@ -27,7 +27,7 @@
 ## Установка
 > Требуется Go версии **1.24+**
 ```bash
-go get github.com/gerfey/messenger@v0.7.0
+go get github.com/gerfey/messenger@v0.8.0
 ```
 
 ## Быстрый старт
@@ -95,6 +95,10 @@ _, _ = bus.Dispatch(ctx, &HelloMessage{Text: "World"})
 
 > Смотри [Сценарии использования](https://github.com/Gerfey/messenger/wiki/Сценарии-использования).
 
+## Производительность
+
+- AMQP (RabbitMQ): [AMQP Benchmark Report](docs/benchmark/AMQP-Benchmark.md)
+
 ## Как внести вклад
 
 1. Форкните репозиторий

docs/benchmark/AMQP-Benchmark.md

@@ -0,0 +1,45 @@
+# AMQP Benchmark Report
+
+Current performance results of the AMQP transport in Messenger (`v0.8.0`), tested with `RabbitMQ` using the `amqp091-go` client.
+
+## Overall Performance
+
+| Benchmark                     | Time (ns/op) | Throughput (msg/sec) | Memory (B/op) | Allocs/op |
+|------------------------------|---------------|---------------------|----------------|-----------|
+| `BenchmarkAMQPSend`          | 465,508       | ~2,148              | 13,822         | 267       |
+| `BenchmarkAMQPConcurrentSend`| 296,922       | ~3,368              | 13,829         | 266       |
+
+*Concurrent sending is ~36% faster with 100B messages.*
+
+---
+
+## Message Size Impact
+
+| Message Size | Time (ns/op) | Throughput (msg/sec) | Memory (B/op) | Allocs/op |
+|------------------|---------------|---------------------|----------------|-----------|
+| 100 B            | 472,831       | ~2,115              | 13,826         | 267       |
+| 1 KB             | 557,170       | ~1,794              | 19,874         | 267       |
+| 10 KB            | 704,831       | ~1,418              | 82,185         | 269       |
+| 100 KB           | 1,788,004     | ~559                | 726,393        | 276       |
+
+*As the payload size increases, throughput decreases and memory pressure on GC grows, as expected.*
+
+---
+
+## Allocations: `pprof` Analysis
+> Collected using `go test -bench=BenchmarkAMQP -benchmem -memprofile mem.out`, analyzed via `pprof`.
+
+(pprof) top
+
+- encoding/json: ~20%
+- amqp091-go (sendOpen, Ack, readLongstr): ~25%
+- Envelope.WithStamp: 5.75%
+- Middleware chain: ~5%
+
+---
+
+## Summary
+
+- AMQP transport in Messenger demonstrates stable and predictable performance
+- Memory and allocation optimization opportunities are being addressed in upcoming versions
+

transport/amqp/benchmark_test.go

@@ -0,0 +1,178 @@
+package amqp_test
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"os"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/gerfey/messenger/api"
+	"github.com/gerfey/messenger/builder"
+	"github.com/gerfey/messenger/config"
+)
+
+const (
+	benchmarkAMQPDSN = "amqp://guest:guest@localhost:5672/"
+	benchmarkTimeout = 60 * time.Second
+)
+
+type BenchmarkMessage struct {
+	ID      string
+	Content string
+	Data    []byte
+}
+
+func (m *BenchmarkMessage) RoutingKey() string {
+	return "benchmark_routing_key"
+}
+
+type BenchmarkHandler struct {
+	processedCount *int64
+	wg             *sync.WaitGroup
+}
+
+func (h *BenchmarkHandler) Handle(_ context.Context, _ *BenchmarkMessage) error {
+	atomic.AddInt64(h.processedCount, 1)
+	if h.wg != nil {
+		h.wg.Done()
+	}
+
+	return nil
+}
+
+func (h *BenchmarkHandler) GetBusName() string {
+	return "default"
+}
+
+func setupAMQPMessenger(b *testing.B, withWaitGroup bool) api.MessageBus {
+	b.Helper()
+
+	logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{
+		Level: slog.LevelError,
+	}))
+
+	cfg := &config.MessengerConfig{
+		DefaultBus: "default",
+		Buses: map[string]config.BusConfig{
+			"default": {},
+		},
+		Transports: map[string]config.TransportConfig{
+			"amqp": {
+				DSN:        benchmarkAMQPDSN,
+				Serializer: "default.transport.serializer",
+				Options: map[string]any{
+					"auto_setup": true,
+					"exchange": map[string]any{
+						"name": "benchmark_exchange",
+						"type": "topic",
+					},
+					"queues": map[string]any{
+						"benchmark_queue": map[string]any{
+							"binding_keys": []string{"benchmark_routing_key"},
+						},
+					},
+				},
+			},
+		},
+		Routing: map[string]string{
+			"*amqp_test.BenchmarkMessage": "amqp",
+		},
+	}
+
+	processedCount := int64(0)
+	var wg *sync.WaitGroup
+	if withWaitGroup {
+		wg = &sync.WaitGroup{}
+	}
+
+	handler := &BenchmarkHandler{
+		processedCount: &processedCount,
+		wg:             wg,
+	}
+
+	builderInstance := builder.NewBuilder(cfg, logger)
+	if err := builderInstance.RegisterHandler(handler); err != nil {
+		b.Fatalf("Register handler failed: %v", err)
+	}
+
+	messenger, err := builderInstance.Build()
+	if err != nil {
+		b.Fatalf("Build messenger failed: %v", err)
+	}
+
+	ctx, cancel := context.WithTimeout(b.Context(), benchmarkTimeout)
+	go func() {
+		defer cancel()
+		if runErr := messenger.Run(ctx); runErr != nil && !errors.Is(runErr, context.Canceled) {
+			b.Logf("Messenger run error: %v", runErr)
+		}
+	}()
+
+	time.Sleep(2 * time.Second)
+
+	bus, err := messenger.GetDefaultBus()
+	if err != nil {
+		b.Fatalf("Get default bus failed: %v", err)
+	}
+
+	return bus
+}
+
+func dispatchMessages(b *testing.B, bus api.MessageBus, size int, parallel bool) {
+	ctx := b.Context()
+	b.ResetTimer()
+	b.ReportAllocs()
+
+	if parallel {
+		concurrency := 10
+		var wg sync.WaitGroup
+		messagesPerWorker := b.N / concurrency
+		for w := range concurrency {
+			wg.Add(1)
+			go func(id int) {
+				defer wg.Done()
+				for i := range messagesPerWorker {
+					bus.Dispatch(ctx, &BenchmarkMessage{
+						ID:      fmt.Sprintf("worker-%d-msg-%d", id, i),
+						Content: "benchmark content",
+						Data:    make([]byte, size),
+					})
+				}
+			}(w)
+		}
+		wg.Wait()
+	} else {
+		for i := range b.N {
+			bus.Dispatch(ctx, &BenchmarkMessage{
+				ID:      fmt.Sprintf("msg-%d", i),
+				Content: "benchmark content",
+				Data:    make([]byte, size),
+			})
+		}
+	}
+}
+
+func BenchmarkAMQPSend(b *testing.B) {
+	bus := setupAMQPMessenger(b, false)
+	dispatchMessages(b, bus, 100, false)
+}
+
+func BenchmarkAMQPConcurrentSend(b *testing.B) {
+	bus := setupAMQPMessenger(b, false)
+	dispatchMessages(b, bus, 100, true)
+}
+
+func BenchmarkAMQPMessageSizes(b *testing.B) {
+	sizes := []int{100, 1024, 10240, 102400}
+	for _, size := range sizes {
+		b.Run(fmt.Sprintf("Size_%dB", size), func(b *testing.B) {
+			bus := setupAMQPMessenger(b, false)
+			dispatchMessages(b, bus, size, false)
+		})
+	}
+}