feat(queue): serialisierte Ansage-Queue mit Prioritäten, now-Sprung, Mute-Store und Rate-Limit-Struktur
- Worker-Loop: eine Ansage gleichzeitig; now > high > normal > low, FIFO je Level - Cancel bricht die laufende Ansage ab (ctx) und leert die Queue (Zahl als Rückgabe) - Close verwirft wartende Items, bricht die laufende Ansage ab und wartet auf den Worker - MuteStore: persistenter Mute-State als Datei mute im Config-Dir (atomar geschrieben) - RateLimiter: Struktur + Semantik für Task 05, bewusst noch nicht angebunden (Task-04-Default unbegrenzt, dokumentierte Seam in rate.go)
This commit is contained in:
m3ta-chiron committed
1 parent
716f6b2fc9
commit
e492baddac
5 files changed
+742
-3
No files matched your search
@@ -1,6 +1,7 @@
|
||||
// Package queue implementiert die Ansage-Queue des Daemons: serialisierte
|
||||
// Abarbeitung (eine Ansage gleichzeitig), Prioritäten (high > normal > low,
|
||||
// FIFO je Level), now-Items springen vor, Rate-Limit mit prioritätsbewusstem
|
||||
// Verwerfen. Implementierung folgt in Task 04 (Queue) und Task 05
|
||||
// (Rate-Limit, Ruhezeiten).
|
||||
// FIFO je Level), now-Items springen vor, dazu den persistenten Mute-State
|
||||
// (Datei mute im Config-Dir) und die Rate-Limit-Struktur. Das Rate-Limit
|
||||
// ist als Struktur vorhanden, aber noch nicht angebunden — Wiring und
|
||||
// Ruhezeiten folgen in Task 05.
|
||||
package queue
|
||||
@@ -0,0 +1,84 @@
|
||||
package queue
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// MuteFileName ist der Name der Mute-Datei im Config-Verzeichnis (bindender
|
||||
// Vertrag: „persistenten Mute-State toggeln (Datei mute im Config-Dir)").
|
||||
const MuteFileName = "mute"
|
||||
|
||||
// ErrNoMutePath meldet einen Set-Versuch ohne konfigurierten Datei-Pfad
|
||||
// (Config-Dir nicht bestimmbar — der Daemon deaktiviert dann Mute mit einer
|
||||
// Start-Warnung statt zu scheitern).
|
||||
var ErrNoMutePath = errors.New("mute: kein Datei-Pfad konfiguriert (Config-Dir nicht bestimmbar)")
|
||||
|
||||
// MuteStore verwaltet den persistenten Mute-State als Datei im
|
||||
// Config-Verzeichnis: Inhalt „true" = stumm, alles andere (oder keine Datei)
|
||||
// = nicht stumm.
|
||||
//
|
||||
// Der Daemon liest die Datei bei jeder /speak-Anfrage neu — dadurch wirkt
|
||||
// auch ein im Degraded Mode (Daemon tot) von der CLI direkt geschriebener
|
||||
// State sofort, und der Zustand überlebt Daemon-Restarts.
|
||||
type MuteStore struct {
|
||||
// Path ist der vollständige Pfad der Mute-Datei. Leer (kein Config-Dir
|
||||
// bestimmbar): Muted() ist immer false, Set ein Fehler.
|
||||
Path string
|
||||
}
|
||||
|
||||
// Muted meldet den persistenten Mute-State. Eine fehlende oder unlesbare
|
||||
// Datei heißt: nicht stumm (Fehlstart ohne Config-Dir wird nie „stumm aus
|
||||
// Versehen").
|
||||
func (s MuteStore) Muted() bool {
|
||||
if s.Path == "" {
|
||||
return false
|
||||
}
|
||||
b, err := os.ReadFile(s.Path)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
return strings.TrimSpace(string(b)) == "true"
|
||||
}
|
||||
|
||||
// Set schreibt den Mute-State atomar (temporäre Datei + Rename, damit
|
||||
// Daemon und CLI nie eine halb geschriebene Datei lesen) und legt das
|
||||
// Config-Verzeichnis bei Bedarf an.
|
||||
func (s MuteStore) Set(muted bool) error {
|
||||
if s.Path == "" {
|
||||
return ErrNoMutePath
|
||||
}
|
||||
dir := filepath.Dir(s.Path)
|
||||
if err := os.MkdirAll(dir, 0o755); err != nil {
|
||||
return err
|
||||
}
|
||||
content := "false\n"
|
||||
if muted {
|
||||
content = "true\n"
|
||||
}
|
||||
tmp, err := os.CreateTemp(dir, ".mute-*")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
name := tmp.Name()
|
||||
if _, err := tmp.WriteString(content); err != nil {
|
||||
tmp.Close()
|
||||
os.Remove(name)
|
||||
return err
|
||||
}
|
||||
if err := tmp.Close(); err != nil {
|
||||
os.Remove(name)
|
||||
return err
|
||||
}
|
||||
if err := os.Chmod(name, 0o644); err != nil {
|
||||
os.Remove(name)
|
||||
return err
|
||||
}
|
||||
if err := os.Rename(name, s.Path); err != nil {
|
||||
os.Remove(name)
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,288 @@
|
||||
package queue
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"sort"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Priority eines Queue-Items — öffentlicher Vertrag der HTTP-API
|
||||
// (low | normal | high; leer wird als normal behandelt).
|
||||
type Priority string
|
||||
|
||||
// Erlaubte Prioritäts-Werte (Spec: high > normal > low, FIFO je Level).
|
||||
const (
|
||||
PriorityLow Priority = "low"
|
||||
PriorityNormal Priority = "normal"
|
||||
PriorityHigh Priority = "high"
|
||||
)
|
||||
|
||||
// Valid meldet, ob p ein erlaubter Prioritäts-Wert ist.
|
||||
func (p Priority) Valid() bool {
|
||||
switch p {
|
||||
case PriorityLow, PriorityNormal, PriorityHigh:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// ErrClosed meldet einen Enqueue-Versuch nach Close (Daemon fährt herunter).
|
||||
var ErrClosed = errors.New("queue: bereits geschlossen")
|
||||
|
||||
// Item ist eine Ansage in der Warteschlange. Alle Felder kommen aus der
|
||||
// HTTP-API (POST /speak); ID und Ablauf-Verwaltung liegen bei der Queue.
|
||||
type Item struct {
|
||||
// ID ist die kurze Ansage-Kennung (Antwort von POST /speak; Grundlage
|
||||
// des --wait-Pollings der CLI). Wird von Enqueue vergeben, wenn leer.
|
||||
ID string
|
||||
|
||||
// Text ist der Ansage-Text.
|
||||
Text string
|
||||
|
||||
// Lang ist die Sprach-Pin (de | en); leer = automatische Erkennung
|
||||
// durch den Daemon (langdetect).
|
||||
Lang string
|
||||
|
||||
// Provider pinnt die Provider-Kette auf eine Stufe (--provider); leer
|
||||
// = konfigurierte Default-Kette.
|
||||
Provider string
|
||||
|
||||
// Voice pinnt eine Voice-ID (--voice); leer = Voice-Mapping/Defaults.
|
||||
Voice string
|
||||
|
||||
// Priority steuert die Abarbeitungs-Reihenfolge (high > normal > low,
|
||||
// FIFO je Level).
|
||||
Priority Priority
|
||||
|
||||
// Now lässt die Ansage vor alle wartenden Items springen (auch vor
|
||||
// high) — sie unterbricht aber nie die laufende Ansage.
|
||||
Now bool
|
||||
|
||||
seq uint64 // Enqueue-Reihenfolge: FIFO-Tiebrecher innerhalb eines Levels
|
||||
}
|
||||
|
||||
// rank liefert die Abarbeitungs-Klasse: now über high über normal über low.
|
||||
func (it Item) rank() int {
|
||||
if it.Now {
|
||||
return 3
|
||||
}
|
||||
switch it.Priority {
|
||||
case PriorityHigh:
|
||||
return 2
|
||||
case PriorityNormal:
|
||||
return 1
|
||||
default:
|
||||
return 0
|
||||
}
|
||||
}
|
||||
|
||||
// Speaker spielt ein Item ab. Im Daemon: Provider-Kette bauen und Speak
|
||||
// aufrufen (Synthese mit Fallback + Wiedergabe über den Audio-Sink — die
|
||||
// Kette aus internal/provider wird unverändert wiederverwendet). ctx wird
|
||||
// abgebrochen, wenn die Ansage abgebrochen (POST /cancel) oder der Daemon
|
||||
// heruntergefahren wird; die Kette bricht ihre Netz- und Prozess-Arbeit
|
||||
// dann ab. Der Rückgabe-Fehler wird protokolliert, nie an die API gereicht.
|
||||
type Speaker func(ctx context.Context, item Item) error
|
||||
|
||||
// Queue ist die serialisierte Ansage-Queue des Daemons: genau eine Ansage
|
||||
// gleichzeitig, Reihenfolge now > high > normal > low (FIFO je Level).
|
||||
//
|
||||
// Die Synchronisation ist bewusst simpel und offensichtlich korrekt statt
|
||||
// clever: EIN Mutex + EIN Cond schützen alles (Items, aktuelles Item,
|
||||
// Closed-Flag, cancelCur). Der Worker ist der einzige, der Items entnimmt;
|
||||
// pop und das Setzen von current/cancelCur geschehen unter demselben Lock,
|
||||
// damit ein Item nie zwischen „wartet" und „läuft" der Cancelierung
|
||||
// entwischen kann. Cancel/Close rufen cancelCur unter dem Lock ab — der
|
||||
// Worker bemerkt die Abbrechung in seiner speak-Funktion über ctx.
|
||||
type Queue struct {
|
||||
speak Speaker
|
||||
log io.Writer
|
||||
|
||||
mu sync.Mutex
|
||||
notEmpty *sync.Cond // Signal: neues Item; Broadcast: Close
|
||||
items []Item
|
||||
current *Item
|
||||
cancelCur context.CancelFunc
|
||||
seq uint64
|
||||
closed bool
|
||||
done chan struct{} // geschlossen, wenn der Worker-Loop endet
|
||||
}
|
||||
|
||||
// New startet die Queue mit ihrem Worker-Loop (eine Ansage gleichzeitig).
|
||||
// log ist optional (nil = still) und empfängt die Lebenslauf-Zeilen der
|
||||
// Items. Close beendet den Loop und wartet auf sein Ende.
|
||||
func New(speak Speaker, log io.Writer) *Queue {
|
||||
q := &Queue{speak: speak, log: log, done: make(chan struct{})}
|
||||
q.notEmpty = sync.NewCond(&q.mu)
|
||||
go q.loop()
|
||||
return q
|
||||
}
|
||||
|
||||
// Enqueue stellt item in die Queue und liefert dessen ID (bei leerer ID neu
|
||||
// vergeben: kurz und zufällig). Nach Close ist Enqueue ein Fehler.
|
||||
func (q *Queue) Enqueue(item Item) (string, error) {
|
||||
q.mu.Lock()
|
||||
if q.closed {
|
||||
q.mu.Unlock()
|
||||
return "", ErrClosed
|
||||
}
|
||||
if item.Priority == "" {
|
||||
item.Priority = PriorityNormal
|
||||
}
|
||||
q.seq++
|
||||
item.seq = q.seq
|
||||
if item.ID == "" {
|
||||
item.ID = newID()
|
||||
}
|
||||
q.items = append(q.items, item)
|
||||
q.mu.Unlock()
|
||||
q.notEmpty.Signal()
|
||||
return item.ID, nil
|
||||
}
|
||||
|
||||
// Pending liefert eine Momentaufnahme der wartenden Items in
|
||||
// Abarbeitungs-Reihenfolge (bestes Item zuerst) — Basis für GET /status.
|
||||
func (q *Queue) Pending() []Item {
|
||||
q.mu.Lock()
|
||||
defer q.mu.Unlock()
|
||||
out := make([]Item, len(q.items))
|
||||
copy(out, q.items)
|
||||
sort.SliceStable(out, func(i, j int) bool { return itemBefore(out[i], out[j]) })
|
||||
return out
|
||||
}
|
||||
|
||||
// itemBefore meldet, ob a vor b abgearbeitet wird: höhere Klasse zuerst,
|
||||
// innerhalb derselben Klasse die frühere Enqueue-Reihenfolge.
|
||||
func itemBefore(a, b Item) bool {
|
||||
if a.rank() != b.rank() {
|
||||
return a.rank() > b.rank()
|
||||
}
|
||||
return a.seq < b.seq
|
||||
}
|
||||
|
||||
// Current liefert das gerade laufende Item (false: Queue ist idle).
|
||||
func (q *Queue) Current() (Item, bool) {
|
||||
q.mu.Lock()
|
||||
defer q.mu.Unlock()
|
||||
if q.current == nil {
|
||||
return Item{}, false
|
||||
}
|
||||
return *q.current, true
|
||||
}
|
||||
|
||||
// Cancel bricht die laufende Ansage ab und leert die Queue. Rückgabe ist
|
||||
// die Zahl der abgebrochenen Ansagen (laufende + wartende).
|
||||
func (q *Queue) Cancel() int {
|
||||
q.mu.Lock()
|
||||
defer q.mu.Unlock()
|
||||
n := len(q.items)
|
||||
q.items = nil
|
||||
if q.current != nil {
|
||||
n++
|
||||
if q.cancelCur != nil {
|
||||
q.cancelCur()
|
||||
}
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// Close fährt die Queue herunter: wartende Items werden verworfen, die
|
||||
// laufende Ansage abgebrochen, der Worker-Loop beendet. Close wartet, bis
|
||||
// der Loop sicher beendet ist (idempotent).
|
||||
func (q *Queue) Close() {
|
||||
q.mu.Lock()
|
||||
if q.closed {
|
||||
q.mu.Unlock()
|
||||
<-q.done
|
||||
return
|
||||
}
|
||||
q.closed = true
|
||||
q.items = nil
|
||||
if q.cancelCur != nil {
|
||||
q.cancelCur()
|
||||
}
|
||||
q.mu.Unlock()
|
||||
q.notEmpty.Broadcast()
|
||||
<-q.done
|
||||
}
|
||||
|
||||
// loop ist der Worker: wartet auf Items und spielt sie einzeln ab. Nach
|
||||
// Close (oder wenn Close während einer Ansage zuschlägt) beendet er sich,
|
||||
// sobald die laufende Ansage zurückgekehrt ist.
|
||||
func (q *Queue) loop() {
|
||||
defer close(q.done)
|
||||
for {
|
||||
q.mu.Lock()
|
||||
for len(q.items) == 0 && !q.closed {
|
||||
q.notEmpty.Wait()
|
||||
}
|
||||
if q.closed {
|
||||
q.mu.Unlock()
|
||||
return
|
||||
}
|
||||
item := q.popLocked()
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
q.current = &item
|
||||
q.cancelCur = cancel
|
||||
q.mu.Unlock()
|
||||
|
||||
q.logf("queue: ansage %s (%q, Priorität %s, now=%v) startet", item.ID, item.Text, item.Priority, item.Now)
|
||||
start := time.Now()
|
||||
err := q.speak(ctx, item)
|
||||
dur := time.Since(start)
|
||||
cancel()
|
||||
|
||||
q.mu.Lock()
|
||||
q.current = nil
|
||||
q.cancelCur = nil
|
||||
q.mu.Unlock()
|
||||
|
||||
switch {
|
||||
case ctx.Err() != nil:
|
||||
q.logf("queue: ansage %s abgebrochen", item.ID)
|
||||
case err != nil:
|
||||
q.logf("queue: ansage %s fehlgeschlagen nach %s: %v", item.ID, dur.Round(time.Millisecond), err)
|
||||
default:
|
||||
q.logf("queue: ansage %s fertig nach %s", item.ID, dur.Round(time.Millisecond))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// popLocked entnimmt das als nächstes fällige Item (mu muss gehalten werden).
|
||||
func (q *Queue) popLocked() Item {
|
||||
best := 0
|
||||
for i := 1; i < len(q.items); i++ {
|
||||
if itemBefore(q.items[i], q.items[best]) {
|
||||
best = i
|
||||
}
|
||||
}
|
||||
item := q.items[best]
|
||||
q.items = append(q.items[:best], q.items[best+1:]...)
|
||||
return item
|
||||
}
|
||||
|
||||
// newID liefert eine kurze, kollisionsarme ID (6 Hex-Zeichen). IDs müssen
|
||||
// nur pro Daemon-Lauf eindeutig sein (--wait-Polling); der Fallback ist
|
||||
// praktisch unerreichbar, aber deterministisch sicher.
|
||||
func newID() string {
|
||||
var b [3]byte
|
||||
if _, err := rand.Read(b[:]); err == nil {
|
||||
return hex.EncodeToString(b[:])
|
||||
}
|
||||
return fmt.Sprintf("id%d", time.Now().UnixNano())
|
||||
}
|
||||
|
||||
// logf schreibt eine Protokoll-Zeile, wenn ein Log gesetzt ist. Wird nie
|
||||
// unter dem Mutex gerufen (Aufrufer achten darauf).
|
||||
func (q *Queue) logf(format string, args ...any) {
|
||||
if q.log != nil {
|
||||
fmt.Fprintf(q.log, format+"\n", args...)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,309 @@
|
||||
package queue
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"slices"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// gesteuerteSpielerin ist ein testbarer Speaker: Jede Ansage meldet ihren
|
||||
// Start auf started (Kanal-Puffer: der Worker blockiert nie) und wartet
|
||||
// dann, bis der Test sie per step freigibt — so ist die Abarbeitung
|
||||
// vollständig deterministisch steuerbar. Ein abgebrochener ctx beendet das
|
||||
// Warten (Cancel-/Close-Pfade). Alle Felder werden nur vom Test gelesen;
|
||||
// started-Reihenfolge == Start-Reihenfolge.
|
||||
type gesteuerteSpielerin struct {
|
||||
started chan Item
|
||||
step chan struct{}
|
||||
consumed int // von warteBis konsumierte Starts (nur Test-Goroutine)
|
||||
}
|
||||
|
||||
func neueSpielerin() *gesteuerteSpielerin {
|
||||
return &gesteuerteSpielerin{started: make(chan Item, 16), step: make(chan struct{})}
|
||||
}
|
||||
|
||||
func (g *gesteuerteSpielerin) speak(ctx context.Context, item Item) error {
|
||||
g.started <- item
|
||||
select {
|
||||
case <-g.step:
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
// warteBis wartet, bis insgesamt n Ansagen gestartet sind, und liefert die
|
||||
// seit dem letzten warteBis-Aufruf neu gestarteten.
|
||||
func (g *gesteuerteSpielerin) warteBis(t *testing.T, n int) []Item {
|
||||
t.Helper()
|
||||
var items []Item
|
||||
deadline := time.After(2 * time.Second)
|
||||
for g.consumed < n {
|
||||
select {
|
||||
case it := <-g.started:
|
||||
g.consumed++
|
||||
items = append(items, it)
|
||||
case <-deadline:
|
||||
t.Fatalf("nur %d von %d Ansagen gestartet", g.consumed, n)
|
||||
}
|
||||
}
|
||||
return items
|
||||
}
|
||||
|
||||
// keineWeitere prüft, dass keine weitere Ansage startet (Poll-Fenster).
|
||||
func (g *gesteuerteSpielerin) keineWeitere(t *testing.T) {
|
||||
t.Helper()
|
||||
select {
|
||||
case it := <-g.started:
|
||||
g.consumed++
|
||||
t.Fatalf("unerwartete Ansage %q gestartet", it.Text)
|
||||
case <-time.After(150 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
|
||||
// TestAbarbeitungsReihenfolge: now vor high vor normal vor low, FIFO je
|
||||
// Level. Der erste Blocker besetzt den Worker; danach werden alle Level
|
||||
// gleichzeitig eingereiht — die Abarbeitung muss der Klasse folgen.
|
||||
func TestAbarbeitungsReihenfolge(t *testing.T) {
|
||||
g := neueSpielerin()
|
||||
q := New(g.speak, nil)
|
||||
defer q.Close()
|
||||
|
||||
erster, err := q.Enqueue(Item{Text: "blocker", Priority: PriorityNormal})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if erster == "" {
|
||||
t.Fatal("Enqueue vergibt keine ID")
|
||||
}
|
||||
g.warteBis(t, 1) // Blocker läuft, Worker gebunden.
|
||||
|
||||
for _, it := range []Item{
|
||||
{Text: "niedrig", Priority: PriorityLow},
|
||||
{Text: "zweit-niedrig", Priority: PriorityLow},
|
||||
{Text: "mittel", Priority: PriorityNormal},
|
||||
{Text: "dringend", Priority: PriorityHigh},
|
||||
{Text: "sofort", Now: true},
|
||||
} {
|
||||
if _, err := q.Enqueue(it); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
g.keineWeitere(t) // Serialisierung: Worker spielt noch den Blocker.
|
||||
|
||||
// Freigeben: alles Übrige läuft in Klassen-Reihenfolge durch.
|
||||
close(g.step)
|
||||
rest := g.warteBis(t, 6)
|
||||
|
||||
var texts []string
|
||||
for _, it := range rest {
|
||||
texts = append(texts, it.Text)
|
||||
}
|
||||
want := []string{"sofort", "dringend", "mittel", "niedrig", "zweit-niedrig"}
|
||||
if !slices.Equal(texts, want) {
|
||||
t.Errorf("Abarbeitung nach dem Blocker = %v, want %v", texts, want)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCurrentUndPending: Status-Sicht auf laufendes Item + Warteschlange.
|
||||
func TestCurrentUndPending(t *testing.T) {
|
||||
g := neueSpielerin()
|
||||
q := New(g.speak, nil)
|
||||
defer q.Close()
|
||||
|
||||
if _, ok := q.Current(); ok {
|
||||
t.Fatal("Current vor erster Ansage: want idle")
|
||||
}
|
||||
|
||||
q.Enqueue(Item{Text: "blocker"})
|
||||
g.warteBis(t, 1)
|
||||
|
||||
q.Enqueue(Item{Text: "hoch", Priority: PriorityHigh})
|
||||
q.Enqueue(Item{Text: "niedrig", Priority: PriorityLow})
|
||||
|
||||
if cur, ok := q.Current(); !ok || cur.Text != "blocker" {
|
||||
t.Errorf("Current = %+v (ok=%v), will Blocker", cur, ok)
|
||||
}
|
||||
var pendingTexts []string
|
||||
for _, it := range q.Pending() {
|
||||
pendingTexts = append(pendingTexts, it.Text)
|
||||
}
|
||||
if !slices.Equal(pendingTexts, []string{"hoch", "niedrig"}) {
|
||||
t.Errorf("Pending = %v, want [hoch niedrig]", pendingTexts)
|
||||
}
|
||||
|
||||
close(g.step)
|
||||
g.warteBis(t, 3)
|
||||
}
|
||||
|
||||
// TestCancel: laufende Ansage wird über ctx abgebrochen, Queue geleert,
|
||||
// Gesamtzahl zurückgemeldet; danach ist die Queue wieder aufnahme- und
|
||||
// spielbereit.
|
||||
func TestCancel(t *testing.T) {
|
||||
g := neueSpielerin()
|
||||
q := New(g.speak, nil)
|
||||
defer q.Close()
|
||||
|
||||
q.Enqueue(Item{Text: "blocker"})
|
||||
g.warteBis(t, 1)
|
||||
|
||||
q.Enqueue(Item{Text: "wartet-eins"})
|
||||
q.Enqueue(Item{Text: "wartet-zwei"})
|
||||
|
||||
if n := q.Cancel(); n != 3 {
|
||||
t.Errorf("Cancel = %d, want 3 (laufende + 2 wartende)", n)
|
||||
}
|
||||
|
||||
g.keineWeitere(t) // Blocker durch ctx beendet, wartende Items sind weg.
|
||||
waitIdle(t, q)
|
||||
|
||||
if n := q.Cancel(); n != 0 {
|
||||
t.Errorf("Cancel auf leerer Queue = %d, want 0", n)
|
||||
}
|
||||
|
||||
// Queue nimmt wieder an und spielt:
|
||||
id, err := q.Enqueue(Item{Text: "danach"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
g.step <- struct{}{} // gibt die wartende Ansage im Speaker frei
|
||||
gestartet := g.warteBis(t, 2)
|
||||
if len(gestartet) != 1 || gestartet[0].ID != id {
|
||||
t.Errorf("neue Ansage nicht gestartet: %+v", gestartet)
|
||||
}
|
||||
}
|
||||
|
||||
// TestClose: wartende Items werden verworfen, die laufende Ansage über den
|
||||
// abgebrochenen ctx beendet, Close wartet auf den Worker-Exit, danach ist
|
||||
// Enqueue ein Fehler (und Close idempotent).
|
||||
func TestClose(t *testing.T) {
|
||||
g := neueSpielerin()
|
||||
q := New(g.speak, nil)
|
||||
|
||||
q.Enqueue(Item{Text: "blocker"})
|
||||
g.warteBis(t, 1)
|
||||
q.Enqueue(Item{Text: "wartet"})
|
||||
|
||||
q.Close()
|
||||
|
||||
g.keineWeitere(t)
|
||||
if _, ok := q.Current(); ok {
|
||||
t.Error("Current nach Close: want idle")
|
||||
}
|
||||
if _, err := q.Enqueue(Item{Text: "zu spät"}); err == nil {
|
||||
t.Error("Enqueue nach Close: want Fehler")
|
||||
}
|
||||
q.Close() // idempotent
|
||||
}
|
||||
|
||||
// TestEnqueueVergibtIDs: IDs sind kurz und eindeutig.
|
||||
func TestEnqueueVergibtIDs(t *testing.T) {
|
||||
g := neueSpielerin()
|
||||
q := New(g.speak, nil)
|
||||
defer q.Close()
|
||||
|
||||
seen := map[string]bool{}
|
||||
for i := 0; i < 20; i++ {
|
||||
id, err := q.Enqueue(Item{Text: "x"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(id) < 4 || seen[id] {
|
||||
t.Errorf("ID %q zu kurz oder doppelt", id)
|
||||
}
|
||||
seen[id] = true
|
||||
}
|
||||
}
|
||||
|
||||
// waitIdle pollt, bis kein Item mehr läuft (Cancel-/Close-Pfade geben das
|
||||
// nach dem Worker-Exit synchron frei; das Pollen deckt die Übergangsphase
|
||||
// ab, ohne auf Timing zu trauen).
|
||||
func waitIdle(t *testing.T, q *Queue) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(time.Second)
|
||||
for {
|
||||
if _, ok := q.Current(); !ok {
|
||||
return
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatal("Queue wird nicht idle")
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
// TestMuteStore: Datei-Roundtrip, Fehlend-Heißt-Nicht-Stumm, atomares Set.
|
||||
func TestMuteStore(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, MuteFileName)
|
||||
|
||||
var s MuteStore
|
||||
if s.Muted() {
|
||||
t.Error("MuteStore ohne Pfad: Muted() = true, want false")
|
||||
}
|
||||
if err := s.Set(true); err == nil {
|
||||
t.Error("Set ohne Pfad: want Fehler")
|
||||
}
|
||||
|
||||
s = MuteStore{Path: path}
|
||||
if s.Muted() {
|
||||
t.Error("ohne Datei: Muted() = true, want false (nie still aus Versehen)")
|
||||
}
|
||||
|
||||
if err := s.Set(true); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !s.Muted() {
|
||||
t.Error("nach Set(true): Muted() = false")
|
||||
}
|
||||
if b, _ := os.ReadFile(path); string(b) != "true\n" {
|
||||
t.Errorf("Datei-Inhalt = %q, want %q", b, "true\n")
|
||||
}
|
||||
|
||||
if err := s.Set(false); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if s.Muted() {
|
||||
t.Error("nach Set(false): Muted() = true")
|
||||
}
|
||||
|
||||
// Müll in der Datei heißt: nicht stumm (nur „true" ist stumm).
|
||||
if err := os.WriteFile(path, []byte("vielleicht\n"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if s.Muted() {
|
||||
t.Error("Datei-Inhalt „vielleicht“: Muted() = true, want false")
|
||||
}
|
||||
}
|
||||
|
||||
// TestRateLimiter: Struktur-Semantik (Task 05 verdrahtet sie in den Loop).
|
||||
func TestRateLimiter(t *testing.T) {
|
||||
// Unbegrenzt (Task-04-Default): Interval 0 erlaubt immer.
|
||||
unlimited := &RateLimiter{}
|
||||
for i := 0; i < 3; i++ {
|
||||
if !unlimited.Allow() {
|
||||
t.Fatal("Interval 0: Allow = false, want true (unbegrenzt)")
|
||||
}
|
||||
unlimited.Reserve()
|
||||
}
|
||||
|
||||
// Begrenzt: nach Reserve ist erst nach Interval wieder Allow.
|
||||
l := &RateLimiter{Interval: 30 * time.Millisecond}
|
||||
if !l.Allow() {
|
||||
t.Fatal("frisches Limit: Allow = false")
|
||||
}
|
||||
l.Reserve()
|
||||
if l.Allow() {
|
||||
t.Fatal("sofort nach Reserve: Allow = true, want false")
|
||||
}
|
||||
if next := l.Next(); time.Now().After(next) {
|
||||
t.Errorf("Next() = %v liegt in der Vergangenheit", next)
|
||||
}
|
||||
time.Sleep(40 * time.Millisecond)
|
||||
if !l.Allow() {
|
||||
t.Fatal("nach Interval: Allow = false, want true")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
package queue
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// RateLimiter ist die Struktur für das Daemon-Gesetz gegen Ansage-Spam
|
||||
// (Spec: Default ca. 1 Ansage / 30 s, konfigurierbar).
|
||||
//
|
||||
// SEAM FÜR TASK 05: Struktur und Semantik stehen und sind getestet, sind
|
||||
// aber noch NICHT an den Worker-Loop angebunden — Task 04 erzeugt keinen
|
||||
// Limiter (unbegrenzt, so verlangt es der Task-04-Brief: „Default unendlich
|
||||
// in T04-Tests, dokumentiert"). Task 05 verdrahtet den Limiter (config
|
||||
// queue.rate_limit) in den Loop und ergänzt die prioritätsbewusste
|
||||
// Entscheidung: low verwerfen, normal parken, high konfigurierbar verstoßen
|
||||
// (queue.allow_high_burst) — dafür ist Next() bereits Teil der Struktur.
|
||||
type RateLimiter struct {
|
||||
// Interval ist der Mindestabstand zwischen zwei Ansage-Starts.
|
||||
// 0 oder negativ = unbegrenzt (Task-04-Default).
|
||||
Interval time.Duration
|
||||
|
||||
mu sync.Mutex
|
||||
last time.Time // Start der zuletzt per Reserve gebuchten Ansage
|
||||
}
|
||||
|
||||
// Allow meldet, ob sofort eine Ansage starten dürfte.
|
||||
func (l *RateLimiter) Allow() bool {
|
||||
if l == nil || l.Interval <= 0 {
|
||||
return true
|
||||
}
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
return time.Since(l.last) >= l.Interval
|
||||
}
|
||||
|
||||
// Reserve bucht einen Ansage-Slot (direkt nach Allow aufrufen).
|
||||
func (l *RateLimiter) Reserve() {
|
||||
if l == nil || l.Interval <= 0 {
|
||||
return
|
||||
}
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
l.last = time.Now()
|
||||
}
|
||||
|
||||
// Next liefert, wann der nächste Ansage-Slot frei wird (Nullzeit, wenn
|
||||
// Allow sofort true liefern würde) — Basis für das Parken wartender Items
|
||||
// in Task 05.
|
||||
func (l *RateLimiter) Next() time.Time {
|
||||
if l == nil || l.Interval <= 0 {
|
||||
return time.Time{}
|
||||
}
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
return l.last.Add(l.Interval)
|
||||
}
|
||||
Reference in new issue
Block a user