Merge pull request #2 from Gerfey/v0.7.0

Author: Gerfey

README.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.6.0` is a pre-release version — feel free to test and report issues!
+> ⚠️ `v0.7.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)
 
@@ -25,7 +25,7 @@
 ## 📦 Installation
 > Requires Go 1.24+
 ```bash
-go get github.com/gerfey/messenger@v0.6.0
+go get github.com/gerfey/messenger@v0.7.0
 ```
 
 ## 🚀 Quick Start
@@ -60,13 +60,6 @@ default_bus: default
 
 buses:
   default: ~
-
-transports:
-  sync:
-    dsn: "in-memory://"
-
-routing:
-  main.HelloMessage: sync
 ```
 > 💡 If no transport is configured for a message, it will be executed synchronously by default (inline handler execution).
 
@@ -75,7 +68,7 @@ routing:
 ```go
 cfg, errConfig := config.LoadConfig("messenger.yaml")
 if errConfig != nil {
-    l.Error("ERROR load config", "error", errConfig)
+    fmt.Println("ERROR load config", "error", errConfig)
     return
 }
 
@@ -88,8 +81,6 @@ go m.Run(ctx)
 
 bus, _ := m.GetDefaultBus()
 _, _ = bus.Dispatch(ctx, &HelloMessage{Text: "World"})
