internal/pipewire: implement client context

This consumes the entire sample, is validated to send identical messages and correctly handle received messages.

Signed-off-by: Ophestra <cat@gensokyo.uk>

1 files changed, 235 insertions, 0 deletions

diff --git a/internal/pipewire/core.go b/internal/pipewire/core.go
index 4b61972b..fe7fb78e 100644
--- a/internal/pipewire/core.go
+++ b/internal/pipewire/core.go
@@ -1,5 +1,11 @@
 package pipewire
 
+import (
+	"errors"
+	"fmt"
+	"strconv"
+)
+
 /* pipewire/core.h */
 
 const (
@@ -240,6 +246,37 @@ func (c *CoreBoundProps) MarshalBinary() ([]byte, error) { return Marshal(c) }
 // UnmarshalBinary satisfies [encoding.BinaryUnmarshaler] via [Unmarshal].
 func (c *CoreBoundProps) UnmarshalBinary(data []byte) error { return Unmarshal(data, c) }
 
+// ErrBadBoundProps is returned when a [CoreBoundProps] event targeting a proxy
+// that should never be targeted is received and processed.
+var ErrBadBoundProps = errors.New("attempted to store bound props on proxy that should never be targeted")
+
+// noAck is embedded by proxies that are never targeted by [CoreBoundProps].
+type noAck struct{}
+
+// setBoundProps should never be called as this proxy should never be targeted by [CoreBoundProps].
+func (noAck) setBoundProps(*CoreBoundProps) error { return ErrBadBoundProps }
+
+// An InconsistentIdError describes an inconsistent state where the server claims an impossible
+// proxy or global id. This is only generated by the [CoreBoundProps] event.
+type InconsistentIdError struct {
+	// Whether the inconsistent id is the global resource id.
+	Global bool
+	// Targeted proxy instance.
+	Proxy fmt.Stringer
+	// Differing ids.
+	ID, ServerID Int
+}
+
+func (e *InconsistentIdError) Error() string {
+	name := "proxy"
+	if e.Global {
+		name = "global"
+	}
+
+	return name + " id " + strconv.Itoa(int(e.ID)) + " targeting " + e.Proxy.String() +
+		" inconsistent with " + strconv.Itoa(int(e.ServerID)) + " claimed by the server"
+}
+
 // CoreHello is the first message sent by a client.
 type CoreHello struct {
 	// The version number of the client, usually PW_VERSION_CORE.
@@ -255,6 +292,16 @@ func (c *CoreHello) MarshalBinary() ([]byte, error) { return Marshal(c) }
 // UnmarshalBinary satisfies [encoding.BinaryUnmarshaler] via [Unmarshal].
 func (c *CoreHello) UnmarshalBinary(data []byte) error { return Unmarshal(data, c) }
 
+// coreHello queues a [CoreHello] message for the PipeWire server.
+// This method should not be called directly, the New function queues this message.
+func (ctx *Context) coreHello() error {
+	return ctx.writeMessage(
+		PW_ID_CORE,
+		PW_CORE_METHOD_HELLO,
+		&CoreHello{PW_VERSION_CORE},
+	)
+}
+
 const (
 	// CoreSyncSequenceOffset is the offset to [Header.Sequence] to produce [CoreSync.Sequence].
 	CoreSyncSequenceOffset = 0x40000000
@@ -279,6 +326,35 @@ func (c *CoreSync) MarshalBinary() ([]byte, error) { return Marshal(c) }
 // UnmarshalBinary satisfies [encoding.BinaryUnmarshaler] via [Unmarshal].
 func (c *CoreSync) UnmarshalBinary(data []byte) error { return Unmarshal(data, c) }
 
+// coreSync queues a [CoreSync] message for the PipeWire server.
+// This is not safe to use directly, callers should use Sync instead.
+func (ctx *Context) coreSync(id Int) error {
+	return ctx.writeMessage(
+		PW_ID_CORE,
+		PW_CORE_METHOD_SYNC,
+		&CoreSync{id, CoreSyncSequenceOffset + Int(ctx.sequence)},
+	)
+}
+
+// ErrNotDone is returned if [Core.Sync] returns from its [Context.Roundtrip] without
+// receiving a [CoreDone] event targeting the [CoreSync] event it delivered.
+var ErrNotDone = errors.New("did not receive a Core::Done event targeting previously delivered Core::Sync")
+
+// Sync queues a [CoreSync] message for the PipeWire server and initiates a Roundtrip.
+func (core *Core) Sync() error {
+	core.done = false
+	if err := core.ctx.coreSync(roundtripSyncID); err != nil {
+		return err
+	}
+	if err := core.ctx.Roundtrip(); err != nil {
+		return err
+	}
+	if !core.done {
+		return ErrNotDone
+	}
+	return nil
+}
+
 // The CorePong message is sent from the client to the server when the server emits the Ping event.
 type CorePong struct {
 	// Copied from [CorePing.ID].
@@ -320,6 +396,19 @@ func (c *CoreGetRegistry) MarshalBinary() ([]byte, error) { return Marshal(c) }
 // UnmarshalBinary satisfies [encoding.BinaryUnmarshaler] via [Unmarshal].
 func (c *CoreGetRegistry) UnmarshalBinary(data []byte) error { return Unmarshal(data, c) }
 
+// GetRegistry queues a [CoreGetRegistry] message for the PipeWire server
+// and returns the address of the newly allocated [Registry].
+func (ctx *Context) GetRegistry() (*Registry, error) {
+	registry := Registry{Objects: make(map[Int]RegistryGlobal), ctx: ctx}
+	newId := ctx.newProxyId(&registry, false)
+	registry.ID = newId
+	return &registry, ctx.writeMessage(
+		PW_ID_CORE,
+		PW_CORE_METHOD_GET_REGISTRY,
+		&CoreGetRegistry{PW_VERSION_REGISTRY, newId},
+	)
+}
+
 // A RegistryGlobal event is emitted to notify a client about a new global object.
 type RegistryGlobal struct {
 	// The global id.
@@ -379,3 +468,149 @@ func (c *RegistryBind) MarshalBinary() ([]byte, error) { return Marshal(c) }
 
 // UnmarshalBinary satisfies [encoding.BinaryUnmarshaler] via [Unmarshal].
 func (c *RegistryBind) UnmarshalBinary(data []byte) error { return Unmarshal(data, c) }
+
+// bind queues a [RegistryBind] message for the PipeWire server
+// and returns the newly allocated proxy id.
+func (registry *Registry) bind(proxy eventProxy, id, version Int) (Int, error) {
+	bind := RegistryBind{
+		ID:      id,
+		Type:    proxy.String(),
+		Version: version,
+		NewID:   registry.ctx.newProxyId(proxy, true),
+	}
+	return bind.NewID, registry.ctx.writeMessage(
+		registry.ID,
+		PW_REGISTRY_METHOD_BIND,
+		&bind,
+	)
+}
+
+// An UnsupportedObjectTypeError is the name of a type not known by the server [Registry].
+type UnsupportedObjectTypeError string
+
+func (e UnsupportedObjectTypeError) Error() string { return "unsupported object type " + string(e) }
+
+// Core holds state of [PW_TYPE_INTERFACE_Core].
+type Core struct {
+	// Additional information from the server, populated or updated during [Context.Roundtrip].
+	Info *CoreInfo `json:"info"`
+
+	// Whether a [CoreDone] event was received during Sync.
+	done bool
+
+	ctx *Context
+	noAck
+}
+
+// ErrUnexpectedDone is a [CoreDone] event with unexpected values.
+var ErrUnexpectedDone = errors.New("multiple Core::Done events targeting Core::Sync")
+
+// An UnknownBoundIdError describes the server claiming to have bound a proxy id that was never allocated.
+type UnknownBoundIdError[E any] struct {
+	// Offending id decoded from Data.
+	Id Int
+	// Event received from the server.
+	Event E
+}
+
+func (e *UnknownBoundIdError[E]) Error() string {
+	return "unknown bound proxy id " + strconv.Itoa(int(e.Id))
+}
+
+func (core *Core) consume(opcode byte, files []int, unmarshal func(v any) error) error {
+	if err := closeReceivedFiles(files...); err != nil {
+		return err
+	}
+
+	switch opcode {
+	case PW_CORE_EVENT_INFO:
+		return unmarshal(&core.Info)
+
+	case PW_CORE_EVENT_DONE:
+		var done CoreDone
+		if err := unmarshal(&done); err != nil {
+			return err
+		}
+		if done.ID == roundtripSyncID && done.Sequence == CoreSyncSequenceOffset+core.ctx.sequence-1 {
+			if core.done {
+				return ErrUnexpectedDone
+			}
+			core.done = true
+		}
+
+		// silently ignore non-matching events because the server sends out
+		// an event with id -1 seq 0 that does not appear to correspond to
+		// anything, and this behaviour is never mentioned in documentation
+		return nil
+
+	case PW_CORE_EVENT_BOUND_PROPS:
+		var boundProps CoreBoundProps
+		if err := unmarshal(&boundProps); err != nil {
+			return err
+		}
+
+		delete(core.ctx.pendingIds, boundProps.ID)
+		proxy, ok := core.ctx.proxy[boundProps.ID]
+		if !ok {
+			return &UnknownBoundIdError[*CoreBoundProps]{Id: boundProps.ID, Event: &boundProps}
+		}
+		return proxy.setBoundProps(&boundProps)
+
+	default:
+		return &UnsupportedOpcodeError{opcode, core.String()}
+	}
+}
+
+func (core *Core) String() string { return PW_TYPE_INTERFACE_Core }
+
+// Registry holds state of [PW_TYPE_INTERFACE_Registry].
+type Registry struct {
+	// Proxy id as tracked by [Context].
+	ID Int `json:"proxy_id"`
+
+	// Global objects received via the [RegistryGlobal] event.
+	//
+	// This requires more processing before it can be used, but is not implemented
+	// as it is not used by Hakurei.
+	Objects map[Int]RegistryGlobal `json:"objects"`
+
+	ctx *Context
+	noAck
+}
+
+// A GlobalIDCollisionError describes a [RegistryGlobal] event stepping on a previous instance of itself.
+type GlobalIDCollisionError struct {
+	// The colliding id.
+	ID Int
+	// Involved events.
+	Previous, Current *RegistryGlobal
+}
+
+func (e *GlobalIDCollisionError) Error() string {
+	return "new Registry::Global event for " + e.Current.Type +
+		" stepping on previous id " + strconv.Itoa(int(e.ID)) + " for " + e.Previous.Type
+}
+
+func (registry *Registry) consume(opcode byte, files []int, unmarshal func(v any) error) error {
+	if err := closeReceivedFiles(files...); err != nil {
+		return err
+	}
+
+	switch opcode {
+	case PW_REGISTRY_EVENT_GLOBAL:
+		var global RegistryGlobal
+		if err := unmarshal(&global); err != nil {
+			return err
+		}
+		if object, ok := registry.Objects[global.ID]; ok {
+			return &GlobalIDCollisionError{global.ID, &object, &global}
+		}
+		registry.Objects[global.ID] = global
+		return nil
+
+	default:
+		return &UnsupportedOpcodeError{opcode, registry.String()}
+	}
+}
+
+func (registry *Registry) String() string { return PW_TYPE_INTERFACE_Registry }