66e40d384e
- Hub слушает *Bus в собственной горутине и вызывает зарегистрированные обработчики по типу события (On[T]); порядок сохраняется. - Мост к UI: колбэки выполняются в горутине Hub → внутри можно переложить работу на поток Fyne (fyne.Do) или thread-safe binding. - Close останавливает горутину и отписывается; Start идемпотентен. - Юнит-тесты: доставка по типу, игнор посторонних типов, несколько обработчиков одного типа, остановка после Close.
119 lines
3.0 KiB
Go
119 lines
3.0 KiB
Go
package events
|
|
|
|
import (
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/kamelion/ratatoskr-go/internal/model"
|
|
)
|
|
|
|
func TestHubDeliversTypedEvent(t *testing.T) {
|
|
bus := New(16)
|
|
h := NewHub(bus)
|
|
|
|
var got atomic.Value
|
|
On[TaskStatusChanged](h, func(e TaskStatusChanged) {
|
|
got.CompareAndSwap(nil, e)
|
|
})
|
|
|
|
h.Start()
|
|
defer h.Close()
|
|
|
|
want := TaskStatusChanged{ID: 7, From: model.StatusReady, To: model.StatusRunning}
|
|
bus.Publish(want)
|
|
|
|
deadline := time.Now().Add(2 * time.Second)
|
|
for time.Now().Before(deadline) {
|
|
if raw := got.Load(); raw != nil {
|
|
ev := raw.(TaskStatusChanged)
|
|
if ev.ID != 7 || ev.From != model.StatusReady || ev.To != model.StatusRunning {
|
|
t.Fatalf("unexpected event: %+v", ev)
|
|
}
|
|
return
|
|
}
|
|
time.Sleep(10 * time.Millisecond)
|
|
}
|
|
t.Fatal("handler was not called")
|
|
}
|
|
|
|
func TestHubIgnoresUnrelatedTypes(t *testing.T) {
|
|
bus := New(16)
|
|
h := NewHub(bus)
|
|
|
|
var calls atomic.Int32
|
|
On[TaskStatusChanged](h, func(TaskStatusChanged) { calls.Add(1) })
|
|
|
|
h.Start()
|
|
defer h.Close()
|
|
|
|
// другие типы событий не должны дойти до этого обработчика
|
|
bus.Publish(AgentActivity{TaskID: 1, Agent: "dev", Stage: "run"})
|
|
bus.Publish(LogLine{Level: "log", Text: "x"})
|
|
|
|
time.Sleep(200 * time.Millisecond)
|
|
if n := calls.Load(); n != 0 {
|
|
t.Fatalf("handler called %d times for unrelated events", n)
|
|
}
|
|
}
|
|
|
|
func TestHubMultipleHandlersSameType(t *testing.T) {
|
|
bus := New(16)
|
|
h := NewHub(bus)
|
|
|
|
var a, b atomic.Int32
|
|
On[HistoryAppended](h, func(HistoryAppended) { a.Add(1) })
|
|
On[HistoryAppended](h, func(HistoryAppended) { b.Add(1) })
|
|
|
|
h.Start()
|
|
defer h.Close()
|
|
|
|
bus.Publish(HistoryAppended{TaskID: 1, Role: "user", Content: "hi"})
|
|
|
|
deadline := time.Now().Add(2 * time.Second)
|
|
for time.Now().Before(deadline) {
|
|
if a.Load() == 1 && b.Load() == 1 {
|
|
return
|
|
}
|
|
time.Sleep(10 * time.Millisecond)
|
|
}
|
|
t.Fatal("not all handlers called")
|
|
}
|
|
|
|
func TestHubCloseStopsDelivery(t *testing.T) {
|
|
bus := New(16)
|
|
h := NewHub(bus)
|
|
|
|
var calls atomic.Int32
|
|
On[TaskCreated](h, func(TaskCreated) { calls.Add(1) })
|
|
|
|
h.Start()
|
|
bus.Publish(TaskCreated{ID: 1})
|
|
|
|
deadline := time.Now().Add(2 * time.Second)
|
|
for time.Now().Before(deadline) && calls.Load() == 0 {
|
|
time.Sleep(10 * time.Millisecond)
|
|
}
|
|
if calls.Load() == 0 {
|
|
t.Fatal("initial delivery failed")
|
|
}
|
|
|
|
h.Close()
|
|
// после Close Hub отписан — события не доходят
|
|
bus.Publish(TaskCreated{ID: 2})
|
|
time.Sleep(150 * time.Millisecond)
|
|
if n := calls.Load(); n > 1 {
|
|
t.Fatalf("handler called %d times after Close", n)
|
|
}
|
|
}
|
|
|
|
func TestHubStartIdempotent(t *testing.T) {
|
|
bus := New(16)
|
|
h := NewHub(bus)
|
|
h.Start()
|
|
h.Start() // повторный Start не должен создавать вторую горутину
|
|
h.Close()
|
|
// если бы было две горутины — Publish блокировался бы на буфере, но закрытие сняло бы блок;
|
|
// просто проверяем, что Close и повторный Start не падают
|
|
bus.Publish(TaskCreated{ID: 1})
|
|
} |