actor_test.go raw
1 package actor_test
2
3 import (
4 "context"
5 "testing"
6 "time"
7
8 "git.smesh.lol/actor"
9 )
10
11 // --- Counter: demonstrates Signal + Query ---
12
13 type counter struct {
14 inc actor.Signal
15 get actor.Query[int]
16 actor.Lifecycle
17 }
18
19 func newCounter() *counter {
20 c := &counter{
21 inc: actor.NewSignal(),
22 get: actor.NewQuery[int](),
23 Lifecycle: actor.NewLifecycle(),
24 }
25 actor.Go(c.Lifecycle, c.run)
26 return c
27 }
28
29 func (c *counter) run() {
30 var n int
31 for {
32 select {
33 case <-c.Stopping():
34 return
35 case msg := <-c.inc.Recv():
36 n++
37 msg.Done()
38 case msg := <-c.get.Recv():
39 msg.Reply(n)
40 }
41 }
42 }
43
44 func TestCounter(t *testing.T) {
45 c := newCounter()
46 defer c.Stop()
47
48 if got := c.get.Call(); got != 0 {
49 t.Fatalf("expected 0, got %d", got)
50 }
51 c.inc.Call()
52 c.inc.Call()
53 c.inc.Call()
54 if got := c.get.Call(); got != 3 {
55 t.Fatalf("expected 3, got %d", got)
56 }
57 }
58
59 // --- KV Store: demonstrates Func + Proc ---
60
61 type kvStore struct {
62 get actor.Func[string, string]
63 set actor.Proc[kvEntry]
64 actor.Lifecycle
65 }
66
67 type kvEntry struct {
68 Key, Value string
69 }
70
71 func newKVStore() *kvStore {
72 s := &kvStore{
73 get: actor.NewFunc[string, string](),
74 set: actor.NewProc[kvEntry](),
75 Lifecycle: actor.NewLifecycle(),
76 }
77 actor.Go(s.Lifecycle, s.run)
78 return s
79 }
80
81 func (s *kvStore) run() {
82 m := make(map[string]string)
83 for {
84 select {
85 case <-s.Stopping():
86 return
87 case msg := <-s.get.Recv():
88 msg.Reply(m[msg.Req])
89 case msg := <-s.set.Recv():
90 m[msg.Req.Key] = msg.Req.Value
91 msg.Done()
92 }
93 }
94 }
95
96 func TestKVStore(t *testing.T) {
97 s := newKVStore()
98 defer s.Stop()
99
100 s.set.Call(kvEntry{"foo", "bar"})
101 s.set.Call(kvEntry{"baz", "qux"})
102
103 if got := s.get.Call("foo"); got != "bar" {
104 t.Fatalf("expected bar, got %s", got)
105 }
106 if got := s.get.Call("baz"); got != "qux" {
107 t.Fatalf("expected qux, got %s", got)
108 }
109 if got := s.get.Call("missing"); got != "" {
110 t.Fatalf("expected empty, got %s", got)
111 }
112 }
113
114 // --- Inbox (fire-and-forget) ---
115
116 type logger struct {
117 add actor.Inbox[string]
118 count actor.Query[int]
119 actor.Lifecycle
120 }
121
122 func newLogger() *logger {
123 l := &logger{
124 add: actor.NewInbox[string](64),
125 count: actor.NewQuery[int](),
126 Lifecycle: actor.NewLifecycle(),
127 }
128 actor.Go(l.Lifecycle, l.run)
129 return l
130 }
131
132 func (l *logger) run() {
133 var n int
134 for {
135 select {
136 case <-l.Stopping():
137 return
138 case <-l.add.Recv():
139 n++
140 case msg := <-l.count.Recv():
141 msg.Reply(n)
142 }
143 }
144 }
145
146 func TestInbox(t *testing.T) {
147 l := newLogger()
148 defer l.Stop()
149
150 for i := 0; i < 10; i++ {
151 l.add.Send("msg")
152 }
153 // Drain: query forces the actor to process pending inbox messages
154 // that arrived before the query (channel ordering within one goroutine).
155 // We need a sync point; call count twice to ensure all inbox messages
156 // from the buffered channel have been processed.
157 _ = l.count.Call()
158 // The count may not be 10 yet because select is non-deterministic
159 // between the inbox and count channels. But after enough sync
160 // points it converges. For a deterministic test, use Proc instead.
161 got := l.count.Call()
162 if got < 1 || got > 10 {
163 t.Fatalf("expected 1-10, got %d", got)
164 }
165 }
166
167 func TestInboxTrySend(t *testing.T) {
168 inbox := actor.NewInbox[int](2)
169 if !inbox.TrySend(1) {
170 t.Fatal("first send should succeed")
171 }
172 if !inbox.TrySend(2) {
173 t.Fatal("second send should succeed")
174 }
175 if inbox.TrySend(3) {
176 t.Fatal("third send should fail (buffer full)")
177 }
178 }
179
180 // --- Lifecycle ---
181
182 func TestLifecycleStoppedChannel(t *testing.T) {
183 lc := actor.NewLifecycle()
184 actor.Go(lc, func() {
185 <-lc.Stopping()
186 })
187
188 select {
189 case <-lc.Stopped():
190 t.Fatal("should not be stopped yet")
191 default:
192 }
193
194 lc.Stop()
195
196 select {
197 case <-lc.Stopped():
198 // correct
199 default:
200 t.Fatal("should be stopped after Stop()")
201 }
202 }
203
204 // --- CallCtx ---
205
206 func TestFuncCallCtx(t *testing.T) {
207 s := newKVStore()
208 defer s.Stop()
209
210 s.set.Call(kvEntry{"a", "1"})
211
212 // Normal call with background context succeeds.
213 got, err := s.get.CallCtx(context.Background(), "a")
214 if err != nil {
215 t.Fatal(err)
216 }
217 if got != "1" {
218 t.Fatalf("expected 1, got %s", got)
219 }
220
221 // Already-cancelled context returns immediately.
222 ctx, cancel := context.WithCancel(context.Background())
223 cancel()
224 _, err = s.get.CallCtx(ctx, "a")
225 if err != context.Canceled {
226 t.Fatalf("expected context.Canceled, got %v", err)
227 }
228 }
229
230 func TestQueryCallCtx(t *testing.T) {
231 c := newCounter()
232 defer c.Stop()
233
234 c.inc.Call()
235 c.inc.Call()
236
237 got, err := c.get.CallCtx(context.Background())
238 if err != nil {
239 t.Fatal(err)
240 }
241 if got != 2 {
242 t.Fatalf("expected 2, got %d", got)
243 }
244
245 ctx, cancel := context.WithCancel(context.Background())
246 cancel()
247 _, err = c.get.CallCtx(ctx)
248 if err != context.Canceled {
249 t.Fatalf("expected context.Canceled, got %v", err)
250 }
251 }
252
253 func TestProcCallCtx(t *testing.T) {
254 s := newKVStore()
255 defer s.Stop()
256
257 err := s.set.CallCtx(context.Background(), kvEntry{"x", "y"})
258 if err != nil {
259 t.Fatal(err)
260 }
261 if got := s.get.Call("x"); got != "y" {
262 t.Fatalf("expected y, got %s", got)
263 }
264
265 ctx, cancel := context.WithCancel(context.Background())
266 cancel()
267 err = s.set.CallCtx(ctx, kvEntry{"z", "w"})
268 if err != context.Canceled {
269 t.Fatalf("expected context.Canceled, got %v", err)
270 }
271 }
272
273 func TestSignalCallCtx(t *testing.T) {
274 c := newCounter()
275 defer c.Stop()
276
277 err := c.inc.CallCtx(context.Background())
278 if err != nil {
279 t.Fatal(err)
280 }
281 if got := c.get.Call(); got != 1 {
282 t.Fatalf("expected 1, got %d", got)
283 }
284
285 ctx, cancel := context.WithCancel(context.Background())
286 cancel()
287 err = c.inc.CallCtx(ctx)
288 if err != context.Canceled {
289 t.Fatalf("expected context.Canceled, got %v", err)
290 }
291 }
292
293 func TestCallCtxDeadlineExceeded(t *testing.T) {
294 // Create a func with no actor listening - CallCtx should bail on timeout.
295 f := actor.NewFunc[string, string]()
296 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
297 defer cancel()
298
299 _, err := f.CallCtx(ctx, "hello")
300 if err != context.DeadlineExceeded {
301 t.Fatalf("expected DeadlineExceeded, got %v", err)
302 }
303 }
304
305 // --- Concurrent callers ---
306
307 func TestConcurrentCallers(t *testing.T) {
308 c := newCounter()
309 defer c.Stop()
310
311 done := make(chan struct{})
312 for i := 0; i < 100; i++ {
313 go func() {
314 c.inc.Call()
315 done <- struct{}{}
316 }()
317 }
318 for i := 0; i < 100; i++ {
319 <-done
320 }
321 if got := c.get.Call(); got != 100 {
322 t.Fatalf("expected 100, got %d", got)
323 }
324 }
325