Files
fun/stream_onclose_test.go
chiyiandDevin 84c11c6fde v1.4.0: generic Stream[T] — typed stream messages, drop (T, *Stream, error)
- Stream[T]: Send(v T) type-safe; message type recovered from the
  instantiated Send signature (reflect has no type-args API)
- method signatures reduced to (error), (T, error), (*Stream[T], error);
  the (T, *Stream, error) first-message form is gone — send it via Send
- return T and stream message T now run checkType at bind time
  (Stream[any] remains the explicit untyped escape hatch)
- generated clients: TS onMessage: (data: T); Go <-chan T
- fix vacuous panic assertions (fmt.Sprint(nil) -> "<nil>") in
  hardening/route test catch helpers

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-09-19 15:39:41 +08:00

47 lines
1.2 KiB
Go

package fun
import (
"sync/atomic"
"testing"
"time"
)
// OnClose 必须允许业务方法在 return *Stream 之前同步注册;旧实现等待 Inject,
// 而 Inject 只有 invoke 在业务方法返回后才调用,形成精确死锁:
//
// business OnClose -> wait ready -> invoke wait business return -> never Inject.
func TestStreamOnCloseBeforeInjectDoesNotBlock(t *testing.T) {
st := &Stream[any]{}
registered := make(chan struct{})
var called atomic.Int32
go func() {
st.OnClose(func() { called.Add(1) })
close(registered)
}()
select {
case <-registered:
// expected: registration is independent from channel injection
case <-time.After(time.Second):
t.Fatal("OnClose blocked before Inject (business method would deadlock)")
}
ch := make(chan any)
done := make(chan struct{})
st.Inject(ch, done)
st.Close()
if got := called.Load(); got != 1 {
t.Fatalf("callback called %d times, want 1", got)
}
}
func TestStreamOnCloseAfterAlreadyClosedRunsImmediately(t *testing.T) {
st := &Stream[any]{}
st.Inject(make(chan any), make(chan struct{}))
st.Close()
var called atomic.Int32
st.OnClose(func() { called.Add(1) })
if got := called.Load(); got != 1 {
t.Fatalf("callback called %d times, want 1", got)
}
}