Добавление корректного статуса

internal/adapter/buffer.go

@@ -17,26 +17,33 @@ type RingBuffer struct {
 	totalBytes atomic.Uint64
 
 	// ===== SPEED =====
-	lastTick  atomic.Int64
-	lastBytes atomic.Uint64
+	speedMu      sync.Mutex
+	lastCalcTime time.Time
+	lastCalcBytes uint64
 
-	bytesPerSec atomic.Value // float64
-	bitsPerSec  atomic.Value // float64
+	// Текущие значения скорости
+	currentBPS atomic.Value // float64
+	currentBPSBits atomic.Value // float64
+
+	// Для скользящего окна
+	recentBytes []timestampedByte
+	recentMu    sync.Mutex
+}
+
+type timestampedByte struct {
+	timestamp time.Time
 }
 
 // =====================
 func NewRingBuffer(size int) *RingBuffer {
 	rb := &RingBuffer{
-		data: make([]byte, size),
+		data:          make([]byte, size),
+		lastCalcTime:  time.Now(),
+		recentBytes:   make([]timestampedByte, 0, 1000), // храним последние 1000 байт
 	}
 
-	now := time.Now().UnixNano()
-
-	rb.lastWrite.Store(now)
-	rb.lastTick.Store(now)
-
-	rb.bytesPerSec.Store(float64(0))
-	rb.bitsPerSec.Store(float64(0))
+	rb.currentBPS.Store(float64(0))
+	rb.currentBPSBits.Store(float64(0))
 
 	return rb
 }
@@ -55,34 +62,103 @@ func (rb *RingBuffer) Write(b byte) {
 	rb.mu.Unlock()
 
 	// ===== ATOMIC STATS =====
-	now := time.Now().UnixNano()
-	rb.lastWrite.Store(now)
+	now := time.Now()
+	rb.lastWrite.Store(now.UnixNano())
+	rb.totalBytes.Add(1)
 
-	total := rb.totalBytes.Add(1)
+	// ===== SPEED CALC (обновляем редко) =====
+	rb.updateSpeedPeriodically(now)
 
-	// ===== SPEED CALC =====
-	lastTick := rb.lastTick.Load()
-	if lastTick == 0 {
-		rb.lastTick.Store(now)
-		rb.lastBytes.Store(total)
+	// ===== Для скользящего окна =====
+	rb.addTimestampedByte(now)
+}
+
+// Периодическое обновление скорости (простое среднее)
+func (rb *RingBuffer) updateSpeedPeriodically(now time.Time) {
+	rb.speedMu.Lock()
+	defer rb.speedMu.Unlock()
+
+	elapsed := now.Sub(rb.lastCalcTime)
+
+	// Обновляем не чаще чем раз в 250мс
+	if elapsed < 250*time.Millisecond {
 		return
 	}
 
-	elapsed := time.Duration(now - lastTick).Seconds()
-	if elapsed < 1.0 {
-		return
+	currentTotal := rb.totalBytes.Load()
+	bytesInPeriod := currentTotal - rb.lastCalcBytes
+
+	if elapsed.Seconds() > 0 {
+		bps := float64(bytesInPeriod) / elapsed.Seconds()
+		rb.currentBPS.Store(bps)
+		rb.currentBPSBits.Store(bps * 8)
 	}
 
-	last := rb.lastBytes.Load()
-	diff := total - last
+	rb.lastCalcTime = now
+	rb.lastCalcBytes = currentTotal
+}
 
-	bps := float64(diff) / elapsed
+// Скользящее окно
+func (rb *RingBuffer) addTimestampedByte(now time.Time) {
+	rb.recentMu.Lock()
+	defer rb.recentMu.Unlock()
 
-	rb.bytesPerSec.Store(bps)
-	rb.bitsPerSec.Store(bps * 8)
+	// Добавляем новую запись
+	rb.recentBytes = append(rb.recentBytes, timestampedByte{timestamp: now})
 
-	rb.lastTick.Store(now)
-	rb.lastBytes.Store(total)
+	// Удаляем старые записи (старше 1 секунды)
+	cutoff := now.Add(-1 * time.Second)
+	cutIdx := 0
+	for i, tb := range rb.recentBytes {
+		if tb.timestamp.After(cutoff) {
+			cutIdx = i
+			break
+		}
+	}
+
+	if cutIdx > 0 {
+		rb.recentBytes = rb.recentBytes[cutIdx:]
+	}
+
+	// Вычисляем скорость на основе скользящего окна
+	if len(rb.recentBytes) > 1 {
+		windowDuration := rb.recentBytes[len(rb.recentBytes)-1].timestamp.Sub(rb.recentBytes[0].timestamp)
+		if windowDuration.Seconds() > 0 {
+			bps := float64(len(rb.recentBytes)) / windowDuration.Seconds()
+			rb.currentBPS.Store(bps)
+			rb.currentBPSBits.Store(bps * 8)
+		}
+	}
+}
+
+// Получить скорость на основе скользящего окна (альтернативный метод)
+func (rb *RingBuffer) GetWindowSpeed() (bytesPerSec, bitsPerSec float64) {
+	rb.recentMu.Lock()
+	defer rb.recentMu.Unlock()
+
+	now := time.Now()
+	cutoff := now.Add(-1 * time.Second)
+
+	// Считаем байты за последнюю секунду
+	count := 0
+	for i := len(rb.recentBytes) - 1; i >= 0; i-- {
+		if rb.recentBytes[i].timestamp.Before(cutoff) {
+			break
+		}
+		count++
+	}
+
+	if count > 0 {
+		// Находим время самого старого байта в окне
+		oldest := rb.recentBytes[len(rb.recentBytes)-count].timestamp
+		duration := now.Sub(oldest).Seconds()
+		if duration > 0 {
+			bytesPerSec = float64(count) / duration
+			bitsPerSec = bytesPerSec * 8
+		}
+	}
+
+	return
 }
 
 // =====================
@@ -142,8 +218,8 @@ func (rb *RingBuffer) Stats() map[string]interface{} {
 	filled := rb.count
 	rb.mu.RUnlock()
 
-	bps, _ := rb.bytesPerSec.Load().(float64)
-	bits, _ := rb.bitsPerSec.Load().(float64)
+	bps, _ := rb.currentBPS.Load().(float64)
+	bits, _ := rb.currentBPSBits.Load().(float64)
 
 	total := rb.totalBytes.Load()
 
@@ -159,4 +235,4 @@ func (rb *RingBuffer) Stats() map[string]interface{} {
 		"bytes_per_sec": bps,
 		"bits_per_sec":  bits,
 	}
-}
+}