Merge pull request 'feat(conn): add directed signals, async calls and poll helpers' (#2) from feat/server-helpers into main

Reviewed-on: https://git.bnema.dev/bnema/zerobus/pulls/2

commit: 07ab4d20ede3a4f77f2854176e8d98872e46f9e0
author: bnema · 7d ago · 2026-10-15 03:07 UTC
merge of: 93dc6b1 + e3bde5d · diff against the first parent
signature: SSH UnverifiedSHA256:VOCA5lYARHcbKDZ7GhTV45VmcjWnJ6ZQIBazj17LivA The key is unknown
refs: maintag: v0.2.0

6 files · +461 -5

M .github/workflows/ci.yml +10 -5


@@ -9,17 +9,22 @@ concurrency:
9	9		group: ci-${{ github.ref }}
10	10		cancel-in-progress: true
11	11		jobs:
	12	+	# The job runs as root in an Arch container, on Forgejo and GitHub alike.
	13	+	# dbus-daemon runs the integration tests against a private bus; gcc is for
	14	+	# the race detector.
12	15		check:
13	16		runs-on: ubuntu-latest
	17	+	container:
	18	+	image: archlinux:latest
	19	+	options: --init
14	20		steps:
	21	+	- run: pacman -Syu --noconfirm --needed git go gcc dbus
15	22		- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
16	23		with:
17	24		persist-credentials: false
18		-	- uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7
19		-	with:
20		-	go-version-file: go.mod
21		-	# dbus-daemon runs the integration tests against a private bus.
22		-	- run: sudo apt-get update && sudo apt-get install -y --no-install-recommends dbus
	25	+	- run: \|
	26	+	git config --global --add safe.directory "$GITHUB_WORKSPACE"
	27	+	test "$(go env GOVERSION \| cut -d. -f1-2)" = "go$(sed -n 's/^go $[0-9]\.[0-9]$.*/\1/p' go.mod)"
23	28		- run: test -z "$(gofmt -l .)"
24	29		- run: go vet ./...
25	30		- run: GOARCH=386 go vet ./...

   group: ci-${{ github.ref }}
   cancel-in-progress: true
 jobs:
+  # The job runs as root in an Arch container, on Forgejo and GitHub alike.
+  # dbus-daemon runs the integration tests against a private bus; gcc is for
+  # the race detector.
   check:
     runs-on: ubuntu-latest
+    container:
+      image: archlinux:latest
+      options: --init
     steps:
+      - run: pacman -Syu --noconfirm --needed git go gcc dbus
       - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
         with:
           persist-credentials: false
-      - uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7
+      - run: |
-        with:
+          git config --global --add safe.directory "$GITHUB_WORKSPACE"
-          go-version-file: go.mod
+          test "$(go env GOVERSION | cut -d. -f1-2)" = "go$(sed -n 's/^go \([0-9]*\.[0-9]*\).*/\1/p' go.mod)"
-      # dbus-daemon runs the integration tests against a private bus.
-      - run: sudo apt-get update && sudo apt-get install -y --no-install-recommends dbus
       - run: test -z "$(gofmt -l .)"
       - run: go vet ./...
       - run: GOARCH=386 go vet ./...

M conn.go +47 -0


@@ -10,6 +10,7 @@ import (
10	10		"os"
11	11		"strconv"
12	12		"strings"
	13	+	"syscall"
13	14		)
14	15
15	16		const (

@@ -264,6 +265,27 @@ func (c Conn) NewSignal(path, iface, member, sig string) Encoder {
264	265		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. An
	271	+	// empty dest makes Send fail: use NewSignal to broadcast.
	272	+	func (c Conn) NewSignalTo(dest, path, iface, member, sig string) Encoder {
	273	+	c.enc.begin(TypeSignal, c.nextSerial(), 0, dest, path, iface, member, "", sig)
	274	+	if dest == "" {
	275	+	c.enc.invalid()
	276	+	}
	277	+	return &c.enc
	278	+	}
	279	+
	280	+	// NewMethodCall starts a method call like NewCall and also returns its
	281	+	// serial. A loop that only uses Send and ReadMessage keeps the serial to
	282	+	// recognize the reply later, as Message.ReplySerial, without blocking in Call.
	283	+	// The serial is the one Send returns for this message.
	284	+	func (c Conn) NewMethodCall(dest, path, iface, member, sig string) (Encoder, uint32) {
	285	+	e := c.NewCall(dest, path, iface, member, sig)
	286	+	return e, c.serial
	287	+	}
	288	+
267	289		// NewReply starts the reply to the method call m and returns the encoder for
268	290		// its body.
269	291		func (c Conn) NewReply(m Message, sig string) *Encoder {

@@ -348,6 +370,31 @@ func replyError(m Message) Error {
348	370		return e
349	371		}
350	372
	373	+	// SyscallConn returns the raw socket of the connection, for a poll loop that
	374	+	// waits for it to become readable together with other file descriptors.
	375	+	// Only wait on the descriptor: read, write and close through the Conn.
	376	+	//
	377	+	// Bytes already read from the socket are not visible to poll: check Buffered
	378	+	// before waiting. After the descriptor turns readable, ReadMessage may still
	379	+	// block until the rest of a partially received message arrives.
	380	+	func (c *Conn) SyscallConn() (syscall.RawConn, error) { return c.c.SyscallConn() }
	381	+
	382	+	// Buffered reports whether ReadMessage can return without waiting for the
	383	+	// socket: a complete message, or a sticky error, is already in the receive
	384	+	// buffer. A poll loop must call ReadMessage until Buffered is false before it
	385	+	// waits on the descriptor again, because ReadMessage reads everything the
	386	+	// socket has and a later poll would not see those bytes.
	387	+	func (c *Conn) Buffered() bool {
	388	+	if c.err != nil {
	389	+	return true
	390	+	}
	391	+	if c.end-c.start < headerLen {
	392	+	return false
	393	+	}
	394	+	n, err := frameLen(c.rbuf[c.start:c.end])
	395	+	return err != nil \|\| c.end-c.start >= n
	396	+	}
	397	+
351	398		// ReadMessage blocks until the next message arrives. The message and its
352	399		// strings are valid until the next call to ReadMessage or Call.
353	400		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,27 @@ func (c Conn) NewSignal(path, iface, member, sig string) Encoder {
264	return &c.enc	265		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. An
		271	+	// empty dest makes Send fail: use NewSignal to broadcast.
		272	+	func (c Conn) NewSignalTo(dest, path, iface, member, sig string) Encoder {
		273	+	c.enc.begin(TypeSignal, c.nextSerial(), 0, dest, path, iface, member, "", sig)
		274	+	if dest == "" {
		275	+	c.enc.invalid()
		276	+	}
		277	+	return &c.enc
		278	+	}
		279	+
		280	+	// NewMethodCall starts a method call like NewCall and also returns its
		281	+	// serial. A loop that only uses Send and ReadMessage keeps the serial to
		282	+	// recognize the reply later, as Message.ReplySerial, without blocking in Call.
		283	+	// The serial is the one Send returns for this message.
		284	+	func (c Conn) NewMethodCall(dest, path, iface, member, sig string) (Encoder, uint32) {
		285	+	e := c.NewCall(dest, path, iface, member, sig)
		286	+	return e, c.serial
		287	+	}
		288	+
267	// NewReply starts the reply to the method call m and returns the encoder for	289		// NewReply starts the reply to the method call m and returns the encoder for
268	// its body.	290		// its body.
269	func (c Conn) NewReply(m Message, sig string) *Encoder {	291		func (c Conn) NewReply(m Message, sig string) *Encoder {

@@ -348,6 +370,31 @@ func replyError(m Message) Error {
348	return e	370		return e
349	}	371		}
350		372
		373	+	// SyscallConn returns the raw socket of the connection, for a poll loop that
		374	+	// waits for it to become readable together with other file descriptors.
		375	+	// Only wait on the descriptor: read, write and close through the Conn.
		376	+	//
		377	+	// Bytes already read from the socket are not visible to poll: check Buffered
		378	+	// before waiting. After the descriptor turns readable, ReadMessage may still
		379	+	// block until the rest of a partially received message arrives.
		380	+	func (c *Conn) SyscallConn() (syscall.RawConn, error) { return c.c.SyscallConn() }
		381	+
		382	+	// Buffered reports whether ReadMessage can return without waiting for the
		383	+	// socket: a complete message, or a sticky error, is already in the receive
		384	+	// buffer. A poll loop must call ReadMessage until Buffered is false before it
		385	+	// waits on the descriptor again, because ReadMessage reads everything the
		386	+	// socket has and a later poll would not see those bytes.
		387	+	func (c *Conn) Buffered() bool {
		388	+	if c.err != nil {
		389	+	return true
		390	+	}
		391	+	if c.end-c.start < headerLen {
		392	+	return false
		393	+	}
		394	+	n, err := frameLen(c.rbuf[c.start:c.end])
		395	+	return err != nil \|\| c.end-c.start >= n
		396	+	}
		397	+
351	// ReadMessage blocks until the next message arrives. The message and its	398		// ReadMessage blocks until the next message arrives. The message and its
352	// strings are valid until the next call to ReadMessage or Call.	399		// strings are valid until the next call to ReadMessage or Call.
353	func (c Conn) ReadMessage() (Message, error) {	400		func (c Conn) ReadMessage() (Message, error) {

M doc.go +21 -0


@@ -2,4 +2,25 @@
2	2		// state. It speaks the D-Bus wire protocol over a Unix socket and decodes
3	3		// messages in place, so strings and arrays point into the receive buffer and
4	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	26		package zerobus


@@ -2,4 +2,25 @@
2	// state. It speaks the D-Bus wire protocol over a Unix socket and decodes	2		// 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 and	3		// 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 zerobus	26		package zerobus

A server_test.go +343 -0

@@ -0,0 +1,343 @@
	1	+	package zerobus
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"net"
	6	+	"path/filepath"
	7	+	"testing"
	8	+	"time"
	9	+	)
	10	+
	11	+	// socketPair returns a Conn whose socket is the client end of a Unix socket,
	12	+	// without the bus handshake, and the other end to write raw frames to it.
	13	+	func socketPair(t testing.T) (Conn, net.Conn) {
	14	+	t.Helper()
	15	+	l, err := net.Listen("unix", filepath.Join(t.TempDir(), "s"))
	16	+	if err != nil {
	17	+	t.Fatal(err)
	18	+	}
	19	+	defer func() { _ = l.Close() }()
	20	+	uc, err := net.DialUnix("unix", nil, &net.UnixAddr{Name: l.Addr().String(), Net: "unix"})
	21	+	if err != nil {
	22	+	t.Fatal(err)
	23	+	}
	24	+	peer, err := l.Accept()
	25	+	if err != nil {
	26	+	t.Fatal(err)
	27	+	}
	28	+	t.Cleanup(func() { _ = uc.Close(); _ = peer.Close() })
	29	+	return &Conn{c: uc, rbuf: make([]byte, defaultBufSz)}, peer
	30	+	}
	31	+
	32	+	func TestNewSignalToSetsDestination(t *testing.T) {
	33	+	var c Conn
	34	+	c.NewSignalTo(":1.42", "/a", "a.b", "Hit", "s").Str("x")
	35	+	b, err := c.enc.finish()
	36	+	if err != nil {
	37	+	t.Fatal(err)
	38	+	}
	39	+	var m Message
	40	+	if err := parse(b, &m); err != nil {
	41	+	t.Fatal(err)
	42	+	}
	43	+	if m.Type != TypeSignal \|\| m.Destination != ":1.42" \|\| m.Path != "/a" \|\| m.Interface != "a.b" \|\| m.Member != "Hit" {
	44	+	t.Fatalf("message %+v", m)
	45	+	}
	46	+	if got := m.Body().Str(); got != "x" {
	47	+	t.Fatalf("body %q", got)
	48	+	}
	49	+
	50	+	c.NewSignal("/a", "a.b", "Hit", "")
	51	+	b, err = c.enc.finish()
	52	+	if err != nil {
	53	+	t.Fatal(err)
	54	+	}
	55	+	if err := parse(b, &m); err != nil \|\| m.Destination != "" {
	56	+	t.Fatalf("broadcast signal has destination %q (%v)", m.Destination, err)
	57	+	}
	58	+
	59	+	c.NewSignalTo("not a name", "/a", "a.b", "Hit", "")
	60	+	if _, err := c.Send(); err == nil {
	61	+	t.Fatal("invalid destination accepted")
	62	+	}
	63	+
	64	+	c.NewSignalTo("", "/a", "a.b", "Hit", "")
	65	+	if _, err := c.Send(); err == nil {
	66	+	t.Fatal("directed signal without destination accepted")
	67	+	}
	68	+	}
	69	+
	70	+	func TestDirectedSignalReachesOnlyDestination(t *testing.T) {
	71	+	addr := privateBus(t)
	72	+	server := dial(t, addr)
	73	+	target := dial(t, addr)
	74	+	other := dial(t, addr)
	75	+	const rule = "type='signal',interface='org.example.Directed'"
	76	+	for _, c := range []*Conn{target, other} {
	77	+	if err := c.AddMatch(rule); err != nil {
	78	+	t.Fatal(err)
	79	+	}
	80	+	}
	81	+
	82	+	server.NewSignalTo(target.UniqueName(), "/a", "org.example.Directed", "Hit", "u").Uint32(7)
	83	+	if _, err := server.Send(); err != nil {
	84	+	t.Fatal(err)
	85	+	}
	86	+
	87	+	// The bus routes in order: a signal for other would come before the
	88	+	// reply to its round trip.
	89	+	var leaked bool
	90	+	other.Handle = func(m *Message) { leaked = leaked \|\| m.Member == "Hit" }
	91	+	server.NewCall(busName, busPath, busName, "GetId", "")
	92	+	if _, err := server.Call(); err != nil {
	93	+	t.Fatal(err)
	94	+	}
	95	+	other.NewCall(busName, busPath, busName, "GetId", "")
	96	+	if _, err := other.Call(); err != nil {
	97	+	t.Fatal(err)
	98	+	}
	99	+	if leaked {
	100	+	t.Fatal("directed signal reached another peer")
	101	+	}
	102	+
	103	+	for {
	104	+	m, err := target.ReadMessage()
	105	+	if err != nil {
	106	+	t.Fatal(err)
	107	+	}
	108	+	if m.Member != "Hit" {
	109	+	continue
	110	+	}
	111	+	if m.Type != TypeSignal \|\| m.Sender != server.UniqueName() \|\| m.Destination != target.UniqueName() \|\| m.Body().Uint32() != 7 {
	112	+	t.Fatalf("message %+v", m)
	113	+	}
	114	+	return
	115	+	}
	116	+	}
	117	+
	118	+	func TestAsyncMethodCallMatching(t *testing.T) {
	119	+	addr := privateBus(t)
	120	+	c := dial(t, addr)
	121	+	other := dial(t, addr)
	122	+
	123	+	// Two requests are in flight at once; the loop only uses Send and
	124	+	// ReadMessage and tells the replies apart by their serial.
	125	+	want := map[uint32]string{}
	126	+	for _, name := range []string{c.UniqueName(), other.UniqueName(), "org.example.Nobody"} {
	127	+	e, serial := c.NewMethodCall(busName, busPath, busName, "GetNameOwner", "s")
	128	+	e.Str(name)
	129	+	sent, err := c.Send()
	130	+	if err != nil {
	131	+	t.Fatal(err)
	132	+	}
	133	+	if sent != serial {
	134	+	t.Fatalf("NewMethodCall serial %d, Send serial %d", serial, sent)
	135	+	}
	136	+	if _, dup := want[serial]; dup {
	137	+	t.Fatalf("serial %d reused", serial)
	138	+	}
	139	+	want[serial] = name
	140	+	}
	141	+
	142	+	for len(want) > 0 {
	143	+	m, err := c.ReadMessage()
	144	+	if err != nil {
	145	+	t.Fatal(err)
	146	+	}
	147	+	if m.Type != TypeMethodReturn && m.Type != TypeError {
	148	+	continue // NameAcquired
	149	+	}
	150	+	name, ok := want[m.ReplySerial]
	151	+	if !ok {
	152	+	t.Fatalf("unexpected reply to serial %d", m.ReplySerial)
	153	+	}
	154	+	delete(want, m.ReplySerial)
	155	+	switch name {
	156	+	case "org.example.Nobody":
	157	+	if m.Type != TypeError {
	158	+	t.Fatalf("reply for %q is %d, want an error", name, m.Type)
	159	+	}
	160	+	default:
	161	+	if m.Type != TypeMethodReturn \|\| m.Body().Str() != name {
	162	+	t.Fatalf("reply for %q is %d", name, m.Type)
	163	+	}
	164	+	}
	165	+	}
	166	+	}
	167	+
	168	+	func TestBufferedAndPollLoop(t *testing.T) {
	169	+	c, peer := socketPair(t)
	170	+	if c.Buffered() {
	171	+	t.Fatal("empty connection is buffered")
	172	+	}
	173	+	two := append(signal("One"), signal("Two")...)
	174	+	three := signal("Three")
	175	+	// Two whole frames and the start of a third arrive in one write.
	176	+	if _, err := peer.Write(append(two, three[:10]...)); err != nil {
	177	+	t.Fatal(err)
	178	+	}
	179	+	waitReadable(t, c)
	180	+	for _, want := range []string{"One", "Two"} {
	181	+	m, err := c.ReadMessage()
	182	+	if err != nil \|\| m.Member != want {
	183	+	t.Fatalf("got %v, %v; want %q", m, err, want)
	184	+	}
	185	+	if got, wantBuf := c.Buffered(), want == "One"; got != wantBuf {
	186	+	t.Fatalf("Buffered after %q = %v, want %v", want, got, wantBuf)
	187	+	}
	188	+	}
	189	+	// The head of a third frame is not a message.
	190	+	if c.Buffered() {
	191	+	t.Fatal("a partial frame counts as buffered")
	192	+	}
	193	+	if _, err := peer.Write(three[10:]); err != nil {
	194	+	t.Fatal(err)
	195	+	}
	196	+	waitReadable(t, c)
	197	+	m, err := c.ReadMessage()
	198	+	if err != nil \|\| m.Member != "Three" {
	199	+	t.Fatalf("got %v, %v", m, err)
	200	+	}
	201	+	if c.Buffered() {
	202	+	t.Fatal("drained connection is buffered")
	203	+	}
	204	+	}
	205	+
	206	+	func TestBufferedReportsStickyError(t *testing.T) {
	207	+	c, peer := socketPair(t)
	208	+	bad := signal("Bad")
	209	+	bad[3] = 9 // protocol version
	210	+	if _, err := peer.Write(append(bad, signal("Next")...)); err != nil {
	211	+	t.Fatal(err)
	212	+	}
	213	+	if _, err := c.ReadMessage(); err == nil {
	214	+	t.Fatal("malformed frame accepted")
	215	+	}
	216	+	if !c.Buffered() {
	217	+	t.Fatal("a sticky error must not make a poll loop wait")
	218	+	}
	219	+	}
	220	+
	221	+	// waitReadable blocks until the socket of c is readable, as a poll loop does.
	222	+	func waitReadable(t testing.T, c Conn) {
	223	+	t.Helper()
	224	+	rc, err := c.SyscallConn()
	225	+	if err != nil {
	226	+	t.Fatal(err)
	227	+	}
	228	+	var fd int
	229	+	if err := rc.Control(func(f uintptr) { fd = int(f) }); err != nil {
	230	+	t.Fatal(err)
	231	+	}
	232	+	if err := waitFd(fd, 5*time.Second); errors.Is(err, errNoWaitFd) {
	233	+	t.Skip(err)
	234	+	} else if err != nil {
	235	+	t.Fatal(err)
	236	+	}
	237	+	}
	238	+
	239	+	func TestSignalAllocations(t *testing.T) {
	240	+	c := dial(t, privateBus(t))
	241	+	send := func() {
	242	+	e := c.NewSignalTo(":1.1", "/a", "org.example.Alloc", "Hit", "us")
	243	+	e.Uint32(1)
	244	+	e.Str("x")
	245	+	if _, err := c.Send(); err != nil {
	246	+	t.Fatal(err)
	247	+	}
	248	+	}
	249	+	send() // grow the buffers once
	250	+	if allocs := testing.AllocsPerRun(200, send); allocs != 0 {
	251	+	t.Fatalf("%v allocations per directed signal, want 0", allocs)
	252	+	}
	253	+	}
	254	+
	255	+	func TestMethodCallAllocations(t *testing.T) {
	256	+	c := dial(t, privateBus(t))
	257	+	var serial uint32
	258	+	call := func() {
	259	+	e, s := c.NewMethodCall(busName, busPath, busName, "GetConnectionUnixProcessID", "s")
	260	+	e.Str(c.UniqueName())
	261	+	sent, err := c.Send()
	262	+	if err != nil \|\| sent != s {
	263	+	t.Fatalf("Send = %d, %v; want %d", sent, err, s)
	264	+	}
	265	+	serial = s
	266	+	}
	267	+	call()
	268	+	if allocs := testing.AllocsPerRun(200, call); allocs != 0 {
	269	+	t.Fatalf("%v allocations per method call, want 0", allocs)
	270	+	}
	271	+	// The replies must be read for the test to leave the bus quiet; they
	272	+	// also prove the calls were well formed.
	273	+	if err := c.c.SetReadDeadline(time.Now().Add(5 * time.Second)); err != nil {
	274	+	t.Fatal(err)
	275	+	}
	276	+	got := 0
	277	+	for got < 202 {
	278	+	m, err := c.ReadMessage()
	279	+	if err != nil {
	280	+	t.Fatalf("after %d of 202 replies: %v", got, err)
	281	+	}
	282	+	if m.Type == TypeMethodReturn && m.ReplySerial <= serial {
	283	+	got++
	284	+	}
	285	+	}
	286	+	}
	287	+
	288	+	func TestReadMessageAllocations(t *testing.T) {
	289	+	addr := privateBus(t)
	290	+	t.Run("method call", func(t *testing.T) {
	291	+	client, server := dial(t, addr), dial(t, addr)
	292	+	ping := func() {
	293	+	e := client.NewCall(server.UniqueName(), "/a", "org.example.Alloc", "Ping", "s")
	294	+	e.SetFlags(FlagNoReplyExpected)
	295	+	e.Str("hello")
	296	+	if _, err := client.Send(); err != nil {
	297	+	t.Fatal(err)
	298	+	}
	299	+	for {
	300	+	m, err := server.ReadMessage()
	301	+	if err != nil {
	302	+	t.Fatal(err)
	303	+	}
	304	+	if m.Type == TypeMethodCall && m.Member == "Ping" {
	305	+	if m.Body().Str() != "hello" {
	306	+	t.Fatal("bad body")
	307	+	}
	308	+	return
	309	+	}
	310	+	}
	311	+	}
	312	+	ping() // grow the buffers once
	313	+	if allocs := testing.AllocsPerRun(200, ping); allocs != 0 {
	314	+	t.Fatalf("%v allocations per method call read, want 0", allocs)
	315	+	}
	316	+	})
	317	+	t.Run("method return", func(t *testing.T) {
	318	+	c := dial(t, addr)
	319	+	ask := func() {
	320	+	e, serial := c.NewMethodCall(busName, busPath, busName, "GetNameOwner", "s")
	321	+	e.Str(busName)
	322	+	if _, err := c.Send(); err != nil {
	323	+	t.Fatal(err)
	324	+	}
	325	+	for {
	326	+	m, err := c.ReadMessage()
	327	+	if err != nil {
	328	+	t.Fatal(err)
	329	+	}
	330	+	if m.Type == TypeMethodReturn && m.ReplySerial == serial {
	331	+	if m.Body().Str() == "" {
	332	+	t.Fatal("empty owner")
	333	+	}
	334	+	return
	335	+	}
	336	+	}
	337	+	}
	338	+	ask()
	339	+	if allocs := testing.AllocsPerRun(200, ask); allocs != 0 {
	340	+	t.Fatalf("%v allocations per method return read, want 0", allocs)
	341	+	}
	342	+	})
	343	+	}

@@ -0,0 +1,343 @@
	1	+	package zerobus
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"net"
	6	+	"path/filepath"
	7	+	"testing"
	8	+	"time"
	9	+	)
	10	+
	11	+	// socketPair returns a Conn whose socket is the client end of a Unix socket,
	12	+	// without the bus handshake, and the other end to write raw frames to it.
	13	+	func socketPair(t testing.T) (Conn, net.Conn) {
	14	+	t.Helper()
	15	+	l, err := net.Listen("unix", filepath.Join(t.TempDir(), "s"))
	16	+	if err != nil {
	17	+	t.Fatal(err)
	18	+	}
	19	+	defer func() { _ = l.Close() }()
	20	+	uc, err := net.DialUnix("unix", nil, &net.UnixAddr{Name: l.Addr().String(), Net: "unix"})
	21	+	if err != nil {
	22	+	t.Fatal(err)
	23	+	}
	24	+	peer, err := l.Accept()
	25	+	if err != nil {
	26	+	t.Fatal(err)
	27	+	}
	28	+	t.Cleanup(func() { _ = uc.Close(); _ = peer.Close() })
	29	+	return &Conn{c: uc, rbuf: make([]byte, defaultBufSz)}, peer
	30	+	}
	31	+
	32	+	func TestNewSignalToSetsDestination(t *testing.T) {
	33	+	var c Conn
	34	+	c.NewSignalTo(":1.42", "/a", "a.b", "Hit", "s").Str("x")
	35	+	b, err := c.enc.finish()
	36	+	if err != nil {
	37	+	t.Fatal(err)
	38	+	}
	39	+	var m Message
	40	+	if err := parse(b, &m); err != nil {
	41	+	t.Fatal(err)
	42	+	}
	43	+	if m.Type != TypeSignal \|\| m.Destination != ":1.42" \|\| m.Path != "/a" \|\| m.Interface != "a.b" \|\| m.Member != "Hit" {
	44	+	t.Fatalf("message %+v", m)
	45	+	}
	46	+	if got := m.Body().Str(); got != "x" {
	47	+	t.Fatalf("body %q", got)
	48	+	}
	49	+
	50	+	c.NewSignal("/a", "a.b", "Hit", "")
	51	+	b, err = c.enc.finish()
	52	+	if err != nil {
	53	+	t.Fatal(err)
	54	+	}
	55	+	if err := parse(b, &m); err != nil \|\| m.Destination != "" {
	56	+	t.Fatalf("broadcast signal has destination %q (%v)", m.Destination, err)
	57	+	}
	58	+
	59	+	c.NewSignalTo("not a name", "/a", "a.b", "Hit", "")
	60	+	if _, err := c.Send(); err == nil {
	61	+	t.Fatal("invalid destination accepted")
	62	+	}
	63	+
	64	+	c.NewSignalTo("", "/a", "a.b", "Hit", "")
	65	+	if _, err := c.Send(); err == nil {
	66	+	t.Fatal("directed signal without destination accepted")
	67	+	}
	68	+	}
	69	+
	70	+	func TestDirectedSignalReachesOnlyDestination(t *testing.T) {
	71	+	addr := privateBus(t)
	72	+	server := dial(t, addr)
	73	+	target := dial(t, addr)
	74	+	other := dial(t, addr)
	75	+	const rule = "type='signal',interface='org.example.Directed'"
	76	+	for _, c := range []*Conn{target, other} {
	77	+	if err := c.AddMatch(rule); err != nil {
	78	+	t.Fatal(err)
	79	+	}
	80	+	}
	81	+
	82	+	server.NewSignalTo(target.UniqueName(), "/a", "org.example.Directed", "Hit", "u").Uint32(7)
	83	+	if _, err := server.Send(); err != nil {
	84	+	t.Fatal(err)
	85	+	}
	86	+
	87	+	// The bus routes in order: a signal for other would come before the
	88	+	// reply to its round trip.
	89	+	var leaked bool
	90	+	other.Handle = func(m *Message) { leaked = leaked \|\| m.Member == "Hit" }
	91	+	server.NewCall(busName, busPath, busName, "GetId", "")
	92	+	if _, err := server.Call(); err != nil {
	93	+	t.Fatal(err)
	94	+	}
	95	+	other.NewCall(busName, busPath, busName, "GetId", "")
	96	+	if _, err := other.Call(); err != nil {
	97	+	t.Fatal(err)
	98	+	}
	99	+	if leaked {
	100	+	t.Fatal("directed signal reached another peer")
	101	+	}
	102	+
	103	+	for {
	104	+	m, err := target.ReadMessage()
	105	+	if err != nil {
	106	+	t.Fatal(err)
	107	+	}
	108	+	if m.Member != "Hit" {
	109	+	continue
	110	+	}
	111	+	if m.Type != TypeSignal \|\| m.Sender != server.UniqueName() \|\| m.Destination != target.UniqueName() \|\| m.Body().Uint32() != 7 {
	112	+	t.Fatalf("message %+v", m)
	113	+	}
	114	+	return
	115	+	}
	116	+	}
	117	+
	118	+	func TestAsyncMethodCallMatching(t *testing.T) {
	119	+	addr := privateBus(t)
	120	+	c := dial(t, addr)
	121	+	other := dial(t, addr)
	122	+
	123	+	// Two requests are in flight at once; the loop only uses Send and
	124	+	// ReadMessage and tells the replies apart by their serial.
	125	+	want := map[uint32]string{}
	126	+	for _, name := range []string{c.UniqueName(), other.UniqueName(), "org.example.Nobody"} {
	127	+	e, serial := c.NewMethodCall(busName, busPath, busName, "GetNameOwner", "s")
	128	+	e.Str(name)
	129	+	sent, err := c.Send()
	130	+	if err != nil {
	131	+	t.Fatal(err)
	132	+	}
	133	+	if sent != serial {
	134	+	t.Fatalf("NewMethodCall serial %d, Send serial %d", serial, sent)
	135	+	}
	136	+	if _, dup := want[serial]; dup {
	137	+	t.Fatalf("serial %d reused", serial)
	138	+	}
	139	+	want[serial] = name
	140	+	}
	141	+
	142	+	for len(want) > 0 {
	143	+	m, err := c.ReadMessage()
	144	+	if err != nil {
	145	+	t.Fatal(err)
	146	+	}
	147	+	if m.Type != TypeMethodReturn && m.Type != TypeError {
	148	+	continue // NameAcquired
	149	+	}
	150	+	name, ok := want[m.ReplySerial]
	151	+	if !ok {
	152	+	t.Fatalf("unexpected reply to serial %d", m.ReplySerial)
	153	+	}
	154	+	delete(want, m.ReplySerial)
	155	+	switch name {
	156	+	case "org.example.Nobody":
	157	+	if m.Type != TypeError {
	158	+	t.Fatalf("reply for %q is %d, want an error", name, m.Type)
	159	+	}
	160	+	default:
	161	+	if m.Type != TypeMethodReturn \|\| m.Body().Str() != name {
	162	+	t.Fatalf("reply for %q is %d", name, m.Type)
	163	+	}
	164	+	}
	165	+	}
	166	+	}
	167	+
	168	+	func TestBufferedAndPollLoop(t *testing.T) {
	169	+	c, peer := socketPair(t)
	170	+	if c.Buffered() {
	171	+	t.Fatal("empty connection is buffered")
	172	+	}
	173	+	two := append(signal("One"), signal("Two")...)
	174	+	three := signal("Three")
	175	+	// Two whole frames and the start of a third arrive in one write.
	176	+	if _, err := peer.Write(append(two, three[:10]...)); err != nil {
	177	+	t.Fatal(err)
	178	+	}
	179	+	waitReadable(t, c)
	180	+	for _, want := range []string{"One", "Two"} {
	181	+	m, err := c.ReadMessage()
	182	+	if err != nil \|\| m.Member != want {
	183	+	t.Fatalf("got %v, %v; want %q", m, err, want)
	184	+	}
	185	+	if got, wantBuf := c.Buffered(), want == "One"; got != wantBuf {
	186	+	t.Fatalf("Buffered after %q = %v, want %v", want, got, wantBuf)
	187	+	}
	188	+	}
	189	+	// The head of a third frame is not a message.
	190	+	if c.Buffered() {
	191	+	t.Fatal("a partial frame counts as buffered")
	192	+	}
	193	+	if _, err := peer.Write(three[10:]); err != nil {
	194	+	t.Fatal(err)
	195	+	}
	196	+	waitReadable(t, c)
	197	+	m, err := c.ReadMessage()
	198	+	if err != nil \|\| m.Member != "Three" {
	199	+	t.Fatalf("got %v, %v", m, err)
	200	+	}
	201	+	if c.Buffered() {
	202	+	t.Fatal("drained connection is buffered")
	203	+	}
	204	+	}
	205	+
	206	+	func TestBufferedReportsStickyError(t *testing.T) {
	207	+	c, peer := socketPair(t)
	208	+	bad := signal("Bad")
	209	+	bad[3] = 9 // protocol version
	210	+	if _, err := peer.Write(append(bad, signal("Next")...)); err != nil {
	211	+	t.Fatal(err)
	212	+	}
	213	+	if _, err := c.ReadMessage(); err == nil {
	214	+	t.Fatal("malformed frame accepted")
	215	+	}
	216	+	if !c.Buffered() {
	217	+	t.Fatal("a sticky error must not make a poll loop wait")
	218	+	}
	219	+	}
	220	+
	221	+	// waitReadable blocks until the socket of c is readable, as a poll loop does.
	222	+	func waitReadable(t testing.T, c Conn) {
	223	+	t.Helper()
	224	+	rc, err := c.SyscallConn()
	225	+	if err != nil {
	226	+	t.Fatal(err)
	227	+	}
	228	+	var fd int
	229	+	if err := rc.Control(func(f uintptr) { fd = int(f) }); err != nil {
	230	+	t.Fatal(err)
	231	+	}
	232	+	if err := waitFd(fd, 5*time.Second); errors.Is(err, errNoWaitFd) {
	233	+	t.Skip(err)
	234	+	} else if err != nil {
	235	+	t.Fatal(err)
	236	+	}
	237	+	}
	238	+
	239	+	func TestSignalAllocations(t *testing.T) {
	240	+	c := dial(t, privateBus(t))
	241	+	send := func() {
	242	+	e := c.NewSignalTo(":1.1", "/a", "org.example.Alloc", "Hit", "us")
	243	+	e.Uint32(1)
	244	+	e.Str("x")
	245	+	if _, err := c.Send(); err != nil {
	246	+	t.Fatal(err)
	247	+	}
	248	+	}
	249	+	send() // grow the buffers once
	250	+	if allocs := testing.AllocsPerRun(200, send); allocs != 0 {
	251	+	t.Fatalf("%v allocations per directed signal, want 0", allocs)
	252	+	}
	253	+	}
	254	+
	255	+	func TestMethodCallAllocations(t *testing.T) {
	256	+	c := dial(t, privateBus(t))
	257	+	var serial uint32
	258	+	call := func() {
	259	+	e, s := c.NewMethodCall(busName, busPath, busName, "GetConnectionUnixProcessID", "s")
	260	+	e.Str(c.UniqueName())
	261	+	sent, err := c.Send()
	262	+	if err != nil \|\| sent != s {
	263	+	t.Fatalf("Send = %d, %v; want %d", sent, err, s)
	264	+	}
	265	+	serial = s
	266	+	}
	267	+	call()
	268	+	if allocs := testing.AllocsPerRun(200, call); allocs != 0 {
	269	+	t.Fatalf("%v allocations per method call, want 0", allocs)
	270	+	}
	271	+	// The replies must be read for the test to leave the bus quiet; they
	272	+	// also prove the calls were well formed.
	273	+	if err := c.c.SetReadDeadline(time.Now().Add(5 * time.Second)); err != nil {
	274	+	t.Fatal(err)
	275	+	}
	276	+	got := 0
	277	+	for got < 202 {
	278	+	m, err := c.ReadMessage()
	279	+	if err != nil {
	280	+	t.Fatalf("after %d of 202 replies: %v", got, err)
	281	+	}
	282	+	if m.Type == TypeMethodReturn && m.ReplySerial <= serial {
	283	+	got++
	284	+	}
	285	+	}
	286	+	}
	287	+
	288	+	func TestReadMessageAllocations(t *testing.T) {
	289	+	addr := privateBus(t)
	290	+	t.Run("method call", func(t *testing.T) {
	291	+	client, server := dial(t, addr), dial(t, addr)
	292	+	ping := func() {
	293	+	e := client.NewCall(server.UniqueName(), "/a", "org.example.Alloc", "Ping", "s")
	294	+	e.SetFlags(FlagNoReplyExpected)
	295	+	e.Str("hello")
	296	+	if _, err := client.Send(); err != nil {
	297	+	t.Fatal(err)
	298	+	}
	299	+	for {
	300	+	m, err := server.ReadMessage()
	301	+	if err != nil {
	302	+	t.Fatal(err)
	303	+	}
	304	+	if m.Type == TypeMethodCall && m.Member == "Ping" {
	305	+	if m.Body().Str() != "hello" {
	306	+	t.Fatal("bad body")
	307	+	}
	308	+	return
	309	+	}
	310	+	}
	311	+	}
	312	+	ping() // grow the buffers once
	313	+	if allocs := testing.AllocsPerRun(200, ping); allocs != 0 {
	314	+	t.Fatalf("%v allocations per method call read, want 0", allocs)
	315	+	}
	316	+	})
	317	+	t.Run("method return", func(t *testing.T) {
	318	+	c := dial(t, addr)
	319	+	ask := func() {
	320	+	e, serial := c.NewMethodCall(busName, busPath, busName, "GetNameOwner", "s")
	321	+	e.Str(busName)
	322	+	if _, err := c.Send(); err != nil {
	323	+	t.Fatal(err)
	324	+	}
	325	+	for {
	326	+	m, err := c.ReadMessage()
	327	+	if err != nil {
	328	+	t.Fatal(err)
	329	+	}
	330	+	if m.Type == TypeMethodReturn && m.ReplySerial == serial {
	331	+	if m.Body().Str() == "" {
	332	+	t.Fatal("empty owner")
	333	+	}
	334	+	return
	335	+	}
	336	+	}
	337	+	}
	338	+	ask()
	339	+	if allocs := testing.AllocsPerRun(200, ask); allocs != 0 {
	340	+	t.Fatalf("%v allocations per method return read, want 0", allocs)
	341	+	}
	342	+	})
	343	+	}

A waitfd_linux_test.go +26 -0

@@ -0,0 +1,26 @@
	1	+	package zerobus
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"fmt"
	6	+	"syscall"
	7	+	"time"
	8	+	"unsafe"
	9	+	)
	10	+
	11	+	// errNoWaitFd is returned where waitFd is not supported; never on Linux.
	12	+	var errNoWaitFd = errors.New("waitFd: not supported on this platform")
	13	+
	14	+	// waitFd blocks until fd is readable or the timeout passes, as a poll loop
	15	+	// does.
	16	+	func waitFd(fd int, timeout time.Duration) error {
	17	+	var set syscall.FdSet
	18	+	bits := int(unsafe.Sizeof(set.Bits[0])) * 8
	19	+	set.Bits[fd/bits] \|= 1 << (uint(fd) % uint(bits))
	20	+	tv := syscall.NsecToTimeval(timeout.Nanoseconds())
	21	+	n, err := syscall.Select(fd+1, &set, nil, nil, &tv)
	22	+	if err != nil \|\| n != 1 {
	23	+	return fmt.Errorf("select = %d, %v", n, err)
	24	+	}
	25	+	return nil
	26	+	}

@@ -0,0 +1,26 @@
	1	+	package zerobus
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"fmt"
	6	+	"syscall"
	7	+	"time"
	8	+	"unsafe"
	9	+	)
	10	+
	11	+	// errNoWaitFd is returned where waitFd is not supported; never on Linux.
	12	+	var errNoWaitFd = errors.New("waitFd: not supported on this platform")
	13	+
	14	+	// waitFd blocks until fd is readable or the timeout passes, as a poll loop
	15	+	// does.
	16	+	func waitFd(fd int, timeout time.Duration) error {
	17	+	var set syscall.FdSet
	18	+	bits := int(unsafe.Sizeof(set.Bits[0])) * 8
	19	+	set.Bits[fd/bits] \|= 1 << (uint(fd) % uint(bits))
	20	+	tv := syscall.NsecToTimeval(timeout.Nanoseconds())
	21	+	n, err := syscall.Select(fd+1, &set, nil, nil, &tv)
	22	+	if err != nil \|\| n != 1 {
	23	+	return fmt.Errorf("select = %d, %v", n, err)
	24	+	}
	25	+	return nil
	26	+	}

A waitfd_other_test.go +14 -0

@@ -0,0 +1,14 @@
	1	+	//go:build !linux
	2	+
	3	+	package zerobus
	4	+
	5	+	import (
	6	+	"errors"
	7	+	"time"
	8	+	)
	9	+
	10	+	// errNoWaitFd means the platform has no waitFd: syscall.FdSet differs
	11	+	// between the BSDs and Darwin, and the poll tests only run on Linux.
	12	+	var errNoWaitFd = errors.New("waitFd: not supported on this platform")
	13	+
	14	+	func waitFd(int, time.Duration) error { return errNoWaitFd }

@@ -0,0 +1,14 @@
	1	+	//go:build !linux
	2	+
	3	+	package zerobus
	4	+
	5	+	import (
	6	+	"errors"
	7	+	"time"
	8	+	)
	9	+
	10	+	// errNoWaitFd means the platform has no waitFd: syscall.FdSet differs
	11	+	// between the BSDs and Darwin, and the poll tests only run on Linux.
	12	+	var errNoWaitFd = errors.New("waitFd: not supported on this platform")
	13	+
	14	+	func waitFd(int, time.Duration) error { return errNoWaitFd }