feat: auto-start embedded NATS server in New()
Pub/sub now works out of the box — New() starts a process-scoped
embedded NATS server with JetStream. The PubSub interface remains
for custom backends via Config(Options{PubSub: ...}).
- Move vianats functionality into nats.go (eliminates circular import)
- Add NATSConn(), JetStream(), EnsureStream(), ReplayHistory[T]() to via
- Delete vianats/ package
- Simplify nats-chatroom and pubsub-crud examples
- Rewrite pubsub tests to use real embedded NATS
This commit is contained in:
108
nats_test.go
108
nats_test.go
@@ -2,7 +2,6 @@ package via
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -11,88 +10,36 @@ import (
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
type mockHandler struct {
|
||||
id int64
|
||||
fn func([]byte)
|
||||
active atomic.Bool
|
||||
}
|
||||
|
||||
// mockPubSub implements PubSub for testing without NATS.
|
||||
type mockPubSub struct {
|
||||
mu sync.Mutex
|
||||
subs map[string][]*mockHandler
|
||||
nextID atomic.Int64
|
||||
}
|
||||
|
||||
func newMockPubSub() *mockPubSub {
|
||||
return &mockPubSub{subs: make(map[string][]*mockHandler)}
|
||||
}
|
||||
|
||||
func (m *mockPubSub) Publish(subject string, data []byte) error {
|
||||
m.mu.Lock()
|
||||
handlers := make([]*mockHandler, len(m.subs[subject]))
|
||||
copy(handlers, m.subs[subject])
|
||||
m.mu.Unlock()
|
||||
for _, h := range handlers {
|
||||
if h.active.Load() {
|
||||
h.fn(data)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *mockPubSub) Subscribe(subject string, handler func(data []byte)) (Subscription, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
mh := &mockHandler{
|
||||
id: m.nextID.Add(1),
|
||||
fn: handler,
|
||||
}
|
||||
mh.active.Store(true)
|
||||
m.subs[subject] = append(m.subs[subject], mh)
|
||||
return &mockSub{handler: mh}, nil
|
||||
}
|
||||
|
||||
func (m *mockPubSub) Close() error { return nil }
|
||||
|
||||
type mockSub struct {
|
||||
handler *mockHandler
|
||||
}
|
||||
|
||||
func (s *mockSub) Unsubscribe() error {
|
||||
s.handler.active.Store(false)
|
||||
return nil
|
||||
}
|
||||
|
||||
func TestPubSub_RoundTrip(t *testing.T) {
|
||||
ps := newMockPubSub()
|
||||
v := New()
|
||||
v.Config(Options{PubSub: ps})
|
||||
defer v.Shutdown()
|
||||
|
||||
var received []byte
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(1)
|
||||
done := make(chan struct{})
|
||||
|
||||
c := newContext("test-ctx", "/", v)
|
||||
c.View(func() h.H { return h.Div() })
|
||||
|
||||
_, err := c.Subscribe("test.topic", func(data []byte) {
|
||||
received = data
|
||||
wg.Done()
|
||||
close(done)
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
err = c.Publish("test.topic", []byte("hello"))
|
||||
require.NoError(t, err)
|
||||
|
||||
wg.Wait()
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("timed out waiting for message")
|
||||
}
|
||||
assert.Equal(t, []byte("hello"), received)
|
||||
}
|
||||
|
||||
func TestPubSub_MultipleSubscribers(t *testing.T) {
|
||||
ps := newMockPubSub()
|
||||
v := New()
|
||||
v.Config(Options{PubSub: ps})
|
||||
defer v.Shutdown()
|
||||
|
||||
var mu sync.Mutex
|
||||
var results []string
|
||||
@@ -119,7 +66,17 @@ func TestPubSub_MultipleSubscribers(t *testing.T) {
|
||||
})
|
||||
|
||||
c1.Publish("broadcast", []byte("msg"))
|
||||
wg.Wait()
|
||||
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
wg.Wait()
|
||||
close(done)
|
||||
}()
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("timed out waiting for messages")
|
||||
}
|
||||
|
||||
assert.Len(t, results, 2)
|
||||
assert.Contains(t, results, "c1:msg")
|
||||
@@ -127,9 +84,8 @@ func TestPubSub_MultipleSubscribers(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestPubSub_SubscriptionCleanupOnDispose(t *testing.T) {
|
||||
ps := newMockPubSub()
|
||||
v := New()
|
||||
v.Config(Options{PubSub: ps})
|
||||
defer v.Shutdown()
|
||||
|
||||
c := newContext("cleanup-ctx", "/", v)
|
||||
c.View(func() h.H { return h.Div() })
|
||||
@@ -144,9 +100,8 @@ func TestPubSub_SubscriptionCleanupOnDispose(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestPubSub_ManualUnsubscribe(t *testing.T) {
|
||||
ps := newMockPubSub()
|
||||
v := New()
|
||||
v.Config(Options{PubSub: ps})
|
||||
defer v.Shutdown()
|
||||
|
||||
c := newContext("unsub-ctx", "/", v)
|
||||
c.View(func() h.H { return h.Div() })
|
||||
@@ -160,28 +115,13 @@ func TestPubSub_ManualUnsubscribe(t *testing.T) {
|
||||
sub.Unsubscribe()
|
||||
|
||||
c.Publish("topic", []byte("ignored"))
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
assert.False(t, called)
|
||||
}
|
||||
|
||||
func TestPubSub_NoOpWhenNotConfigured(t *testing.T) {
|
||||
v := New()
|
||||
|
||||
c := newContext("noop-ctx", "/", v)
|
||||
c.View(func() h.H { return h.Div() })
|
||||
|
||||
err := c.Publish("topic", []byte("data"))
|
||||
assert.Error(t, err)
|
||||
|
||||
sub, err := c.Subscribe("topic", func(data []byte) {})
|
||||
assert.Error(t, err)
|
||||
assert.Nil(t, sub)
|
||||
}
|
||||
|
||||
func TestPubSub_NoOpDuringPanicCheck(t *testing.T) {
|
||||
ps := newMockPubSub()
|
||||
v := New()
|
||||
v.Config(Options{PubSub: ps})
|
||||
defer v.Shutdown()
|
||||
|
||||
// Panic-check context has id=""
|
||||
c := newContext("", "/", v)
|
||||
|
||||
Reference in New Issue
Block a user