1package xturbo
2
3import (
4 "bytes"
5 "encoding/binary"
6 "errors"
7 "fmt"
8 "io"
9 "testing"
10 "time"
11
12 "bnema.dev/xturbo/internal/xtest"
13 "golang.org/x/sys/unix"
14)
15
16// dispatchConn returns a Conn over one end of a non-blocking socketpair with
17// default kernel buffers, and the other end acting as the server. The XFIXES
18// and Generic Event extensions are negotiated at the golden capture's bases.
19func dispatchConn(t testing.TB) (*Conn, int) {
20 t.Helper()
21 fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM|unix.SOCK_NONBLOCK|unix.SOCK_CLOEXEC, 0)
22 if err != nil {
23 t.Fatal(err)
24 }
25 c := newConnState(fds[0], nil, setupInfo{})
26 t.Cleanup(func() { _ = c.Close(); _ = unix.Close(fds[1]) })
27 c.ext.Set(ExtXFixes, 138, 87, 140)
28 c.ext.Set(ExtGenericEvent, 0x83, 0, 0)
29 c.buf.next = 1 << 40 // every widened seq in the tests is acknowledgeable
30 return c, fds[1]
31}
32
33// Packet builders. seq is the 16-bit wire value.
34func pktEvent(code uint8, seq uint16, fill byte) []byte {
35 p := bytes.Repeat([]byte{fill}, 32)
36 p[0] = code
37 binary.LittleEndian.PutUint16(p[2:], seq)
38 return p
39}
40
41func pktReply(seq uint16, data int) []byte {
42 words := (data + 3) / 4
43 p := make([]byte, 32+4*words)
44 p[0], p[1] = 1, 7
45 binary.LittleEndian.PutUint16(p[2:], seq)
46 binary.LittleEndian.PutUint32(p[4:], uint32(words))
47 for i := 8; i < len(p); i++ {
48 p[i] = byte(i * 7)
49 }
50 return p
51}
52
53func pktError(code uint8, seq uint16, major uint8) []byte {
54 p := make([]byte, 32)
55 p[1] = code
56 binary.LittleEndian.PutUint16(p[2:], seq)
57 binary.LittleEndian.PutUint32(p[4:], 0xdead)
58 p[10] = major
59 return p
60}
61
62func pktXGE(seq uint16, evtype uint16, tail int) []byte {
63 words := (tail + 3) / 4
64 p := make([]byte, 32+4*words)
65 p[0], p[1] = codeGenericEvent, 0x83
66 binary.LittleEndian.PutUint16(p[2:], seq)
67 binary.LittleEndian.PutUint32(p[4:], uint32(words))
68 binary.LittleEndian.PutUint16(p[8:], evtype)
69 for i := 32; i < len(p); i++ {
70 p[i] = byte(i)
71 }
72 return p
73}
74
75// record is one callback as seen by a test handler; data is a copy of the
76// packet bytes (Raw, plus Tail for events).
77type record struct {
78 class byte // 'e', 'r', 'x'
79 seq Seq
80 kind uint32
81 data []byte
82}
83
84func (r record) String() string {
85 return fmt.Sprintf("%c seq %d kind %#x %d bytes", r.class, r.seq, r.kind, len(r.data))
86}
87
88type recorder struct {
89 got []record
90 h Handler
91}
92
93// newRecorder records every callback of c. Replies reach OnReply only when
94// registered: on each MapRequest event (the first packet of scriptedStream)
95// it registers the stream's replies, at ev.Seq+1, +4 and +6.
96func newRecorder(c *Conn) *recorder {
97 r := &recorder{}
98 r.h = Handler{
99 OnEvent: func(ev *Event) {
100 expectScripted(c, ev)
101 r.got = append(r.got, record{'e', ev.Seq, uint32(ev.Kind), append(bytes.Clone(ev.Raw[:]), ev.Tail...)})
102 },
103 OnReply: func(rp *Reply) {
104 r.got = append(r.got, record{'r', rp.Seq, uint32(rp.Kind), bytes.Clone(rp.Raw)})
105 },
106 OnError: func(e *XError) {
107 r.got = append(r.got, record{'x', e.Seq, uint32(e.Code), bytes.Clone(e.Raw[:])})
108 },
109 }
110 return r
111}
112
113// feed writes data to the server end, dispatching whenever the socket fills,
114// then dispatches what remains.
115func feed(t testing.TB, c *Conn, srv int, data []byte, h *Handler) {
116 t.Helper()
117 for len(data) > 0 {
118 n, err := unix.Write(srv, data)
119 if err == unix.EAGAIN {
120 n = 0
121 } else if err != nil {
122 t.Fatal(err)
123 }
124 data = data[n:]
125 pump(t, c, h)
126 }
127 pump(t, c, h)
128}
129
130// pump dispatches until the socket holds nothing more: one Dispatch reads a
131// bounded amount.
132func pump(t testing.TB, c *Conn, h *Handler) {
133 t.Helper()
134 for {
135 if err := c.Dispatch(h); err != nil {
136 t.Fatal(err)
137 }
138 fds := []unix.PollFd{{Fd: int32(c.fd), Events: unix.POLLIN}}
139 if n, _ := unix.Poll(fds, 0); n == 0 {
140 return
141 }
142 }
143}
144
145func golden(t testing.TB, name string) []byte {
146 t.Helper()
147 raw, err := xtest.LoadHex("testdata/golden/" + name + ".hex")
148 if err != nil {
149 t.Fatal(err)
150 }
151 return raw
152}
153
154// scriptedStream is a mixed stream of goldens and synthetic packets with
155// increasing wire sequence numbers, and the callbacks it must produce.
156func scriptedStream(t testing.TB) ([]byte, []record) {
157 var stream []byte
158 var want []record
159 add := func(class byte, kind uint32, p []byte, seq uint16) {
160 binary.LittleEndian.PutUint16(p[2:], seq)
161 stream = append(stream, p...)
162 want = append(want, record{class, Seq(seq), kind, p})
163 }
164 add('e', 20, golden(t, "xproto/MapRequest.event"), 3)
165 add('r', uint32(kindUser), golden(t, "xproto/GetGeometry.reply"), 4)
166 add('x', 3, golden(t, "xproto/BadWindow.error"), 5)
167 add('e', uint32(ExtGenericEvent)<<8|0x81, golden(t, "xproto/GeGeneric.event"), 6)
168 add('e', uint32(ExtXFixes)<<8, golden(t, "xfixes/SelectionNotify.event"), 6)
169 add('r', uint32(kindUser), golden(t, "xproto/GetProperty.reply"), 7)
170 km := golden(t, "xproto/KeymapNotify.event") // no sequence: inherits 7
171 stream = append(stream, km...)
172 want = append(want, record{'e', 7, codeKeymapNotify, km})
173 add('e', uint32(KindUnknown), pktEvent(80, 0, 0x11), 8) // unbound code
174 add('e', uint32(ExtGenericEvent)<<8|0x80, pktXGE(0, 0, 0), 8)
175 add('r', uint32(kindUser), pktReply(0, 0), 9)
176 sent := pktEvent(0x80|33, 0, 0x22) // SendEvent ClientMessage
177 add('e', 33, sent, 9)
178 return stream, want
179}
180
181// expectScripted registers the replies of scriptedStream when its first
182// event arrives. Allocation-free, so the alloc tests use it too.
183func expectScripted(c *Conn, ev *Event) {
184 if ev.Kind != 20 {
185 return
186 }
187 for _, d := range [...]Seq{1, 4, 6} {
188 if err := c.Expect(ev.Seq+d, kindUser, uint32(d), 0, 0, 0); err != nil {
189 panic(err)
190 }
191 }
192}
193
194// expect registers seqs with kind kindUser.
195func expect(t testing.TB, c *Conn, seqs ...Seq) {
196 t.Helper()
197 for _, s := range seqs {
198 if err := c.Expect(s, kindUser, 0, 0, 0, 0); err != nil {
199 t.Fatal(err)
200 }
201 }
202}
203
204func checkRecords(t testing.TB, got, want []record) {
205 t.Helper()
206 if len(got) != len(want) {
207 t.Fatalf("%d callbacks, want %d: %v", len(got), len(want), got)
208 }
209 for i := range want {
210 g, w := got[i], want[i]
211 if g.class != w.class || g.seq != w.seq || g.kind != w.kind || !bytes.Equal(g.data, w.data) {
212 t.Fatalf("callback %d: %v, want %v", i, g, w)
213 }
214 }
215}
216
217func TestDispatchStream(t *testing.T) {
218 stream, want := scriptedStream(t)
219 t.Run("one read", func(t *testing.T) {
220 c, srv := dispatchConn(t)
221 r := newRecorder(c)
222 feed(t, c, srv, stream, &r.h)
223 checkRecords(t, r.got, want)
224 if c.buf.acked != 9 || c.seq != 9 {
225 t.Fatalf("acked %d seq %d", c.buf.acked, c.seq)
226 }
227 if st := c.Stats(); st.UnknownEvents != 1 || st.DroppedReplies != 0 || st.UnhandledErrors != 0 || st.MaxRingUse != 3 {
228 t.Fatalf("%+v", st)
229 }
230 })
231 t.Run("split at every byte", func(t *testing.T) {
232 for k := range len(stream) + 1 {
233 c, srv := dispatchConn(t)
234 r := newRecorder(c)
235 feed(t, c, srv, stream[:k], &r.h)
236 feed(t, c, srv, stream[k:], &r.h)
237 checkRecords(t, r.got, want)
238 }
239 })
240 t.Run("byte by byte", func(t *testing.T) {
241 c, srv := dispatchConn(t)
242 r := newRecorder(c)
243 for i := range stream {
244 feed(t, c, srv, stream[i:i+1], &r.h)
245 }
246 checkRecords(t, r.got, want)
247 })
248 t.Run("nil handler and nil fields", func(t *testing.T) {
249 c, srv := dispatchConn(t)
250 feed(t, c, srv, stream, nil)
251 feed(t, c, srv, stream, &Handler{})
252 if c.rpos != c.rend {
253 t.Fatal("stream not consumed")
254 }
255 })
256}
257
258func TestDispatchSeqWidening(t *testing.T) {
259 c, srv := dispatchConn(t)
260 r := newRecorder(c)
261 // Wire seqs climbing through two 16-bit wraps.
262 wires := []uint16{1, 0x7fff, 0xfffe, 0xffff, 0, 3, 0x8000, 0xffff, 2}
263 wantSeq := []Seq{1, 0x7fff, 0xfffe, 0xffff, 0x10000, 0x10003, 0x18000, 0x1ffff, 0x20002}
264 var stream []byte
265 for i, w := range wires {
266 switch i % 3 {
267 case 0:
268 stream = append(stream, pktReply(w, 4)...)
269 case 1:
270 stream = append(stream, pktEvent(22, w, 0)...)
271 default:
272 stream = append(stream, pktError(3, w, 12)...)
273 }
274 }
275 stream = append(stream, golden(t, "xproto/KeymapNotify.event")...)
276 expect(t, c, 1, 0xffff, 0x18000)
277 feed(t, c, srv, stream, &r.h)
278 if len(r.got) != len(wires)+1 {
279 t.Fatal(r.got)
280 }
281 for i, s := range wantSeq {
282 if r.got[i].seq != s {
283 t.Fatalf("packet %d: seq %#x, want %#x", i, r.got[i].seq, s)
284 }
285 }
286 if last := r.got[len(wires)]; last.seq != 0x20002 || c.buf.acked != 0x20002 {
287 t.Fatalf("KeymapNotify seq %#x, acked %#x", last.seq, c.buf.acked)
288 }
289
290 // ack never passes requests that were not sent.
291 c2, srv2 := dispatchConn(t)
292 c2.buf.next = 5
293 if err := c2.expect(9, kindUser, [4]uint32{}); err != nil { // internal: Expect rejects unsent seqs
294 t.Fatal(err)
295 }
296 feed(t, c2, srv2, pktReply(9, 0), nil)
297 if c2.buf.acked != 0 || c2.seq != 9 {
298 t.Fatalf("acked %d seq %d", c2.buf.acked, c2.seq)
299 }
300}
301
302func TestDispatchGrowth(t *testing.T) {
303 c, srv := dispatchConn(t)
304 r := newRecorder(c)
305 big := pktReply(1, 300<<10) // larger than the 64 KiB ring
306 xge := pktXGE(2, 5, 100<<10)
307 stream := append(append(append([]byte{}, pktEvent(22, 1, 0)...), big...), xge...)
308 expect(t, c, 1)
309 feed(t, c, srv, stream, &r.h)
310 if len(r.got) != 3 || !bytes.Equal(r.got[1].data, big) || !bytes.Equal(r.got[2].data, xge) {
311 t.Fatalf("%v", r.got)
312 }
313 if len(c.rb) != 512<<10 {
314 t.Fatalf("ring %d bytes, want 512 KiB", len(c.rb))
315 }
316}
317
318func TestDispatchMaxPacket(t *testing.T) {
319 t.Run("exactly MaxPacket", func(t *testing.T) {
320 c, srv := dispatchConn(t)
321 c.maxPacket = 256 << 10
322 r := newRecorder(c)
323 p := pktReply(1, c.maxPacket-32)
324 expect(t, c, 1)
325 feed(t, c, srv, p, &r.h)
326 if len(r.got) != 1 || len(r.got[0].data) != c.maxPacket || len(c.rb) != c.maxPacket {
327 t.Fatalf("%v ring %d", r.got, len(c.rb))
328 }
329 })
330 for _, tc := range []struct {
331 name string
332 p []byte
333 }{
334 {"reply", pktReply(1, 0)},
335 {"XGE", pktXGE(1, 0, 0)},
336 } {
337 t.Run(tc.name+" beyond MaxPacket", func(t *testing.T) {
338 c, srv := dispatchConn(t)
339 r := newRecorder(c)
340 p := tc.p
341 binary.LittleEndian.PutUint32(p[4:], (MaxPacket-32)/4+1)
342 stream := append(pktEvent(22, 1, 0), p...)
343 if _, err := unix.Write(srv, stream); err != nil {
344 t.Fatal(err)
345 }
346 if err := c.Dispatch(&r.h); !errors.Is(err, ErrPacketTooLarge) {
347 t.Fatal(err)
348 }
349 if len(r.got) != 1 || !errors.Is(c.Dispatch(&r.h), ErrPacketTooLarge) || !errors.Is(c.buf.Err(), ErrPacketTooLarge) {
350 t.Fatal("not sticky")
351 }
352 })
353 }
354 t.Run("length overflowing 32 bits", func(t *testing.T) {
355 if n, ok := packetSize(bytes.Repeat([]byte{1, 0, 0, 0, 0xff, 0xff, 0xff, 0xff}, 4), MaxPacket); ok {
356 t.Fatal(n)
357 }
358 })
359}
360
361func TestDispatchEOF(t *testing.T) {
362 for _, tail := range [][]byte{nil, pktReply(2, 64)[:40]} {
363 c, srv := dispatchConn(t)
364 r := newRecorder(c)
365 data := append(pktEvent(22, 1, 0), tail...)
366 if _, err := unix.Write(srv, data); err != nil {
367 t.Fatal(err)
368 }
369 unix.Shutdown(srv, unix.SHUT_WR)
370 if err := c.Dispatch(&r.h); err != io.EOF {
371 t.Fatal(err)
372 }
373 if len(r.got) != 1 || c.Dispatch(&r.h) != io.EOF || !errors.Is(c.buf.Err(), ErrClosed) {
374 t.Fatal("EOF not sticky")
375 }
376 }
377 c, _ := dispatchConn(t)
378 c.Close()
379 if !errors.Is(c.Dispatch(nil), ErrClosed) {
380 t.Fatal("Dispatch after Close")
381 }
382}
383
384func TestDispatchCloseInCallback(t *testing.T) {
385 c, srv := dispatchConn(t)
386 n := 0
387 h := Handler{OnEvent: func(*Event) { n++; c.Close() }}
388 if _, err := unix.Write(srv, append(pktEvent(22, 1, 0), pktEvent(22, 2, 0)...)); err != nil {
389 t.Fatal(err)
390 }
391 if err := c.Dispatch(&h); !errors.Is(err, ErrClosed) || n != 1 {
392 t.Fatal(err, n)
393 }
394}
395
396func TestDispatchReentrant(t *testing.T) {
397 c, srv := dispatchConn(t)
398 var nested error
399 var got []Seq
400 h := Handler{OnEvent: func(ev *Event) {
401 got = append(got, ev.Seq)
402 if len(got) == 1 {
403 nested = c.Dispatch(nil)
404 }
405 }}
406 var stream []byte
407 for i := range 5 {
408 stream = append(stream, pktEvent(22, uint16(i+1), 0)...)
409 }
410 if _, err := unix.Write(srv, stream); err != nil {
411 t.Fatal(err)
412 }
413 if err := c.Dispatch(&h); err != nil {
414 t.Fatal(err)
415 }
416 if !errors.Is(nested, errReentrantDispatch) || len(got) != 5 || got[4] != 5 {
417 t.Fatal(nested, got)
418 }
419 if err := c.Dispatch(&h); err != nil || c.dispatching { // usable afterwards
420 t.Fatal(err)
421 }
422}
423
424func TestDispatchBounded(t *testing.T) {
425 c, srv := dispatchConn(t)
426 stop := make(chan struct{})
427 done := make(chan struct{})
428 go func() { // keep the socket full until told to stop
429 defer close(done)
430 buf := bytes.Repeat(pktEvent(22, 1, 0), 2048)
431 for {
432 select {
433 case <-stop:
434 return
435 default:
436 }
437 if _, err := unix.Write(srv, buf); err != nil && err != unix.EAGAIN {
438 return
439 }
440 }
441 }()
442 defer func() { close(stop); <-done }()
443 events := 0
444 h := Handler{OnEvent: func(*Event) { events++ }}
445 fds := []unix.PollFd{{Fd: int32(c.fd), Events: unix.POLLIN}}
446 for n, _ := unix.Poll(fds, 1000); n == 0; n, _ = unix.Poll(fds, 1000) {
447 t.Fatal("writer never filled the socket")
448 }
449 res := make(chan error, 1)
450 go func() { res <- c.Dispatch(&h) }()
451 select {
452 case err := <-res:
453 if err != nil {
454 t.Fatal(err)
455 }
456 case <-time.After(5 * time.Second):
457 t.Fatal("Dispatch did not return under a flooding server")
458 }
459 // one budget of reads, plus at most the bytes already in the ring
460 if max := 2 * len(c.rb) / 32; events == 0 || events > max {
461 t.Fatalf("%d events, want 1..%d", events, max)
462 }
463}
464
465func TestDispatchReservedKindError(t *testing.T) {
466 c, srv := dispatchConn(t)
467 r := newRecorder(c)
468 if err := c.expect(1, kindDiscard, [4]uint32{}); err != nil {
469 t.Fatal(err)
470 }
471 feed(t, c, srv, pktError(3, 1, 0), &r.h)
472 if len(r.got) != 0 || c.Stats().UnhandledErrors != 0 {
473 t.Fatalf("reserved-kind error reached OnError: %v", r.got)
474 }
475 if c.live != 0 {
476 t.Fatal("slot not freed")
477 }
478}
479
480func TestDispatchClearsBorrowedViews(t *testing.T) {
481 c, srv := dispatchConn(t)
482 var h Handler
483 h.OnReply = func(*Reply) {}
484 expect(t, c, 2)
485 stream := append(append(pktReply(1, 64), pktReply(2, 64)...), pktXGE(3, 5, 64)...)
486 feed(t, c, srv, stream, &h) // reply 1 is dropped, 2 delivered, 3 an event with Tail
487 if c.reply.Raw != nil || c.ev.Tail != nil {
488 t.Fatalf("retained views: reply %d ev %d", len(c.reply.Raw), len(c.ev.Tail))
489 }
490 feed(t, c, srv, pktReply(4, 8), nil) // dropped reply
491 if c.reply.Raw != nil {
492 t.Fatal("dropped reply retained")
493 }
494}
495
496// recordedStream repeats the scripted stream (events, replies, errors) up to
497// about size bytes.
498func recordedStream(t testing.TB, size int) ([]byte, int) {
499 unit, want := scriptedStream(t)
500 var s []byte
501 n := 0
502 for len(s) < size {
503 s = append(s, unit...)
504 n += len(want)
505 }
506 return s, n
507}
508
509func TestAllocDispatch(t *testing.T) {
510 stream, packets := recordedStream(t, 1<<20)
511 c, srv := dispatchConn(t)
512 var events, replies, errs int
513 var sink byte
514 h := Handler{
515 OnEvent: func(ev *Event) { expectScripted(c, ev); events++; sink ^= ev.Raw[1]; sink ^= byte(len(ev.Tail)) },
516 OnReply: func(r *Reply) { replies++; sink ^= r.Raw[len(r.Raw)-1] },
517 OnError: func(e *XError) { errs++; sink ^= e.Code },
518 }
519 feed(t, c, srv, stream, &h) // warm the ring
520 if n := testing.AllocsPerRun(5, func() { feed(t, c, srv, stream, &h) }); n != 0 {
521 t.Fatalf("Dispatch allocates %v times per 1 MiB", n)
522 }
523 if events+replies+errs != 7*packets || errs == 0 || replies == 0 || c.Stats().DroppedReplies != 0 {
524 t.Fatal(events, replies, errs, c.Stats())
525 }
526 _ = sink
527}
528
529// mixedPackets is a 10k-packet stream: core events, replies, errors, XFIXES
530// and XGE events.
531func mixedPackets(t testing.TB) []byte {
532 var s []byte
533 for i := range 10000 {
534 seq := uint16(i)
535 switch i % 5 {
536 case 0:
537 s = append(s, pktEvent(22, seq, 0)...)
538 case 1:
539 s = append(s, pktReply(seq, 24)...)
540 case 2:
541 s = append(s, pktError(3, seq, 12)...)
542 case 3:
543 s = append(s, pktEvent(87, seq, 1)...)
544 default:
545 s = append(s, pktXGE(seq, 1, 8)...)
546 }
547 }
548 return s
549}
550
551func BenchmarkDispatch10k(b *testing.B) {
552 stream := mixedPackets(b)
553 c, srv := dispatchConn(b)
554 var sink int
555 h := Handler{
556 OnEvent: func(ev *Event) {
557 if ev.Kind == 22 {
558 _ = c.Expect(ev.Seq+1, kindUser, 0, 0, 0, 0)
559 }
560 sink += int(ev.Kind)
561 },
562 OnReply: func(r *Reply) { sink += len(r.Raw) },
563 OnError: func(e *XError) { sink += int(e.Code) },
564 }
565 feed(b, c, srv, stream, &h)
566 b.SetBytes(int64(len(stream)))
567 b.ReportAllocs()
568 for b.Loop() {
569 feed(b, c, srv, stream, &h)
570 }
571 if c.Stats().DroppedReplies != 0 {
572 b.Fatal(c.Stats())
573 }
574 _ = sink
575}
576
577// FuzzDispatchStream feeds an arbitrary byte stream in arbitrary chunks.
578// Invariants: no panic; the callbacks' bytes concatenate to the stream's
579// complete packets minus the (unregistered, so dropped) replies, which
580// Stats.DroppedReplies counts; Seq never decreases; Dispatch ends with io.EOF
581// (or ErrPacketTooLarge) after the server hangs up.
582func FuzzDispatchStream(f *testing.F) {
583 stream, _ := scriptedStream(f)
584 f.Add(stream, uint16(7))
585 f.Add(stream, uint16(0x2000|7))
586 f.Add(stream, uint16(0))
587 f.Add(append(pktReply(1, 1024), pktXGE(2, 1, 64)...), uint16(33))
588 f.Fuzz(func(t *testing.T, stream []byte, chunk uint16) {
589 if len(stream) > 1<<16 {
590 return
591 }
592 c, srv := dispatchConn(t)
593 c.maxPacket = 8 << 10
594 c.rb = make([]byte, packetHeadSize+int(chunk>>8)%96) // tiny ring: fuzz growth
595 var got []byte
596 var last Seq
597 mono := func(s Seq) {
598 if s < last {
599 t.Fatalf("seq %d after %d", s, last)
600 }
601 last = s
602 }
603 h := Handler{
604 OnEvent: func(ev *Event) { mono(ev.Seq); got = append(append(got, ev.Raw[:]...), ev.Tail...) },
605 OnReply: func(r *Reply) { mono(r.Seq); got = append(got, r.Raw...) },
606 OnError: func(e *XError) { mono(e.Seq); got = append(got, e.Raw[:]...) },
607 }
608 step := int(chunk)%97 + 1
609 var err error
610 for p := stream; len(p) > 0 && err == nil; {
611 k := min(step, len(p))
612 if _, werr := unix.Write(srv, p[:k]); werr != nil {
613 t.Fatal(werr)
614 }
615 p = p[k:]
616 err = c.Dispatch(&h)
617 }
618 unix.Shutdown(srv, unix.SHUT_WR)
619 for err == nil {
620 err = c.Dispatch(&h)
621 }
622 if err != io.EOF && !errors.Is(err, ErrPacketTooLarge) {
623 t.Fatal(err)
624 }
625 var want []byte
626 var replies uint64
627 for p := stream; len(p) >= 32; {
628 n, ok := packetSize(p, c.maxPacket)
629 if !ok || n > len(p) {
630 break
631 }
632 if p[0] == 1 {
633 replies++
634 } else {
635 want = append(want, p[:n]...)
636 }
637 p = p[n:]
638 }
639 if !bytes.Equal(got, want) || c.Stats().DroppedReplies != replies {
640 t.Fatalf("callbacks diverge from the stream: %d/%d bytes, %d dropped of %d replies", len(got), len(want), c.Stats().DroppedReplies, replies)
641 }
642 })
643}