detached 12c08ea

feat(conn): add directed signals, async calls and poll helpers

commit
12c08ea8ecf1a8f37968309294ead70e6c47133d
author
brice · · 2026-10-14 18:41 UTC
parent
93dc6b1
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
3 files · +403 -0
Unified
UnifiedSplit
M conn.go +43 -0
@@ -10,6 +10,7 @@ import (
1010 "os"
1111 "strconv"
1212 "strings"
13+ "syscall"
1314 )
1415
1516 const (
@@ -264,6 +265,23 @@ func (c *Conn) NewSignal(path, iface, member, sig string) *Encoder {
264265 return &c.enc
265266 }
266267
268+// NewSignalTo starts a signal addressed to the single peer dest, a unique
269+// name such as ":1.42" or a well-known name, and returns the encoder for its
270+// body. Only dest receives it, whatever match rules other peers have.
271+func (c *Conn) NewSignalTo(dest, path, iface, member, sig string) *Encoder {
272+ c.enc.begin(TypeSignal, c.nextSerial(), 0, dest, path, iface, member, "", sig)
273+ return &c.enc
274+}
275+
276+// NewMethodCall starts a method call like NewCall and also returns its
277+// serial. A loop that only uses Send and ReadMessage keeps the serial to
278+// recognize the reply later, as Message.ReplySerial, without blocking in Call.
279+// The serial is the one Send returns for this message.
280+func (c *Conn) NewMethodCall(dest, path, iface, member, sig string) (*Encoder, uint32) {
281+ e := c.NewCall(dest, path, iface, member, sig)
282+ return e, c.serial
283+}
284+
267285 // NewReply starts the reply to the method call m and returns the encoder for
268286 // its body.
269287 func (c *Conn) NewReply(m *Message, sig string) *Encoder {
@@ -348,6 +366,31 @@ func replyError(m *Message) *Error {
348366 return e
349367 }
350368
369+// SyscallConn returns the raw socket of the connection, for a poll loop that
370+// waits for it to become readable together with other file descriptors.
371+// Only wait on the descriptor: read, write and close through the Conn.
372+//
373+// Bytes already read from the socket are not visible to poll: check Buffered
374+// before waiting. After the descriptor turns readable, ReadMessage may still
375+// block until the rest of a partially received message arrives.
376+func (c *Conn) SyscallConn() (syscall.RawConn, error) { return c.c.SyscallConn() }
377+
378+// Buffered reports whether ReadMessage can return without waiting for the
379+// socket: a complete message, or a sticky error, is already in the receive
380+// buffer. A poll loop must call ReadMessage until Buffered is false before it
381+// waits on the descriptor again, because ReadMessage reads everything the
382+// socket has and a later poll would not see those bytes.
383+func (c *Conn) Buffered() bool {
384+ if c.err != nil {
385+ return true
386+ }
387+ if c.end-c.start < headerLen {
388+ return false
389+ }
390+ n, err := frameLen(c.rbuf[c.start:c.end])
391+ return err != nil || c.end-c.start >= n
392+}
393+
351394 // ReadMessage blocks until the next message arrives. The message and its
352395 // strings are valid until the next call to ReadMessage or Call.
353396 func (c *Conn) ReadMessage() (*Message, error) {
@@ -10,6 +10,7 @@ import (
10 "os"10 "os"
11 "strconv"11 "strconv"
12 "strings"12 "strings"
13+ "syscall"
13 )14 )
14 15
15 const (16 const (
@@ -264,6 +265,23 @@ func (c *Conn) NewSignal(path, iface, member, sig string) *Encoder {
264 return &c.enc265 return &c.enc
265 }266 }
266 267
268+// NewSignalTo starts a signal addressed to the single peer dest, a unique
269+// name such as ":1.42" or a well-known name, and returns the encoder for its
270+// body. Only dest receives it, whatever match rules other peers have.
271+func (c *Conn) NewSignalTo(dest, path, iface, member, sig string) *Encoder {
272+ c.enc.begin(TypeSignal, c.nextSerial(), 0, dest, path, iface, member, "", sig)
273+ return &c.enc
274+}
275+
276+// NewMethodCall starts a method call like NewCall and also returns its
277+// serial. A loop that only uses Send and ReadMessage keeps the serial to
278+// recognize the reply later, as Message.ReplySerial, without blocking in Call.
279+// The serial is the one Send returns for this message.
280+func (c *Conn) NewMethodCall(dest, path, iface, member, sig string) (*Encoder, uint32) {
281+ e := c.NewCall(dest, path, iface, member, sig)
282+ return e, c.serial
283+}
284+
267 // NewReply starts the reply to the method call m and returns the encoder for285 // NewReply starts the reply to the method call m and returns the encoder for
268 // its body.286 // its body.
269 func (c *Conn) NewReply(m *Message, sig string) *Encoder {287 func (c *Conn) NewReply(m *Message, sig string) *Encoder {
@@ -348,6 +366,31 @@ func replyError(m *Message) *Error {
348 return e366 return e
349 }367 }
350 368
369+// SyscallConn returns the raw socket of the connection, for a poll loop that
370+// waits for it to become readable together with other file descriptors.
371+// Only wait on the descriptor: read, write and close through the Conn.
372+//
373+// Bytes already read from the socket are not visible to poll: check Buffered
374+// before waiting. After the descriptor turns readable, ReadMessage may still
375+// block until the rest of a partially received message arrives.
376+func (c *Conn) SyscallConn() (syscall.RawConn, error) { return c.c.SyscallConn() }
377+
378+// Buffered reports whether ReadMessage can return without waiting for the
379+// socket: a complete message, or a sticky error, is already in the receive
380+// buffer. A poll loop must call ReadMessage until Buffered is false before it
381+// waits on the descriptor again, because ReadMessage reads everything the
382+// socket has and a later poll would not see those bytes.
383+func (c *Conn) Buffered() bool {
384+ if c.err != nil {
385+ return true
386+ }
387+ if c.end-c.start < headerLen {
388+ return false
389+ }
390+ n, err := frameLen(c.rbuf[c.start:c.end])
391+ return err != nil || c.end-c.start >= n
392+}
393+
351 // ReadMessage blocks until the next message arrives. The message and its394 // ReadMessage blocks until the next message arrives. The message and its
352 // strings are valid until the next call to ReadMessage or Call.395 // strings are valid until the next call to ReadMessage or Call.
353 func (c *Conn) ReadMessage() (*Message, error) {396 func (c *Conn) ReadMessage() (*Message, error) {
M doc.go +21 -0
@@ -2,4 +2,25 @@
22 // state. It speaks the D-Bus wire protocol over a Unix socket and decodes
33 // messages in place, so strings and arrays point into the receive buffer and
44 // stay valid only until the next read.
5+//
6+// # Clients and servers
7+//
8+// A Conn has one encoder and one receive buffer. A client may block in Call,
9+// which sends a method call and waits for its reply, and may set Handle to see
10+// the messages that arrive meanwhile.
11+//
12+// A server loop uses only Send and ReadMessage, never Call or Handle. It
13+// answers a method call with NewReply or NewError, emits NewSignal or the
14+// directed NewSignalTo, and sends its own requests with NewMethodCall, which
15+// returns the serial. The reply comes back later through ReadMessage as a
16+// message whose ReplySerial is that serial. Decode a message completely before
17+// the next Send is built and the next read: both reuse the same buffers.
18+//
19+// # Polling
20+//
21+// A loop that also waits for other descriptors gets the socket from
22+// SyscallConn and waits for it to be readable. ReadMessage reads all the bytes
23+// available, so call it while Buffered reports true before waiting again.
24+// After the descriptor turns readable, ReadMessage may still block for the
25+// rest of a message that has only partly arrived.
526 package zerobus
@@ -2,4 +2,25 @@
2 // state. It speaks the D-Bus wire protocol over a Unix socket and decodes2 // state. It speaks the D-Bus wire protocol over a Unix socket and decodes
3 // messages in place, so strings and arrays point into the receive buffer and3 // messages in place, so strings and arrays point into the receive buffer and
4 // stay valid only until the next read.4 // stay valid only until the next read.
5+//
6+// # Clients and servers
7+//
8+// A Conn has one encoder and one receive buffer. A client may block in Call,
9+// which sends a method call and waits for its reply, and may set Handle to see
10+// the messages that arrive meanwhile.
11+//
12+// A server loop uses only Send and ReadMessage, never Call or Handle. It
13+// answers a method call with NewReply or NewError, emits NewSignal or the
14+// directed NewSignalTo, and sends its own requests with NewMethodCall, which
15+// returns the serial. The reply comes back later through ReadMessage as a
16+// message whose ReplySerial is that serial. Decode a message completely before
17+// the next Send is built and the next read: both reuse the same buffers.
18+//
19+// # Polling
20+//
21+// A loop that also waits for other descriptors gets the socket from
22+// SyscallConn and waits for it to be readable. ReadMessage reads all the bytes
23+// available, so call it while Buffered reports true before waiting again.
24+// After the descriptor turns readable, ReadMessage may still block for the
25+// rest of a message that has only partly arrived.
5 package zerobus26 package zerobus
A server_test.go +339 -0
@@ -0,0 +1,339 @@
1+package zerobus
2+
3+import (
4+ "net"
5+ "path/filepath"
6+ "syscall"
7+ "testing"
8+ "time"
9+ "unsafe"
10+)
11+
12+// socketPair returns a Conn whose socket is the client end of a Unix socket,
13+// without the bus handshake, and the other end to write raw frames to it.
14+func socketPair(t *testing.T) (*Conn, net.Conn) {
15+ t.Helper()
16+ l, err := net.Listen("unix", filepath.Join(t.TempDir(), "s"))
17+ if err != nil {
18+ t.Fatal(err)
19+ }
20+ defer func() { _ = l.Close() }()
21+ uc, err := net.DialUnix("unix", nil, &net.UnixAddr{Name: l.Addr().String(), Net: "unix"})
22+ if err != nil {
23+ t.Fatal(err)
24+ }
25+ peer, err := l.Accept()
26+ if err != nil {
27+ t.Fatal(err)
28+ }
29+ t.Cleanup(func() { _ = uc.Close(); _ = peer.Close() })
30+ return &Conn{c: uc, rbuf: make([]byte, defaultBufSz)}, peer
31+}
32+
33+func TestNewSignalToSetsDestination(t *testing.T) {
34+ var c Conn
35+ c.NewSignalTo(":1.42", "/a", "a.b", "Hit", "s").Str("x")
36+ b, err := c.enc.finish()
37+ if err != nil {
38+ t.Fatal(err)
39+ }
40+ var m Message
41+ if err := parse(b, &m); err != nil {
42+ t.Fatal(err)
43+ }
44+ if m.Type != TypeSignal || m.Destination != ":1.42" || m.Path != "/a" || m.Interface != "a.b" || m.Member != "Hit" {
45+ t.Fatalf("message %+v", m)
46+ }
47+ if got := m.Body().Str(); got != "x" {
48+ t.Fatalf("body %q", got)
49+ }
50+
51+ c.NewSignal("/a", "a.b", "Hit", "")
52+ b, err = c.enc.finish()
53+ if err != nil {
54+ t.Fatal(err)
55+ }
56+ if err := parse(b, &m); err != nil || m.Destination != "" {
57+ t.Fatalf("broadcast signal has destination %q (%v)", m.Destination, err)
58+ }
59+
60+ c.NewSignalTo("not a name", "/a", "a.b", "Hit", "")
61+ if _, err := c.Send(); err == nil {
62+ t.Fatal("invalid destination accepted")
63+ }
64+}
65+
66+func TestDirectedSignalReachesOnlyDestination(t *testing.T) {
67+ addr := privateBus(t)
68+ server := dial(t, addr)
69+ target := dial(t, addr)
70+ other := dial(t, addr)
71+ const rule = "type='signal',interface='org.example.Directed'"
72+ for _, c := range []*Conn{target, other} {
73+ if err := c.AddMatch(rule); err != nil {
74+ t.Fatal(err)
75+ }
76+ }
77+
78+ server.NewSignalTo(target.UniqueName(), "/a", "org.example.Directed", "Hit", "u").Uint32(7)
79+ if _, err := server.Send(); err != nil {
80+ t.Fatal(err)
81+ }
82+
83+ // The bus routes in order: a signal for other would come before the
84+ // reply to its round trip.
85+ var leaked bool
86+ other.Handle = func(m *Message) { leaked = leaked || m.Member == "Hit" }
87+ server.NewCall(busName, busPath, busName, "GetId", "")
88+ if _, err := server.Call(); err != nil {
89+ t.Fatal(err)
90+ }
91+ other.NewCall(busName, busPath, busName, "GetId", "")
92+ if _, err := other.Call(); err != nil {
93+ t.Fatal(err)
94+ }
95+ if leaked {
96+ t.Fatal("directed signal reached another peer")
97+ }
98+
99+ for {
100+ m, err := target.ReadMessage()
101+ if err != nil {
102+ t.Fatal(err)
103+ }
104+ if m.Member != "Hit" {
105+ continue
106+ }
107+ if m.Type != TypeSignal || m.Sender != server.UniqueName() || m.Destination != target.UniqueName() || m.Body().Uint32() != 7 {
108+ t.Fatalf("message %+v", m)
109+ }
110+ return
111+ }
112+}
113+
114+func TestAsyncMethodCallMatching(t *testing.T) {
115+ addr := privateBus(t)
116+ c := dial(t, addr)
117+ other := dial(t, addr)
118+
119+ // Two requests are in flight at once; the loop only uses Send and
120+ // ReadMessage and tells the replies apart by their serial.
121+ want := map[uint32]string{}
122+ for _, name := range []string{c.UniqueName(), other.UniqueName(), "org.example.Nobody"} {
123+ e, serial := c.NewMethodCall(busName, busPath, busName, "GetNameOwner", "s")
124+ e.Str(name)
125+ sent, err := c.Send()
126+ if err != nil {
127+ t.Fatal(err)
128+ }
129+ if sent != serial {
130+ t.Fatalf("NewMethodCall serial %d, Send serial %d", serial, sent)
131+ }
132+ if _, dup := want[serial]; dup {
133+ t.Fatalf("serial %d reused", serial)
134+ }
135+ want[serial] = name
136+ }
137+
138+ for len(want) > 0 {
139+ m, err := c.ReadMessage()
140+ if err != nil {
141+ t.Fatal(err)
142+ }
143+ if m.Type != TypeMethodReturn && m.Type != TypeError {
144+ continue // NameAcquired
145+ }
146+ name, ok := want[m.ReplySerial]
147+ if !ok {
148+ t.Fatalf("unexpected reply to serial %d", m.ReplySerial)
149+ }
150+ delete(want, m.ReplySerial)
151+ switch name {
152+ case "org.example.Nobody":
153+ if m.Type != TypeError {
154+ t.Fatalf("reply for %q is %d, want an error", name, m.Type)
155+ }
156+ default:
157+ if m.Type != TypeMethodReturn || m.Body().Str() != name {
158+ t.Fatalf("reply for %q is %d", name, m.Type)
159+ }
160+ }
161+ }
162+}
163+
164+func TestBufferedAndPollLoop(t *testing.T) {
165+ c, peer := socketPair(t)
166+ if c.Buffered() {
167+ t.Fatal("empty connection is buffered")
168+ }
169+ two := append(signal("One"), signal("Two")...)
170+ three := signal("Three")
171+ // Two whole frames and the start of a third arrive in one write.
172+ if _, err := peer.Write(append(two, three[:10]...)); err != nil {
173+ t.Fatal(err)
174+ }
175+ waitReadable(t, c)
176+ for _, want := range []string{"One", "Two"} {
177+ m, err := c.ReadMessage()
178+ if err != nil || m.Member != want {
179+ t.Fatalf("got %v, %v; want %q", m, err, want)
180+ }
181+ if got, wantBuf := c.Buffered(), want == "One"; got != wantBuf {
182+ t.Fatalf("Buffered after %q = %v, want %v", want, got, wantBuf)
183+ }
184+ }
185+ // The head of a third frame is not a message.
186+ if c.Buffered() {
187+ t.Fatal("a partial frame counts as buffered")
188+ }
189+ if _, err := peer.Write(three[10:]); err != nil {
190+ t.Fatal(err)
191+ }
192+ waitReadable(t, c)
193+ m, err := c.ReadMessage()
194+ if err != nil || m.Member != "Three" {
195+ t.Fatalf("got %v, %v", m, err)
196+ }
197+ if c.Buffered() {
198+ t.Fatal("drained connection is buffered")
199+ }
200+}
201+
202+func TestBufferedReportsStickyError(t *testing.T) {
203+ c, peer := socketPair(t)
204+ bad := signal("Bad")
205+ bad[3] = 9 // protocol version
206+ if _, err := peer.Write(append(bad, signal("Next")...)); err != nil {
207+ t.Fatal(err)
208+ }
209+ if _, err := c.ReadMessage(); err == nil {
210+ t.Fatal("malformed frame accepted")
211+ }
212+ if !c.Buffered() {
213+ t.Fatal("a sticky error must not make a poll loop wait")
214+ }
215+}
216+
217+// waitReadable blocks until the socket of c is readable, as a poll loop does.
218+func waitReadable(t *testing.T, c *Conn) {
219+ t.Helper()
220+ rc, err := c.SyscallConn()
221+ if err != nil {
222+ t.Fatal(err)
223+ }
224+ var fd int
225+ if err := rc.Control(func(f uintptr) { fd = int(f) }); err != nil {
226+ t.Fatal(err)
227+ }
228+ var set syscall.FdSet
229+ bits := int(unsafe.Sizeof(set.Bits[0])) * 8
230+ set.Bits[fd/bits] |= 1 << (uint(fd) % uint(bits))
231+ tv := syscall.Timeval{Sec: 5}
232+ n, err := syscall.Select(fd+1, &set, nil, nil, &tv)
233+ if err != nil || n != 1 {
234+ t.Fatalf("select = %d, %v", n, err)
235+ }
236+}
237+
238+func TestSignalAllocations(t *testing.T) {
239+ c := dial(t, privateBus(t))
240+ send := func() {
241+ e := c.NewSignalTo(":1.1", "/a", "org.example.Alloc", "Hit", "us")
242+ e.Uint32(1)
243+ e.Str("x")
244+ if _, err := c.Send(); err != nil {
245+ t.Fatal(err)
246+ }
247+ }
248+ send() // grow the buffers once
249+ if allocs := testing.AllocsPerRun(200, send); allocs != 0 {
250+ t.Fatalf("%v allocations per directed signal, want 0", allocs)
251+ }
252+}
253+
254+func TestMethodCallAllocations(t *testing.T) {
255+ c := dial(t, privateBus(t))
256+ var serial uint32
257+ call := func() {
258+ e, s := c.NewMethodCall(busName, busPath, busName, "GetConnectionUnixProcessID", "s")
259+ e.Str(c.UniqueName())
260+ sent, err := c.Send()
261+ if err != nil || sent != s {
262+ t.Fatalf("Send = %d, %v; want %d", sent, err, s)
263+ }
264+ serial = s
265+ }
266+ call()
267+ if allocs := testing.AllocsPerRun(200, call); allocs != 0 {
268+ t.Fatalf("%v allocations per method call, want 0", allocs)
269+ }
270+ // The replies must be read for the test to leave the bus quiet; they
271+ // also prove the calls were well formed.
272+ deadline := time.Now().Add(5 * time.Second)
273+ for got := 0; got < 202 && time.Now().Before(deadline); {
274+ m, err := c.ReadMessage()
275+ if err != nil {
276+ t.Fatal(err)
277+ }
278+ if m.Type == TypeMethodReturn && m.ReplySerial <= serial {
279+ got++
280+ }
281+ }
282+}
283+
284+func TestReadMessageAllocations(t *testing.T) {
285+ addr := privateBus(t)
286+ t.Run("method call", func(t *testing.T) {
287+ client, server := dial(t, addr), dial(t, addr)
288+ ping := func() {
289+ e := client.NewCall(server.UniqueName(), "/a", "org.example.Alloc", "Ping", "s")
290+ e.SetFlags(FlagNoReplyExpected)
291+ e.Str("hello")
292+ if _, err := client.Send(); err != nil {
293+ t.Fatal(err)
294+ }
295+ for {
296+ m, err := server.ReadMessage()
297+ if err != nil {
298+ t.Fatal(err)
299+ }
300+ if m.Type == TypeMethodCall && m.Member == "Ping" {
301+ if m.Body().Str() != "hello" {
302+ t.Fatal("bad body")
303+ }
304+ return
305+ }
306+ }
307+ }
308+ ping() // grow the buffers once
309+ if allocs := testing.AllocsPerRun(200, ping); allocs != 0 {
310+ t.Fatalf("%v allocations per method call read, want 0", allocs)
311+ }
312+ })
313+ t.Run("method return", func(t *testing.T) {
314+ c := dial(t, addr)
315+ ask := func() {
316+ e, serial := c.NewMethodCall(busName, busPath, busName, "GetNameOwner", "s")
317+ e.Str(busName)
318+ if _, err := c.Send(); err != nil {
319+ t.Fatal(err)
320+ }
321+ for {
322+ m, err := c.ReadMessage()
323+ if err != nil {
324+ t.Fatal(err)
325+ }
326+ if m.Type == TypeMethodReturn && m.ReplySerial == serial {
327+ if m.Body().Str() == "" {
328+ t.Fatal("empty owner")
329+ }
330+ return
331+ }
332+ }
333+ }
334+ ask()
335+ if allocs := testing.AllocsPerRun(200, ask); allocs != 0 {
336+ t.Fatalf("%v allocations per method return read, want 0", allocs)
337+ }
338+ })
339+}
@@ -0,0 +1,339 @@
1+package zerobus
2+
3+import (
4+ "net"
5+ "path/filepath"
6+ "syscall"
7+ "testing"
8+ "time"
9+ "unsafe"
10+)
11+
12+// socketPair returns a Conn whose socket is the client end of a Unix socket,
13+// without the bus handshake, and the other end to write raw frames to it.
14+func socketPair(t *testing.T) (*Conn, net.Conn) {
15+ t.Helper()
16+ l, err := net.Listen("unix", filepath.Join(t.TempDir(), "s"))
17+ if err != nil {
18+ t.Fatal(err)
19+ }
20+ defer func() { _ = l.Close() }()
21+ uc, err := net.DialUnix("unix", nil, &net.UnixAddr{Name: l.Addr().String(), Net: "unix"})
22+ if err != nil {
23+ t.Fatal(err)
24+ }
25+ peer, err := l.Accept()
26+ if err != nil {
27+ t.Fatal(err)
28+ }
29+ t.Cleanup(func() { _ = uc.Close(); _ = peer.Close() })
30+ return &Conn{c: uc, rbuf: make([]byte, defaultBufSz)}, peer
31+}
32+
33+func TestNewSignalToSetsDestination(t *testing.T) {
34+ var c Conn
35+ c.NewSignalTo(":1.42", "/a", "a.b", "Hit", "s").Str("x")
36+ b, err := c.enc.finish()
37+ if err != nil {
38+ t.Fatal(err)
39+ }
40+ var m Message
41+ if err := parse(b, &m); err != nil {
42+ t.Fatal(err)
43+ }
44+ if m.Type != TypeSignal || m.Destination != ":1.42" || m.Path != "/a" || m.Interface != "a.b" || m.Member != "Hit" {
45+ t.Fatalf("message %+v", m)
46+ }
47+ if got := m.Body().Str(); got != "x" {
48+ t.Fatalf("body %q", got)
49+ }
50+
51+ c.NewSignal("/a", "a.b", "Hit", "")
52+ b, err = c.enc.finish()
53+ if err != nil {
54+ t.Fatal(err)
55+ }
56+ if err := parse(b, &m); err != nil || m.Destination != "" {
57+ t.Fatalf("broadcast signal has destination %q (%v)", m.Destination, err)
58+ }
59+
60+ c.NewSignalTo("not a name", "/a", "a.b", "Hit", "")
61+ if _, err := c.Send(); err == nil {
62+ t.Fatal("invalid destination accepted")
63+ }
64+}
65+
66+func TestDirectedSignalReachesOnlyDestination(t *testing.T) {
67+ addr := privateBus(t)
68+ server := dial(t, addr)
69+ target := dial(t, addr)
70+ other := dial(t, addr)
71+ const rule = "type='signal',interface='org.example.Directed'"
72+ for _, c := range []*Conn{target, other} {
73+ if err := c.AddMatch(rule); err != nil {
74+ t.Fatal(err)
75+ }
76+ }
77+
78+ server.NewSignalTo(target.UniqueName(), "/a", "org.example.Directed", "Hit", "u").Uint32(7)
79+ if _, err := server.Send(); err != nil {
80+ t.Fatal(err)
81+ }
82+
83+ // The bus routes in order: a signal for other would come before the
84+ // reply to its round trip.
85+ var leaked bool
86+ other.Handle = func(m *Message) { leaked = leaked || m.Member == "Hit" }
87+ server.NewCall(busName, busPath, busName, "GetId", "")
88+ if _, err := server.Call(); err != nil {
89+ t.Fatal(err)
90+ }
91+ other.NewCall(busName, busPath, busName, "GetId", "")
92+ if _, err := other.Call(); err != nil {
93+ t.Fatal(err)
94+ }
95+ if leaked {
96+ t.Fatal("directed signal reached another peer")
97+ }
98+
99+ for {
100+ m, err := target.ReadMessage()
101+ if err != nil {
102+ t.Fatal(err)
103+ }
104+ if m.Member != "Hit" {
105+ continue
106+ }
107+ if m.Type != TypeSignal || m.Sender != server.UniqueName() || m.Destination != target.UniqueName() || m.Body().Uint32() != 7 {
108+ t.Fatalf("message %+v", m)
109+ }
110+ return
111+ }
112+}
113+
114+func TestAsyncMethodCallMatching(t *testing.T) {
115+ addr := privateBus(t)
116+ c := dial(t, addr)
117+ other := dial(t, addr)
118+
119+ // Two requests are in flight at once; the loop only uses Send and
120+ // ReadMessage and tells the replies apart by their serial.
121+ want := map[uint32]string{}
122+ for _, name := range []string{c.UniqueName(), other.UniqueName(), "org.example.Nobody"} {
123+ e, serial := c.NewMethodCall(busName, busPath, busName, "GetNameOwner", "s")
124+ e.Str(name)
125+ sent, err := c.Send()
126+ if err != nil {
127+ t.Fatal(err)
128+ }
129+ if sent != serial {
130+ t.Fatalf("NewMethodCall serial %d, Send serial %d", serial, sent)
131+ }
132+ if _, dup := want[serial]; dup {
133+ t.Fatalf("serial %d reused", serial)
134+ }
135+ want[serial] = name
136+ }
137+
138+ for len(want) > 0 {
139+ m, err := c.ReadMessage()
140+ if err != nil {
141+ t.Fatal(err)
142+ }
143+ if m.Type != TypeMethodReturn && m.Type != TypeError {
144+ continue // NameAcquired
145+ }
146+ name, ok := want[m.ReplySerial]
147+ if !ok {
148+ t.Fatalf("unexpected reply to serial %d", m.ReplySerial)
149+ }
150+ delete(want, m.ReplySerial)
151+ switch name {
152+ case "org.example.Nobody":
153+ if m.Type != TypeError {
154+ t.Fatalf("reply for %q is %d, want an error", name, m.Type)
155+ }
156+ default:
157+ if m.Type != TypeMethodReturn || m.Body().Str() != name {
158+ t.Fatalf("reply for %q is %d", name, m.Type)
159+ }
160+ }
161+ }
162+}
163+
164+func TestBufferedAndPollLoop(t *testing.T) {
165+ c, peer := socketPair(t)
166+ if c.Buffered() {
167+ t.Fatal("empty connection is buffered")
168+ }
169+ two := append(signal("One"), signal("Two")...)
170+ three := signal("Three")
171+ // Two whole frames and the start of a third arrive in one write.
172+ if _, err := peer.Write(append(two, three[:10]...)); err != nil {
173+ t.Fatal(err)
174+ }
175+ waitReadable(t, c)
176+ for _, want := range []string{"One", "Two"} {
177+ m, err := c.ReadMessage()
178+ if err != nil || m.Member != want {
179+ t.Fatalf("got %v, %v; want %q", m, err, want)
180+ }
181+ if got, wantBuf := c.Buffered(), want == "One"; got != wantBuf {
182+ t.Fatalf("Buffered after %q = %v, want %v", want, got, wantBuf)
183+ }
184+ }
185+ // The head of a third frame is not a message.
186+ if c.Buffered() {
187+ t.Fatal("a partial frame counts as buffered")
188+ }
189+ if _, err := peer.Write(three[10:]); err != nil {
190+ t.Fatal(err)
191+ }
192+ waitReadable(t, c)
193+ m, err := c.ReadMessage()
194+ if err != nil || m.Member != "Three" {
195+ t.Fatalf("got %v, %v", m, err)
196+ }
197+ if c.Buffered() {
198+ t.Fatal("drained connection is buffered")
199+ }
200+}
201+
202+func TestBufferedReportsStickyError(t *testing.T) {
203+ c, peer := socketPair(t)
204+ bad := signal("Bad")
205+ bad[3] = 9 // protocol version
206+ if _, err := peer.Write(append(bad, signal("Next")...)); err != nil {
207+ t.Fatal(err)
208+ }
209+ if _, err := c.ReadMessage(); err == nil {
210+ t.Fatal("malformed frame accepted")
211+ }
212+ if !c.Buffered() {
213+ t.Fatal("a sticky error must not make a poll loop wait")
214+ }
215+}
216+
217+// waitReadable blocks until the socket of c is readable, as a poll loop does.
218+func waitReadable(t *testing.T, c *Conn) {
219+ t.Helper()
220+ rc, err := c.SyscallConn()
221+ if err != nil {
222+ t.Fatal(err)
223+ }
224+ var fd int
225+ if err := rc.Control(func(f uintptr) { fd = int(f) }); err != nil {
226+ t.Fatal(err)
227+ }
228+ var set syscall.FdSet
229+ bits := int(unsafe.Sizeof(set.Bits[0])) * 8
230+ set.Bits[fd/bits] |= 1 << (uint(fd) % uint(bits))
231+ tv := syscall.Timeval{Sec: 5}
232+ n, err := syscall.Select(fd+1, &set, nil, nil, &tv)
233+ if err != nil || n != 1 {
234+ t.Fatalf("select = %d, %v", n, err)
235+ }
236+}
237+
238+func TestSignalAllocations(t *testing.T) {
239+ c := dial(t, privateBus(t))
240+ send := func() {
241+ e := c.NewSignalTo(":1.1", "/a", "org.example.Alloc", "Hit", "us")
242+ e.Uint32(1)
243+ e.Str("x")
244+ if _, err := c.Send(); err != nil {
245+ t.Fatal(err)
246+ }
247+ }
248+ send() // grow the buffers once
249+ if allocs := testing.AllocsPerRun(200, send); allocs != 0 {
250+ t.Fatalf("%v allocations per directed signal, want 0", allocs)
251+ }
252+}
253+
254+func TestMethodCallAllocations(t *testing.T) {
255+ c := dial(t, privateBus(t))
256+ var serial uint32
257+ call := func() {
258+ e, s := c.NewMethodCall(busName, busPath, busName, "GetConnectionUnixProcessID", "s")
259+ e.Str(c.UniqueName())
260+ sent, err := c.Send()
261+ if err != nil || sent != s {
262+ t.Fatalf("Send = %d, %v; want %d", sent, err, s)
263+ }
264+ serial = s
265+ }
266+ call()
267+ if allocs := testing.AllocsPerRun(200, call); allocs != 0 {
268+ t.Fatalf("%v allocations per method call, want 0", allocs)
269+ }
270+ // The replies must be read for the test to leave the bus quiet; they
271+ // also prove the calls were well formed.
272+ deadline := time.Now().Add(5 * time.Second)
273+ for got := 0; got < 202 && time.Now().Before(deadline); {
274+ m, err := c.ReadMessage()
275+ if err != nil {
276+ t.Fatal(err)
277+ }
278+ if m.Type == TypeMethodReturn && m.ReplySerial <= serial {
279+ got++
280+ }
281+ }
282+}
283+
284+func TestReadMessageAllocations(t *testing.T) {
285+ addr := privateBus(t)
286+ t.Run("method call", func(t *testing.T) {
287+ client, server := dial(t, addr), dial(t, addr)
288+ ping := func() {
289+ e := client.NewCall(server.UniqueName(), "/a", "org.example.Alloc", "Ping", "s")
290+ e.SetFlags(FlagNoReplyExpected)
291+ e.Str("hello")
292+ if _, err := client.Send(); err != nil {
293+ t.Fatal(err)
294+ }
295+ for {
296+ m, err := server.ReadMessage()
297+ if err != nil {
298+ t.Fatal(err)
299+ }
300+ if m.Type == TypeMethodCall && m.Member == "Ping" {
301+ if m.Body().Str() != "hello" {
302+ t.Fatal("bad body")
303+ }
304+ return
305+ }
306+ }
307+ }
308+ ping() // grow the buffers once
309+ if allocs := testing.AllocsPerRun(200, ping); allocs != 0 {
310+ t.Fatalf("%v allocations per method call read, want 0", allocs)
311+ }
312+ })
313+ t.Run("method return", func(t *testing.T) {
314+ c := dial(t, addr)
315+ ask := func() {
316+ e, serial := c.NewMethodCall(busName, busPath, busName, "GetNameOwner", "s")
317+ e.Str(busName)
318+ if _, err := c.Send(); err != nil {
319+ t.Fatal(err)
320+ }
321+ for {
322+ m, err := c.ReadMessage()
323+ if err != nil {
324+ t.Fatal(err)
325+ }
326+ if m.Type == TypeMethodReturn && m.ReplySerial == serial {
327+ if m.Body().Str() == "" {
328+ t.Fatal("empty owner")
329+ }
330+ return
331+ }
332+ }
333+ }
334+ ask()
335+ if allocs := testing.AllocsPerRun(200, ask); allocs != 0 {
336+ t.Fatalf("%v allocations per method return read, want 0", allocs)
337+ }
338+ })
339+}

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 1

Tags 3

↑↓ moveenter openesc close