class_name KeetUdsClient
extends Node
## Unix Domain Socket client for keet-daemon.
## Uses framed-stream protocol (4-byte LE length prefix + JSON).
## Drop-in replacement for KeetClient with push event support.
##
## LOOPBACK TCP when the launcher says so (HYPERTPS_DAEMON_TCP = "127.0.0.1:<port>",
## HYPERTPS_DAEMON_TOKEN): always on Windows, where StreamPeerUDS cannot work --
## NetSocketWinSock::open rejects Family::UNIX. The daemon closes a TCP connection
## whose first frame is not {"auth": <token>}, so that frame goes out before
## anything else, every connect.
signal connected
signal disconnected
signal error_occurred(message: String)
signal status_updated(status: Dictionary)
signal event_received(event_name: String, data: Dictionary)
@export var socket_path: String = "/tmp/keet-daemon.sock"
@export var auto_connect: bool = true
@export_range(1.0, 30.0) var reconnect_delay: float = 5.0
@export_range(1.0, 120.0) var max_reconnect_delay: float = 30.0
## How long a call may go unanswered before its callback fails with "Timed out".
## Without it, a daemon that accepts a call and never replies (getIdentity waiting
## on keet-core's boot, a StateBase open stuck on a missing block) left the caller
## waiting for ever with no error. Generous on purpose: StateBase admission can
## legitimately take ~20 s. Override per call with call_rpc's timeout_sec.
@export_range(0.1, 600.0) var rpc_timeout_sec: float = 30.0
var is_connected: bool = false
var _current_reconnect_delay: float = 0.0
var _peer: StreamPeerSocket # StreamPeerUDS, or StreamPeerTCP (see header)
var _tcp_host: String = ""
var _tcp_port: int = 0
var _tcp_token: String = ""
var _reconnect_timer: Timer
var _next_id: int = 1
var _pending: Dictionary = {} # id -> {callback: Callable, method: String, deadline_ms: int}
var _recv_buf: PackedByteArray = PackedByteArray()
var _frame_len: int = -1 # -1 = waiting for header
func _ready() -> void:
var tcp := OS.get_environment("HYPERTPS_DAEMON_TCP")
if tcp != "":
var at := tcp.rfind(":")
_tcp_host = tcp.substr(0, at)
_tcp_port = tcp.substr(at + 1).to_int()
_tcp_token = OS.get_environment("HYPERTPS_DAEMON_TOKEN")
print("[KeetUdsClient] daemon over TCP %s:%d" % [_tcp_host, _tcp_port])
_peer = _new_peer()
_reconnect_timer = Timer.new()
_reconnect_timer.wait_time = reconnect_delay
_reconnect_timer.one_shot = true
_reconnect_timer.timeout.connect(_try_connect)
add_child(_reconnect_timer)
if auto_connect:
_try_connect()
func _process(_delta: float) -> void:
if _peer == null:
return
_expire_pending(Time.get_ticks_msec())
_peer.poll()
var status := _peer.get_status()
match status:
StreamPeerSocket.STATUS_CONNECTED:
if not is_connected:
is_connected = true
_current_reconnect_delay = 0.0
if _tcp_port > 0:
_send_frame(JSON.stringify({"auth": _tcp_token})) # before ANY rpc
connected.emit()
_check_status()
_read_available()
StreamPeerSocket.STATUS_CONNECTING:
pass # Still connecting
StreamPeerSocket.STATUS_NONE, StreamPeerSocket.STATUS_ERROR:
if is_connected:
is_connected = false
_fail_pending("Disconnected")
disconnected.emit()
_schedule_reconnect()
## Make a JSON-RPC call to keet-daemon.
## callback(result, error) is called exactly once: when the response arrives, on
## disconnect, or with "Timed out after …" once timeout_sec (default
## rpc_timeout_sec) passes unanswered. A reply arriving after the timeout is dropped.
func call_rpc(method: String, params: Dictionary = {}, callback: Callable = Callable(), timeout_sec: float = -1.0) -> void:
if not is_connected:
if callback.is_valid():
callback.call(null, "Not connected")
return
var id := _next_id
_next_id += 1
if callback.is_valid():
var t := timeout_sec if timeout_sec > 0.0 else rpc_timeout_sec
_pending[id] = {"callback": callback, "method": method,
"deadline_ms": Time.get_ticks_msec() + int(t * 1000.0)}
var msg := {"id": id, "method": method, "params": params}
_send_frame(JSON.stringify(msg))
## Async RPC via await. Returns {"result": ..., "error": ...}.
func rpc_async(method: String, params: Dictionary = {}) -> Dictionary:
var response := {"result": null, "error": null}
var done := false
call_rpc(method, params, func(result, err):
response.result = result
response.error = err
done = true
)
while not done:
await get_tree().process_frame
return response
## Manually trigger a connection attempt.
func try_connect() -> void:
_try_connect()
func _try_connect() -> void:
if is_connected:
return
_reconnect_timer.stop()
_recv_buf = PackedByteArray()
_frame_len = -1
# NOTE: do NOT gate on FileAccess.file_exists(socket_path) — Godot's
# FileAccess returns false for Unix domain sockets (they're S_IFSOCK,
# not regular files), so the guard skips connect forever and the chat
# stays "connecting" indefinitely. If the socket is missing, connect
# will fail with err != OK and we'll _schedule_reconnect anyway.
# Create fresh peer for each attempt
_peer = _new_peer()
var err: Error
if _tcp_port > 0:
err = (_peer as StreamPeerTCP).connect_to_host(_tcp_host, _tcp_port)
else:
err = (_peer as StreamPeerUDS).connect_to_host(socket_path)
if err != OK:
_schedule_reconnect()
func _new_peer() -> StreamPeerSocket:
var peer: StreamPeerSocket = StreamPeerTCP.new() if _tcp_port > 0 else StreamPeerUDS.new()
peer.set_big_endian(false)
return peer
func _schedule_reconnect() -> void:
if _reconnect_timer.is_stopped() and not is_connected:
if _current_reconnect_delay == 0.0:
_current_reconnect_delay = reconnect_delay
else:
_current_reconnect_delay = minf(_current_reconnect_delay * 1.5, max_reconnect_delay)
_reconnect_timer.wait_time = _current_reconnect_delay
_reconnect_timer.start()
func _check_status() -> void:
call_rpc("status", {}, func(result, err):
if err == null and result is Dictionary:
status_updated.emit(result)
)
func _send_frame(json_str: String) -> void:
var data := json_str.to_utf8_buffer()
var length := data.size()
# 4-byte little-endian length prefix
var header := PackedByteArray()
header.resize(4)
header[0] = length & 0xFF
header[1] = (length >> 8) & 0xFF
header[2] = (length >> 16) & 0xFF
header[3] = (length >> 24) & 0xFF
var frame := header + data
_peer.put_data(frame)
func _read_available() -> void:
var available := _peer.get_available_bytes()
if available <= 0:
return
var result := _peer.get_data(available)
if result[0] != OK:
return
_recv_buf += result[1]
_parse_frames()
func _parse_frames() -> void:
while true:
# Need at least 4 bytes for the length header
if _frame_len < 0:
if _recv_buf.size() < 4:
break
_frame_len = _recv_buf[0] | (_recv_buf[1] << 8) | (_recv_buf[2] << 16) | (_recv_buf[3] << 24)
_recv_buf = _recv_buf.slice(4)
# Wait until we have the full frame
if _recv_buf.size() < _frame_len:
break
# Extract the frame
var frame_data := _recv_buf.slice(0, _frame_len)
_recv_buf = _recv_buf.slice(_frame_len)
_frame_len = -1
var json_str := frame_data.get_string_from_utf8()
var parsed: Variant = JSON.parse_string(json_str)
if parsed is Dictionary:
_on_message(parsed)
func _on_message(msg: Dictionary) -> void:
# Push event (server-initiated, no id field)
if msg.has("event"):
var event_name: String = msg.get("event", "")
var data: Dictionary = msg.get("data", {}) if msg.get("data") is Dictionary else {}
var room_id: Variant = msg.get("roomId", "")
if room_id is String and room_id != "":
data["roomId"] = room_id
event_received.emit(event_name, data)
return
# RPC response (has id field)
var id: Variant = msg.get("id", null)
if id == null:
return
# id may come back as float from JSON parsing
var id_int: int = int(id)
if not _pending.has(id_int):
return
var entry: Dictionary = _pending[id_int]
_pending.erase(id_int)
var callback: Callable = entry.get("callback", Callable())
if not callback.is_valid():
return
var ok: bool = msg.get("ok", false)
if ok:
callback.call(msg.get("data"), null)
else:
callback.call(null, msg.get("error", "Unknown error"))
## Fail every call whose deadline has passed. Erased BEFORE its callback runs, so
## a callback that issues a new call cannot see or re-expire itself.
func _expire_pending(now_ms: int) -> void:
var expired: Array[int] = []
for id: int in _pending.keys():
if now_ms >= int(_pending[id].get("deadline_ms", 0x7FFFFFFFFFFFFFFF)):
expired.append(id)
for id: int in expired:
var entry: Dictionary = _pending[id]
_pending.erase(id)
var callback: Callable = entry.get("callback", Callable())
if callback.is_valid():
callback.call(null, "Timed out waiting for keet-daemon: %s" % String(entry.get("method", "?")))
func _fail_pending(reason: String) -> void:
for id: int in _pending.keys():
var entry: Dictionary = _pending[id]
var callback: Callable = entry.get("callback", Callable())
if callback.is_valid():
callback.call(null, reason)
_pending.clear()