package bwtest import ( "context" "encoding/json" "fmt" "os" "sync" ) const DefaultConfigPath = "/etc/proxmox-router/bwtest.json" // Manager coordinates the BwTest server and client subsystems type Manager struct { mu sync.RWMutex configPath string server *BwTestServer client *BwTestClient history []*HistoricalTestRecord } // NewManager initializes the Bandwidth Test manager and starts the background server func NewManager(cfgPath string) *Manager { if cfgPath == "" { cfgPath = DefaultConfigPath } cfg := ServerConfig{ Enabled: true, Port: 2000, ListenAddr: "0.0.0.0", AllowAnonymous: true, } m := &Manager{ configPath: cfgPath, history: make([]*HistoricalTestRecord, 0, 50), } // Try loading persisted configuration if loaded, err := m.loadConfig(); err == nil { cfg = *loaded } m.server = NewServer(cfg) m.client = NewClient() if cfg.Enabled { if err := m.server.Start(); err != nil { fmt.Printf("[BwTestManager] Warning: failed to start server on port %d: %v\n", cfg.Port, err) } } return m } func (m *Manager) loadConfig() (*ServerConfig, error) { data, err := os.ReadFile(m.configPath) if err != nil { return nil, err } var cfg ServerConfig if err := json.Unmarshal(data, &cfg); err != nil { return nil, err } return &cfg, nil } func (m *Manager) saveConfigLocked(cfg ServerConfig) error { _ = os.MkdirAll("/etc/proxmox-router", 0755) data, err := json.MarshalIndent(cfg, "", " ") if err != nil { return err } return os.WriteFile(m.configPath, data, 0644) } // GetServerStatus returns metrics and status for the local BwTest server func (m *Manager) GetServerStatus() ServerStatus { return m.server.Status() } // SetServerConfig updates server configuration (enabled state, port) func (m *Manager) SetServerConfig(cfg ServerConfig) error { m.mu.Lock() defer m.mu.Unlock() if cfg.Port <= 0 { cfg.Port = 2000 } if cfg.ListenAddr == "" { cfg.ListenAddr = "0.0.0.0" } m.server.Stop() m.server.config = cfg if cfg.Enabled { if err := m.server.Start(); err != nil { return err } } return m.saveConfigLocked(cfg) } // StartTest initiates an asynchronous bandwidth test against the specified target func (m *Manager) StartTest(ctx context.Context, req SessionRequest) error { return m.client.Run(ctx, req) } // StopTest terminates any ongoing bandwidth test func (m *Manager) StopTest() { m.client.Stop() } // GetTestStatus retrieves current live test status and telemetry samples func (m *Manager) GetTestStatus() TestStatus { st := m.client.GetStatus() // If test completed with a summary, persist into history if !st.Running && st.Summary != nil { m.mu.Lock() alreadyExists := false for _, h := range m.history { if h.ID == st.SessionID { alreadyExists = true break } } if !alreadyExists && st.SessionID != "" { record := &HistoricalTestRecord{ ID: st.SessionID, Target: fmt.Sprintf("%s:%d", st.Target, st.Port), Protocol: st.Protocol, Direction: st.Direction, Timestamp: st.StartTime, Summary: st.Summary, } m.history = append([]*HistoricalTestRecord{record}, m.history...) if len(m.history) > 50 { m.history = m.history[:50] } } m.mu.Unlock() } return st } // SubscribeClient returns a stream of real-time telemetry events func (m *Manager) SubscribeClient() <-chan LiveSample { return m.client.Subscribe() } // GetHistory returns past completed bandwidth test results func (m *Manager) GetHistory() []*HistoricalTestRecord { m.mu.RLock() defer m.mu.RUnlock() res := make([]*HistoricalTestRecord, len(m.history)) copy(res, m.history) return res } // Stop shuts down the manager and server listeners cleanly func (m *Manager) Stop() { m.client.Stop() m.server.Stop() }