128 lines
3.2 KiB
Go
128 lines
3.2 KiB
Go
package broker
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"sync"
|
|
|
|
"git.warky.dev/wdevs/pgsql-broker/pkg/broker/adapter"
|
|
"git.warky.dev/wdevs/pgsql-broker/pkg/broker/config"
|
|
)
|
|
|
|
// Broker manages multiple database instances
|
|
type Broker struct {
|
|
config *config.Config
|
|
logger adapter.Logger
|
|
version string
|
|
instances []*DatabaseInstance
|
|
ctx context.Context
|
|
cancel context.CancelFunc
|
|
shutdown bool
|
|
mu sync.RWMutex
|
|
}
|
|
|
|
// New creates a new broker that manages multiple database connections
|
|
func New(cfg *config.Config, logger adapter.Logger, version string) (*Broker, error) {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
|
|
broker := &Broker{
|
|
config: cfg,
|
|
logger: logger.With("component", "broker"),
|
|
version: version,
|
|
instances: make([]*DatabaseInstance, 0, len(cfg.Databases)),
|
|
ctx: ctx,
|
|
cancel: cancel,
|
|
}
|
|
|
|
return broker, nil
|
|
}
|
|
|
|
// Start begins all database instances
|
|
func (b *Broker) Start() error {
|
|
b.logger.Info("starting broker", "database_count", len(b.config.Databases))
|
|
|
|
// Create and start an instance for each database
|
|
for i, dbCfg := range b.config.Databases {
|
|
b.logger.Info("starting database instance", "name", dbCfg.Name, "host", dbCfg.Host, "database", dbCfg.Database)
|
|
|
|
// Create database adapter
|
|
dbAdapter := adapter.NewPostgresAdapter(dbCfg.ToPostgresConfig(), b.logger)
|
|
|
|
// Create database instance
|
|
instance, err := NewDatabaseInstance(b.config, &dbCfg, dbAdapter, b.logger, b.version, b.ctx)
|
|
if err != nil {
|
|
// Stop any already-started instances
|
|
b.stopInstances()
|
|
return fmt.Errorf("failed to create database instance %d (%s): %w", i, dbCfg.Name, err)
|
|
}
|
|
|
|
// Start the instance
|
|
if err := instance.Start(); err != nil {
|
|
// Stop any already-started instances
|
|
b.stopInstances()
|
|
return fmt.Errorf("failed to start database instance %d (%s): %w", i, dbCfg.Name, err)
|
|
}
|
|
|
|
b.instances = append(b.instances, instance)
|
|
b.logger.Info("database instance started", "name", dbCfg.Name, "instance_id", instance.ID)
|
|
}
|
|
|
|
b.logger.Info("broker started successfully", "database_instances", len(b.instances))
|
|
return nil
|
|
}
|
|
|
|
// Stop gracefully stops all database instances
|
|
func (b *Broker) Stop() error {
|
|
b.mu.Lock()
|
|
if b.shutdown {
|
|
b.mu.Unlock()
|
|
return nil
|
|
}
|
|
b.shutdown = true
|
|
b.mu.Unlock()
|
|
|
|
b.logger.Info("stopping broker")
|
|
|
|
// Cancel context
|
|
b.cancel()
|
|
|
|
// Stop all instances
|
|
b.stopInstances()
|
|
|
|
b.logger.Info("broker stopped")
|
|
return nil
|
|
}
|
|
|
|
// stopInstances stops all database instances
|
|
func (b *Broker) stopInstances() {
|
|
var wg sync.WaitGroup
|
|
for _, instance := range b.instances {
|
|
wg.Add(1)
|
|
go func(inst *DatabaseInstance) {
|
|
defer wg.Done()
|
|
if err := inst.Stop(); err != nil {
|
|
b.logger.Error("failed to stop instance", "name", inst.DatabaseName, "error", err)
|
|
}
|
|
}(instance)
|
|
}
|
|
wg.Wait()
|
|
}
|
|
|
|
// GetStats returns statistics for all database instances
|
|
func (b *Broker) GetStats() map[string]interface{} {
|
|
b.mu.RLock()
|
|
defer b.mu.RUnlock()
|
|
|
|
stats := map[string]interface{}{
|
|
"database_count": len(b.instances),
|
|
}
|
|
|
|
instanceStats := make(map[string]interface{})
|
|
for _, instance := range b.instances {
|
|
instanceStats[instance.DatabaseName] = instance.GetStats()
|
|
}
|
|
stats["instances"] = instanceStats
|
|
|
|
return stats
|
|
}
|