Derive transport registries from one list

The five transport/address lookup maps each hand-guarded `uds`
with its own `if HAS_UDS:` block (3 in `ipc/_types`, 2 in
`discovery/_addr`) — easy to let drift so a backend half-registers
(known by address but not by key, listed but unroutable, &c).

- build one `_msg_transports` / `_address_protos` list per module
  (TCP always, UDS only when `HAS_UDS`), then DERIVE every map
  from it via each backend's ClassVars (`codec_key`,
  `address_type`, `proto_key`). Adding a backend now touches one
  list, and the maps can't disagree.

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
windows_support_round2
Gud Boi 2026-06-30 02:46:00 -04:00
parent 7ae8ddfb2c
commit d89ece08e4
2 changed files with 42 additions and 21 deletions

View File

@ -172,24 +172,36 @@ class Address(Protocol):
...
_address_types: bidict[str, Type[Address]] = {
'tcp': TCPAddress,
}
# the address types available on this host: TCP always, UDS only
# where usable (`HAS_UDS`). Both registries derive from this single
# list via each type's `proto_key`.
_address_protos: list[Type[Address]] = [TCPAddress]
if HAS_UDS:
_address_types['uds'] = UDSAddress
_address_protos.append(UDSAddress)
_address_types: bidict[str, Type[Address]] = bidict({
cls.proto_key: cls
for cls in _address_protos
})
# TODO! really these are discovery sys default addrs ONLY useful for
# when none is provided to a root actor on first boot.
_default_lo_addrs: dict[str, UnwrappedAddress] = {
'tcp': TCPAddress.get_root().unwrap(),
cls.proto_key: cls.get_root().unwrap()
for cls in _address_protos
}
if HAS_UDS:
_default_lo_addrs['uds'] = UDSAddress.get_root().unwrap()
def get_address_cls(name: str) -> Type[Address]:
return _address_types[name]
try:
return _address_types[name]
except KeyError:
raise NotImplementedError(
f'No IPC transport backend for {name!r} on this '
f'platform!\n'
f'(available: {list(_address_types)})\n'
)
def is_wrapped_addr(addr: any) -> bool:
@ -287,7 +299,14 @@ def default_lo_addrs(
for an input transport key set.
'''
return [
_default_lo_addrs[transport]
for transport in transports
]
lo_addrs: list[UnwrappedAddress] = []
for transport in transports:
try:
lo_addrs.append(_default_lo_addrs[transport])
except KeyError:
raise NotImplementedError(
f'No default loopback addr for transport '
f'{transport!r} on this platform!\n'
f'(available: {list(_default_lo_addrs)})\n'
)
return lo_addrs

View File

@ -41,26 +41,28 @@ from tractor.ipc._uds import (
# hosts `HAS_UDS` is `False` and the runtime registers TCP only.
Address = TCPAddress|UDSAddress
# manually updated list of all supported msg transport types
# the available msg-transport backends on this host: TCP always,
# UDS only where usable (`HAS_UDS`). The lookup maps below are all
# DERIVED from this single list via each backend's ClassVars
# (`codec_key`, `address_type`) — register a backend here and every
# map picks it up; no per-map `if HAS_UDS` to keep in sync.
_msg_transports: list[Type[MsgTransport]] = [
MsgpackTCPStream,
]
if HAS_UDS:
_msg_transports.append(MsgpackUDSStream)
# map a `MsgTransportKey` to its `MsgTransport` type
# map a `MsgTransportKey` -> `MsgTransport` type
_key_to_transport: dict[MsgTransportKey, Type[MsgTransport]] = {
('msgpack', 'tcp'): MsgpackTCPStream,
(t.codec_key, t.address_type.proto_key): t
for t in _msg_transports
}
if HAS_UDS:
_key_to_transport[('msgpack', 'uds')] = MsgpackUDSStream
# map an `Address`-wrapper to its `MsgTransport` type
# map an `Address`-wrapper -> `MsgTransport` type
_addr_to_transport: dict[Type[Address], Type[MsgTransport]] = {
TCPAddress: MsgpackTCPStream,
t.address_type: t
for t in _msg_transports
}
if HAS_UDS:
_addr_to_transport[UDSAddress] = MsgpackUDSStream
# ------------------------------------------------------------