-
-time.Sleep(5 * time.Second)
 ```
 
 ## 🔍 More Examples

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.6.0` — это пре-релиз. Тестируйте и сообщайте о багах!
+> ⚠️ Версия `v0.7.0` — это пре-релиз. Тестируйте и сообщайте о багах!
 
 > 📚 Полная документация доступна на [GitHub Wiki](https://github.com/Gerfey/messenger/wiki/Documentation)
 
@@ -27,7 +27,7 @@
 ## 📦 Установка
 > Требуется Go версии **1.24+**
 ```bash
-go get github.com/gerfey/messenger@v0.6.0
+go get github.com/gerfey/messenger@v0.7.0
 ```
 
 ## 🚀 Быстрый старт
@@ -62,13 +62,6 @@ default_bus: default
 
 buses:
   default: ~
-
-transports:
-  sync:
-    dsn: "in-memory://"
-
-routing:
-  main.HelloMessage: sync
 ```
 > 💡 Если транспорт для сообщения не указан — оно будет выполнено синхронно (inline).
 
@@ -77,7 +70,7 @@ routing:
 ```go
 cfg, errConfig := config.LoadConfig("messenger.yaml")
 if errConfig != nil {
-    l.Error("ERROR load config", "error", errConfig)
+    fmt.Println("ERROR load config", "error", errConfig)
     return
 }
 
@@ -90,8 +83,6 @@ go m.Run(ctx)
 
 bus, _ := m.GetDefaultBus()
 _, _ = bus.Dispatch(ctx, &HelloMessage{Text: "World"})
-
-time.Sleep(5 * time.Second)
 ```
 
 ## 🔍 Больше примеров

api/transport.go

@@ -2,33 +2,35 @@ package api
 
 import (
 	"context"
+	"reflect"
 
 	"github.com/gerfey/messenger/config"
 )
 
 type Transport interface {
-	Name() string
-	Send(context.Context, Envelope) error
-	Receive(context.Context, func(context.Context, Envelope) error) error
-}
-
-type RetryableTransport interface {
-	Transport
-	Retry(context.Context, Envelope) error
+	Sender
+	Receiver
 }
 
 type Sender interface {
+	Name() string
 	Send(context.Context, Envelope) error
 }
 
 type Receiver interface {
 	Receive(context.Context, func(context.Context, Envelope) error) error
 }
 
-type TransportLocator interface {
-	Register(string, Transport) error
-	GetAllTransports() []Transport
-	GetTransport(string) Transport
+type RetryableTransport interface {
+	Transport
+	Retry(context.Context, Envelope) error
+}
+
+type SenderLocator interface {
+	Register(string, Sender) error
+	GetSenders(Envelope) []Sender
+	RegisterMessageType(reflect.Type, []string)
+	SetFallback([]string)
 }
 
 type TransportFactory interface {

builder/builder.go

@@ -21,14 +21,15 @@ import (
 	"github.com/gerfey/messenger/transport"
 	"github.com/gerfey/messenger/transport/amqp"
 	"github.com/gerfey/messenger/transport/inmemory"
+	"github.com/gerfey/messenger/transport/sync"
 )
 
 type Builder struct {
 	cfg               *config.MessengerConfig
 	resolver          api.TypeResolver
 	transportFactory  *transport.FactoryChain
 	handlersLocator   api.HandlerLocator
-	transportLocator  api.TransportLocator
+	senderLocator     api.SenderLocator
 	middlewareLocator api.MiddlewareLocator
 	busLocator        api.BusLocator
 	eventDispatcher   api.EventDispatcher
@@ -38,19 +39,22 @@ type Builder struct {
 func NewBuilder(cfg *config.MessengerConfig, logger *slog.Logger) api.Builder {
 	resolver := NewResolver()
 
+	busLocator := bus.NewLocator()
+
 	tf := transport.NewFactoryChain(
-		amqp.NewTransportFactory(resolver, logger),
-		inmemory.NewTransportFactory(resolver, logger),
+		amqp.NewTransportFactory(logger, resolver),
+		inmemory.NewTransportFactory(logger, resolver),
+		sync.NewTransportFactory(logger, busLocator),
 	)
 
 	return &Builder{
 		cfg:               cfg,
 		resolver:          resolver,
 		transportFactory:  tf,
 		handlersLocator:   handler.NewHandlerLocator(),
-		transportLocator:  transport.NewLocator(),
+		senderLocator:     transport.NewSenderLocator(),
 		middlewareLocator: middleware.NewMiddlewareLocator(),
-		busLocator:        bus.NewLocator(),
+		busLocator:        busLocator,
 		eventDispatcher:   event.NewEventDispatcher(logger),
 		logger:            logger,
 	}
@@ -91,25 +95,16 @@ func (b *Builder) RegisterListener(event any, listener any) {
 }
 
 func (b *Builder) Build() (api.Messenger, error) {
-	router := routing.NewRouter()
-	for msgTypeStr, transportName := range b.cfg.Routing {
-		t, err := b.handlersLocator.ResolveMessageType(msgTypeStr)
-		if err != nil {
-			return nil, fmt.Errorf("failed to resolve message type '%s' in routing configuration: %w", msgTypeStr, err)
-		}
-		router.RouteTypeTo(t, transportName)
-	}
-
 	b.registerStamps()
 
-	if err := b.setupBuses(router); err != nil {
+	if err := b.setupBuses(); err != nil {
 		return nil, err
 	}
 
-	return b.createMessenger(router)
+	return b.createMessenger()
 }
 
-func (b *Builder) setupBuses(router api.Router) error {
+func (b *Builder) setupBuses() error {
 	for name, cfg := range b.cfg.Buses {
 		var chain []api.Middleware
 
@@ -124,9 +119,9 @@ func (b *Builder) setupBuses(router api.Router) error {
 		chain = append(chain, implementation.NewAddBusNameMiddleware(name))
 		chain = append(
 			chain,
-			implementation.NewSendMessageMiddleware(router, b.transportLocator, b.eventDispatcher, b.logger),
+			implementation.NewSendMessageMiddleware(b.logger, b.senderLocator, b.eventDispatcher),
 		)
-		chain = append(chain, implementation.NewHandleMessageMiddleware(b.handlersLocator, b.logger))
+		chain = append(chain, implementation.NewHandleMessageMiddleware(b.logger, b.handlersLocator))
 
 		createNewBus := bus.NewBus(chain...)
 
@@ -139,7 +134,48 @@ func (b *Builder) setupBuses(router api.Router) error {
 	return nil
 }
 
-func (b *Builder) createMessenger(router api.Router) (api.Messenger, error) {
+func (b *Builder) createMessenger() (api.Messenger, error) {
+	router, err := b.setupRouting()
+	if err != nil {
+		return nil, err
+	}
+
+	busMap := b.createBusMap()
+	handlerManager := b.createHandlerManager(busMap)
+	manager := transport.NewManager(b.logger, handlerManager, b.eventDispatcher)
+
+	createdTransports, transportNames, err := b.createTransports(manager)
+	if err != nil {
+		return nil, err
+	}
+
+	b.setupFallbackTransports(transportNames)
+	b.setupRetryListeners(createdTransports)
+
+	defaultBus, ok := b.busLocator.Get(b.cfg.DefaultBus)
+	if !ok {
+		return nil, fmt.Errorf("default_bus %q not found", defaultBus)
+	}
+
+	return messenger.NewMessenger(b.cfg.DefaultBus, manager, b.busLocator, router), nil
+}
+
+func (b *Builder) setupRouting() (api.Router, error) {
+	router := routing.NewRouter()
+
+	for msgTypeStr, transportName := range b.cfg.Routing {
+		t, err := b.handlersLocator.ResolveMessageType(msgTypeStr)
+		if err != nil {
+			return nil, fmt.Errorf("failed to resolve message type '%s' in routing configuration: %w", msgTypeStr, err)
+		}
+		router.RouteTypeTo(t, transportName)
+		b.senderLocator.RegisterMessageType(t, []string{transportName})
+	}
+
+	return router, nil
+}
+
+func (b *Builder) createBusMap() map[reflect.Type]string {
 	busMap := make(map[reflect.Type]string)
 	for _, h := range b.handlersLocator.GetAll() {
 		busName := h.BusName
@@ -149,7 +185,11 @@ func (b *Builder) createMessenger(router api.Router) (api.Messenger, error) {
 		busMap[h.InputType] = busName
 	}
 
-	handlerManager := func(ctx context.Context, env api.Envelope) error {
+	return busMap
+}
+
+func (b *Builder) createHandlerManager(busMap map[reflect.Type]string) func(context.Context, api.Envelope) error {
+	return func(ctx context.Context, env api.Envelope) error {
 		msgType := reflect.TypeOf(env.Message())
 		busName, ok := busMap[msgType]
 		if !ok {
@@ -165,27 +205,47 @@ func (b *Builder) createMessenger(router api.Router) (api.Messenger, error) {
 
 		return err
 	}
+}
 
-	manager := transport.NewManager(handlerManager, b.eventDispatcher, b.logger)
+func (b *Builder) createTransports(manager *transport.Manager) (map[string]api.Transport, []string, error) {
+	var transportNames []string
+	createdTransports := make(map[string]api.Transport)
+
+	b.createdSyncTransport(createdTransports)
 
 	for name, tCfg := range b.cfg.Transports {
 		tr, err := b.transportFactory.CreateTransport(name, tCfg)
 		if err != nil {
-			return nil, fmt.Errorf("failed to create transport '%s': %w", name, err)
+			return nil, nil, fmt.Errorf("failed to create transport '%s': %w", name, err)
 		}
 
+		createdTransports[name] = tr
+	}
+
+	for nameTransport, tr := range createdTransports {
 		manager.AddTransport(tr)
+		transportNames = append(transportNames, nameTransport)
 
-		errTransportLocator := b.transportLocator.Register(name, tr)
+		errTransportLocator := b.senderLocator.Register(nameTransport, tr)
 		if errTransportLocator != nil {
-			return nil, fmt.Errorf("failed to register transport '%s': %w", name, errTransportLocator)
+			return nil, nil, fmt.Errorf("failed to register transport '%s': %w", nameTransport, errTransportLocator)
 		}
 	}
 
+	return createdTransports, transportNames, nil
+}
+
+func (b *Builder) setupFallbackTransports(transportNames []string) {
+	if len(b.cfg.Routing) == 0 && len(transportNames) > 0 {
+		b.senderLocator.SetFallback(transportNames)
+	}
+}
+
+func (b *Builder) setupRetryListeners(createdTransports map[string]api.Transport) {
 	for name, tCfg := range b.cfg.Transports {
-		tr := b.transportLocator.GetTransport(name)
+		t := createdTransports[name]
 
-		if retryable, ok := tr.(api.RetryableTransport); ok && tCfg.RetryStrategy != nil {
+		if retryable, ok := t.(api.RetryableTransport); ok && tCfg.RetryStrategy != nil {
 			strategy := retry.NewMultiplierRetryStrategy(
 				tCfg.RetryStrategy.MaxRetries,
 				tCfg.RetryStrategy.Delay,
@@ -195,23 +255,29 @@ func (b *Builder) createMessenger(router api.Router) (api.Messenger, error) {
 
 			var failureTransport api.Transport
 			if b.cfg.FailureTransport != "" {
-				failureTransport = b.transportLocator.GetTransport(b.cfg.FailureTransport)
+				if ft, exists := createdTransports[b.cfg.FailureTransport]; exists {
+					failureTransport = ft
+				}
 			}
 
-			lst := listener.NewSendFailedMessageForRetryListener(retryable, failureTransport, strategy, b.logger)
+			lst := listener.NewSendFailedMessageForRetryListener(b.logger, retryable, failureTransport, strategy)
 			b.eventDispatcher.AddListener(event.SendFailedMessageEvent{}, lst)
 		}
 	}
-
-	defaultBus, ok := b.busLocator.Get(b.cfg.DefaultBus)
-	if !ok {
-		return nil, fmt.Errorf("default_bus %q not found", defaultBus)
-	}
-
-	return messenger.NewMessenger(b.cfg.DefaultBus, manager, b.busLocator, router), nil
 }
 
 func (b *Builder) registerStamps() {
 	b.resolver.RegisterStamp(stamps.BusNameStamp{})
 	b.resolver.RegisterStamp(stamps.RedeliveryStamp{})
 }
+
+func (b *Builder) createdSyncTransport(createdTransports map[string]api.Transport) {
+	cfg := config.TransportConfig{
+		DSN:     "sync://",
+		Options: config.OptionsConfig{},
+	}
+
+	if syncTransport, err := b.transportFactory.CreateTransport("sync", cfg); err == nil {
+		createdTransports["sync"] = syncTransport
+	}
+}