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.
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
var is_connected: bool = false
var _peer: StreamPeerUDS
var _reconnect_timer: Timer
var _next_id: int = 1
var _pending: Dictionary = {} # id -> {callback: Callable, method: String}
var _recv_buf: PackedByteArray = PackedByteArray()
var _frame_len: int = -1 # -1 = waiting for header
func _ready() -> void:
_peer = StreamPeerUDS.new()
_peer.set_big_endian(false)
_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
_peer.poll()
var status := _peer.get_status()
match status:
StreamPeerUDS.STATUS_CONNECTED:
if not is_connected:
is_connected = true
connected.emit()
_check_status()
_read_available()
StreamPeerUDS.STATUS_CONNECTING:
pass # Still connecting
StreamPeerUDS.STATUS_NONE, StreamPeerUDS.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 when the response arrives.
func call_rpc(method: String, params: Dictionary = {}, callback: Callable = Callable()) -> 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():
_pending[id] = {"callback": callback, "method": method}
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
# Create fresh peer for each attempt
_peer = StreamPeerUDS.new()
_peer.set_big_endian(false)
var err := _peer.connect_to_host(socket_path)
if err != OK:
_schedule_reconnect()
func _schedule_reconnect() -> void:
if _reconnect_timer.is_stopped() and not is_connected:
_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"))
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()