""" DASP (Datagram Authenticated Session Protocol) transport layer. Ported from sedonaOntrol/src/sedona/dasp/ Java sources. Provides reliable, authenticated, full-duplex UDP communication as the transport layer beneath the Sox application protocol. """ import hashlib import os import queue import random import socket import struct import threading import time # ──────────────────────────────────────────────────────────────── # Constants (from DaspConst.java) # ──────────────────────────────────────────────────────────────── VERSION_VAL = 0x0100 IDEAL_MAX_VAL = 512 ABS_MAX_VAL = 1024 RECEIVE_MAX_VAL = 31 RECEIVE_TIMEOUT_VAL = 30_000 # ms CONNECT_TIMEOUT_VAL = 10_000 # ms SEND_RETRY_VAL = 1000 # ms MAX_RETRY_BACKOFF = 8000 # ms (Ontrol) ACK_MORE_BYTE_LIMIT = 7 # (Ontrol) DIGEST_ALGORITHM = "SHA-1" # (Ontrol: also "SHA-256") IDEAL_MAX_DEF = 512 ABS_MAX_DEF = 512 RECEIVE_MAX_DEF = 31 RECEIVE_TIMEOUT_DEF = 30_000 # ms # Message types DISCOVER = 0 HELLO = 1 CHALLENGE = 2 AUTHENTICATE = 3 WELCOME = 4 KEEPALIVE = 5 DATAGRAM = 6 CLOSE = 7 # Header field value encodings (low 2 bits of field id) NIL = 0 U2 = 1 STR = 2 BYTES = 3 # Header field ids HDR_VERSION = 0x05 HDR_REMOTE_ID = 0x09 HDR_DIGEST_ALGORITHM = 0x0E HDR_NONCE = 0x13 HDR_USERNAME = 0x16 HDR_DIGEST = 0x1B HDR_IDEAL_MAX = 0x1D HDR_ABS_MAX = 0x21 HDR_ACK = 0x25 HDR_ACK_MORE = 0x2B HDR_RECEIVE_MAX = 0x2D HDR_RECEIVE_TIMEOUT = 0x31 HDR_ERROR_CODE = 0x35 HDR_PLATFORM_ID = 0x3A HDR_MESSAGE_LEN = 0x81 # (Ontrol) # Error codes ERR_INCOMPATIBLE_VERSION = 0xE1 ERR_BUSY = 0xE2 ERR_DIGEST_NOT_SUPPORTED = 0xE3 ERR_NOT_AUTHENTICATED = 0xE4 ERR_TIMEOUT = 0xE5 DEFAULT_PORT = 1876 _MSG_TYPE_NAMES = { DISCOVER: "DISCOVER", HELLO: "HELLO", CHALLENGE: "CHALLENGE", AUTHENTICATE: "AUTHENTICATE", WELCOME: "WELCOME", KEEPALIVE: "KEEPALIVE", DATAGRAM: "DATAGRAM", CLOSE: "CLOSE", } def _trace(direction, msg, datagram=None): import app_state if not app_state.TRACE_DASP: return import datetime ts = datetime.datetime.now().strftime("%H:%M:%S.%f")[:-3] name = _MSG_TYPE_NAMES.get(msg.msg_type, f"?{msg.msg_type}") seq = f" seq={msg.seq_num}" if msg.seq_num >= 0 else "" ack = f" ack={msg.ack}" if msg.ack >= 0 else "" plen = len(msg.payload) if msg.payload else 0 hexdump = "" if datagram: hexdump = " [" + datagram.hex(" ") + "]" print(f"[DASP] {ts} {direction} {name}{seq}{ack} payload={plen}{hexdump}") class DaspException(Exception): def __init__(self, msg, error_code=-1): super().__init__(msg) self.error_code = error_code # ──────────────────────────────────────────────────────────────── # DaspMsg (from DaspMsg.java) # ──────────────────────────────────────────────────────────────── class DaspMsg: """Binary encoder/decoder for DASP messages.""" def __init__(self): self.session_id = -1 self.msg_type = -1 self.seq_num = -1 # headers self.version = -1 self.remote_id = -1 self.digest_algorithm = None self.nonce = None self.username = None self.digest = None self.ideal_max = -1 self.abs_max = -1 self.ack = -1 self.ack_more = None self.receive_max = -1 self.receive_timeout = -1 # ms internally self.error_code = -1 self.platform_id = None self.include_message_len = True # (Ontrol) # payload self.payload = b"" # ── Decode ────────────────────────────────────────────────── @staticmethod def decode(buf, length=None): if length is None: length = len(buf) msg = DaspMsg() msg._decode(buf, length) return msg def _decode(self, buf, length): self.session_id = (buf[0] << 8) | buf[1] self.seq_num = (buf[2] << 8) | buf[3] self.msg_type = (buf[4] & 0xFF) >> 4 num_fields = buf[4] & 0x0F pos = 5 for _ in range(num_fields): fid = buf[pos] & 0xFF pos += 1 ftype = fid & 0x03 u2_val = 0 str_val = None bytes_val = None if ftype == NIL: pass elif ftype == U2: u2_val = (buf[pos] << 8) | buf[pos + 1] pos += 2 elif ftype == STR: start = pos while buf[pos] != 0: pos += 1 str_val = buf[start:pos].decode("utf-8") pos += 1 # skip null terminator elif ftype == BYTES: blen = buf[pos] & 0xFF pos += 1 bytes_val = bytes(buf[pos:pos + blen]) pos += blen if fid == HDR_VERSION: self.version = u2_val elif fid == HDR_REMOTE_ID: self.remote_id = u2_val elif fid == HDR_DIGEST_ALGORITHM: self.digest_algorithm = str_val elif fid == HDR_NONCE: self.nonce = bytes_val elif fid == HDR_USERNAME: self.username = str_val elif fid == HDR_DIGEST: self.digest = bytes_val elif fid == HDR_IDEAL_MAX: self.ideal_max = u2_val elif fid == HDR_ABS_MAX: self.abs_max = u2_val elif fid == HDR_ACK: self.ack = u2_val elif fid == HDR_ACK_MORE: self.ack_more = bytes_val elif fid == HDR_RECEIVE_MAX: self.receive_max = u2_val elif fid == HDR_RECEIVE_TIMEOUT: self.receive_timeout = u2_val * 1000 elif fid == HDR_ERROR_CODE: self.error_code = u2_val elif fid == HDR_PLATFORM_ID: self.platform_id = str_val elif fid == HDR_MESSAGE_LEN: pass # not used after decode self.payload = bytes(buf[pos:length]) # ── Encode ────────────────────────────────────────────────── def encode(self): buf = bytearray(ABS_MAX_VAL) length = self._encode_into(buf) return bytes(buf[:length]) def _encode_into(self, buf): struct.pack_into(">HH", buf, 0, self.session_id & 0xFFFF, self.seq_num & 0xFFFF) pos = 5 # skip byte 4, will backpatch has_ideal_max = self.ideal_max > -1 and self.ideal_max != IDEAL_MAX_DEF has_abs_max = self.abs_max > -1 and self.abs_max != ABS_MAX_DEF has_receive_max = self.receive_max > -1 and self.receive_max != RECEIVE_MAX_DEF has_receive_timeout = self.receive_timeout > -1 and self.receive_timeout != RECEIVE_TIMEOUT_DEF num = 0 # Ontrol: MESSAGE_LEN first (backpatched) msg_len_pos = -1 if self.include_message_len: num += 1 buf[pos] = HDR_MESSAGE_LEN pos += 1 msg_len_pos = pos pos += 2 # placeholder for length if self.version > -1: num += 1; buf[pos] = HDR_VERSION; pos += 1 pos = _put_u2(buf, pos, self.version) if self.remote_id > -1: num += 1; buf[pos] = HDR_REMOTE_ID; pos += 1 pos = _put_u2(buf, pos, self.remote_id) if self.digest_algorithm is not None: num += 1; buf[pos] = HDR_DIGEST_ALGORITHM; pos += 1 pos = _put_str(buf, pos, self.digest_algorithm) if self.nonce is not None: num += 1; buf[pos] = HDR_NONCE; pos += 1 pos = _put_bytes(buf, pos, self.nonce) if self.username is not None: num += 1; buf[pos] = HDR_USERNAME; pos += 1 pos = _put_str(buf, pos, self.username) if self.digest is not None: num += 1; buf[pos] = HDR_DIGEST; pos += 1 pos = _put_bytes(buf, pos, self.digest) if has_ideal_max: num += 1; buf[pos] = HDR_IDEAL_MAX; pos += 1 pos = _put_u2(buf, pos, self.ideal_max) if has_abs_max: num += 1; buf[pos] = HDR_ABS_MAX; pos += 1 pos = _put_u2(buf, pos, self.abs_max) if self.ack > -1: num += 1; buf[pos] = HDR_ACK; pos += 1 pos = _put_u2(buf, pos, self.ack) if self.ack_more is not None: num += 1; buf[pos] = HDR_ACK_MORE; pos += 1 pos = _put_bytes(buf, pos, self.ack_more) if has_receive_max: num += 1; buf[pos] = HDR_RECEIVE_MAX; pos += 1 pos = _put_u2(buf, pos, self.receive_max) if has_receive_timeout: num += 1; buf[pos] = HDR_RECEIVE_TIMEOUT; pos += 1 pos = _put_u2(buf, pos, int(self.receive_timeout / 1000)) if self.error_code > -1: num += 1; buf[pos] = HDR_ERROR_CODE; pos += 1 pos = _put_u2(buf, pos, self.error_code) if self.platform_id is not None: num += 1; buf[pos] = HDR_PLATFORM_ID; pos += 1 pos = _put_str(buf, pos, self.platform_id) # backpatch msg_type | num_fields buf[4] = ((self.msg_type & 0x0F) << 4) | (num & 0x0F) # payload if self.payload: buf[pos:pos + len(self.payload)] = self.payload pos += len(self.payload) # backpatch MESSAGE_LEN if msg_len_pos >= 0: _put_u2(buf, msg_len_pos, pos) return pos def _put_u2(buf, pos, val): buf[pos] = (val >> 8) & 0xFF buf[pos + 1] = val & 0xFF return pos + 2 def _put_str(buf, pos, val): encoded = val.encode("ascii") buf[pos:pos + len(encoded)] = encoded pos += len(encoded) buf[pos] = 0 return pos + 1 def _put_bytes(buf, pos, val): buf[pos] = len(val) & 0xFF pos += 1 buf[pos:pos + len(val)] = val return pos + len(val) # ──────────────────────────────────────────────────────────────── # ReceiveWindow (from ReceiveWindow.java) # ──────────────────────────────────────────────────────────────── class ReceiveWindow: """Manages incoming message sequence numbers and acknowledgements.""" MAX = RECEIVE_MAX_VAL # 31 def __init__(self, session): self._session = session self._lock = threading.Lock() self._seq_num = -1 self._seq_more = -1 self._acked = False def unacked(self): with self._lock: return not self._acked def init(self, num): with self._lock: if self._seq_num >= 0: raise RuntimeError("ReceiveWindow already initialized") self._seq_num = (num - 1) & 0xFFFF self._seq_more = 1 self._acked = False def set_ack_headers(self, msg): with self._lock: payload_len = len(msg.payload) # room for 5-byte fixed header + 3-byte ack header = 8 if payload_len + 8 <= self._session.ideal_max: msg.ack = self._seq_num self._acked = True # room for ack_more: +3 bytes = 11 total overhead if self._seq_more > 0x01 and payload_len + 11 <= self._session.ideal_max: max_bytes = self._session.ideal_max - payload_len - 10 msg.ack_more = _to_ack_more(self._seq_more, max_bytes) def receive(self, num): with self._lock: if num < 0 or num > 0xFFFF: raise RuntimeError(f"Invalid seq num: {num}") if self._seq_num < 0: raise RuntimeError("ReceiveWindow not initialized") start = self._seq_num end = start + self.MAX if end <= 0xFFFF: if num <= start or num > end: self._acked = False return False else: end -= 0xFFFF if num <= start and num > end: self._acked = False return False # next in sequence — slide window if num == (start + 1) & 0xFFFF: while True: self._seq_num = (self._seq_num + 1) & 0xFFFF self._seq_more = (self._seq_more >> 1) & 0x7FFFFFFF if (self._seq_more & 0x02) == 0: break self._seq_more |= 0x01 else: # out of order bit = 1 << ((num - start) & 0xFFFF) if self._seq_more & bit: return False self._seq_more |= bit self._acked = False return True def _to_ack_more(seq_more, max_bytes): if seq_more <= 0xFF or max_bytes <= 1: return bytes([seq_more & 0xFF]) if seq_more <= 0xFFFF or max_bytes <= 2: return bytes([(seq_more >> 8) & 0xFF, seq_more & 0xFF]) if seq_more <= 0xFFFFFF or max_bytes <= 3: return bytes([(seq_more >> 16) & 0xFF, (seq_more >> 8) & 0xFF, seq_more & 0xFF]) return bytes([(seq_more >> 24) & 0xFF, (seq_more >> 16) & 0xFF, (seq_more >> 8) & 0xFF, seq_more & 0xFF]) # ──────────────────────────────────────────────────────────────── # SendWindow (from SendWindow.java) # ──────────────────────────────────────────────────────────────── class _Packet: __slots__ = ("seq_num", "payload", "acked", "enqueued_time", "sent_time", "send_attempts", "next") def __init__(self): self.seq_num = 0 self.payload = b"" self.acked = False self.enqueued_time = 0.0 self.sent_time = 0.0 self.send_attempts = 1 self.next = None class SendWindow: """Manages the outgoing datagram queue with reliable delivery.""" FAST_RETRANSMIT_DUP_ACKS = 2 def __init__(self, session): self._session = session self._cond = threading.Condition() self._head = None self._tail = None self._is_alive = True self._size = 0 self._seq_num = random.randint(0, 0xFFFF) self.send_size = 8 self.send_retry = SEND_RETRY_VAL / 1000.0 # seconds self.max_retry_backoff = MAX_RETRY_BACKOFF / 1000.0 # seconds self._last_checked_ack = -1 self._dup_ack_count = 0 def size(self): with self._cond: return self._size def cur_seq_num(self): return self._seq_num & 0xFFFF def kill(self): with self._cond: self._is_alive = False self._cond.notify_all() def clear(self): with self._cond: self._size = 0 self._head = None self._tail = None self._cond.notify_all() def send(self, payload): p = _Packet() p.enqueued_time = _ticks() with self._cond: p.seq_num = self._seq_num & 0xFFFF self._seq_num = (self._seq_num + 1) & 0xFFFF self._enqueue(p) p.payload = payload p.sent_time = _ticks() self._session._send_raw(self._to_msg(p)) def _enqueue(self, p): with self._cond: while self._is_alive and (self._full() or self._exceeds_ack_more_limit(p)): self._cond.wait() if not self._is_alive: return if self._tail is None: self._head = self._tail = p else: self._tail.next = p self._tail = p self._size += 1 self._cond.notify_all() def _full(self): return self._size >= self.send_size def _exceeds_ack_more_limit(self, p): last_ack = self._session.last_ack if last_ack < 0: return False span = (p.seq_num - last_ack) & 0xFFFF return span > self._session.ack_more_byte_limit def _to_msg(self, packet): msg = DaspMsg() msg.msg_type = DATAGRAM msg.session_id = self._session.remote_id msg.seq_num = packet.seq_num msg.payload = packet.payload self._session.receive_window.set_ack_headers(msg) return msg def check_ack_headers(self, msg): with self._cond: ack_num = msg.ack if ack_num < 0: return if self._head is None: self._track_dup_acks(ack_num) return unack_start = (ack_num + 1) & 0xFFFF unack_end = (unack_start + 10000) & 0xFFFF p = self._head while p is not None: if not p.acked and self._is_acked(unack_start, unack_end, msg, p.seq_num): p.acked = True if p is self._head and p.acked: self._head = p.next self._size -= 1 p = p.next if self._head is None: self._tail = None self._track_dup_acks(ack_num) self._cond.notify_all() def _track_dup_acks(self, ack_num): if ack_num != self._last_checked_ack: self._last_checked_ack = ack_num self._dup_ack_count = 0 return self._dup_ack_count += 1 if self._dup_ack_count < self.FAST_RETRANSMIT_DUP_ACKS: return self._dup_ack_count = 0 if self._head is not None and not self._head.acked: self._resend(self._head) @staticmethod def _is_acked(unack_start, unack_end, msg, seq_num): if unack_start <= unack_end: if seq_num < unack_start or seq_num > unack_end: return True else: if seq_num < unack_start and seq_num > unack_end: return True ack_more = msg.ack_more if ack_more is not None: diff = (seq_num - msg.ack) & 0xFFFF index = len(ack_more) - (diff >> 3) - 1 if 0 <= index < len(ack_more): ack_more_byte = ack_more[index] & 0xFF mask = 1 << (diff & 0x07) if ack_more_byte & mask: return True return False def send_retries(self): with self._cond: now = _ticks() oldest = 0.0 p = self._head while p is not None: if not p.acked: backoff_shift = min(p.send_attempts - 1, 32) interval = min(self.send_retry * (1 << backoff_shift), self.max_retry_backoff) if now - p.sent_time >= interval: self._resend(p) oldest = max(oldest, now - p.enqueued_time) p = p.next return oldest def _resend(self, p): p.sent_time = _ticks() p.send_attempts += 1 self._session.num_retries += 1 self._session._send_raw(self._to_msg(p)) # ──────────────────────────────────────────────────────────────── # DaspSession (from DaspSession.java) # ──────────────────────────────────────────────────────────────── class DaspSession: """ Client endpoint of a DASP session. Handles the handshake, keepalive, reliable send/receive, and session close. """ def __init__(self, dasp_socket, session_id, host, port, **options): self.socket = dasp_socket self.id = session_id self.host = host self.port = port self.remote_id = -1 self.is_closed = False self.close_cause = "???" self.user = "" self.password = "" self.ideal_max = options.get("ideal_max", IDEAL_MAX_DEF) self.abs_max = options.get("abs_max", ABS_MAX_DEF) self.receive_timeout = options.get("receive_timeout", RECEIVE_TIMEOUT_VAL) / 1000.0 # seconds self.connect_timeout = options.get("connect_timeout", CONNECT_TIMEOUT_VAL) / 1000.0 # seconds self.ack_more_byte_limit = options.get("ack_more_byte_limit", ACK_MORE_BYTE_LIMIT) self.digest_algorithm = options.get("digest_algorithm", DIGEST_ALGORITHM) self.include_message_len = options.get("include_message_len", True) self.receive_window = ReceiveWindow(self) self.send_window = SendWindow(self) self.send_window.send_retry = options.get("send_retry", SEND_RETRY_VAL) / 1000.0 self.send_window.max_retry_backoff = options.get("max_retry_backoff", MAX_RETRY_BACKOFF) / 1000.0 self._receive_queue = queue.Queue(maxsize=2000) self.remote_receive_max = RECEIVE_MAX_DEF self.last_ack = -1 self.last_send = _ticks() self.last_receive = _ticks() self.num_sent = 0 self.num_received = 0 self.num_retries = 0 self._nonce = None self._close_lock = threading.RLock() self.on_close = None # callback # ── Client handshake ──────────────────────────────────────── def connect(self): res = self._hello() if res.msg_type == CLOSE: raise DaspException("Connection denied", res.error_code) elif res.msg_type == CHALLENGE: self.remote_id = res.remote_id welcome = self._authenticate(res) elif res.msg_type == WELCOME: self.remote_id = res.remote_id welcome = res else: raise DaspException(f"Unexpected response to hello: {res.msg_type}") if welcome.msg_type == CLOSE: raise DaspException("Connection denied", welcome.error_code) elif welcome.msg_type == WELCOME: self._tune(welcome) else: raise DaspException(f"Expected welcome, not {welcome.msg_type}") def _hello(self): msg = DaspMsg() msg.msg_type = HELLO msg.session_id = 0xFFFF msg.seq_num = self.send_window.cur_seq_num() msg.remote_id = self.id msg.version = VERSION_VAL msg.ideal_max = self.ideal_max msg.abs_max = self.abs_max msg.receive_max = RECEIVE_MAX_VAL msg.receive_timeout = int(self.receive_timeout * 1000) timeout = max(self.connect_timeout / 3, 1.0) for _ in range(3): self._send_raw(msg) try: res = self._receive_queue.get(timeout=timeout) return res except queue.Empty: continue raise DaspException("No response from hello") def _authenticate(self, challenge): algorithm = challenge.digest_algorithm or "SHA-1" if algorithm not in ("SHA-1", "SHA-256"): self.close(ERR_DIGEST_NOT_SUPPORTED, "digest not supported") raise DaspException("Unsupported algorithm", ERR_DIGEST_NOT_SUPPORTED) # credentials: always SHA-1 of "user:pass" cred = hashlib.sha1(f"{self.user}:{self.password}".encode("utf-8")).digest() # outer digest uses negotiated algorithm algo_name = "sha1" if algorithm == "SHA-1" else "sha256" md = hashlib.new(algo_name) md.update(cred) md.update(challenge.nonce) digest_bytes = md.digest() msg = DaspMsg() msg.msg_type = AUTHENTICATE msg.session_id = self.remote_id msg.seq_num = self.send_window.cur_seq_num() msg.username = self.user msg.digest = digest_bytes timeout = max(self.connect_timeout / 3, 1.0) for _ in range(3): self._send_raw(msg) try: res = self._receive_queue.get(timeout=timeout) if res.msg_type != CHALLENGE: return res # drop duplicate challenge from multiple hellos except queue.Empty: continue raise DaspException("No response from authenticate") def _tune(self, msg): remote_ideal = msg.ideal_max if msg.ideal_max > -1 else IDEAL_MAX_DEF remote_abs = msg.abs_max if msg.abs_max > -1 else ABS_MAX_DEF remote_recv_max = msg.receive_max if msg.receive_max > -1 else RECEIVE_MAX_DEF remote_timeout = (msg.receive_timeout if msg.receive_timeout > -1 else RECEIVE_TIMEOUT_DEF) / 1000.0 self.ideal_max = min(remote_ideal, self.ideal_max) self.abs_max = min(remote_abs, self.abs_max) self.remote_receive_max = remote_recv_max self.receive_timeout = max(remote_timeout, self.receive_timeout) self.send_window.send_size = min(self.remote_receive_max, self.ack_more_byte_limit) self.receive_window.init(msg.seq_num) # ── Send / Receive ────────────────────────────────────────── def send(self, payload): if self.is_closed: raise DaspException(f"Session closed: {self.close_cause}") self.send_window.send(payload) def receive(self, timeout=1.0): if self.is_closed: raise DaspException(f"Session closed: {self.close_cause}") try: msg = self._receive_queue.get(timeout=timeout) except queue.Empty: return None if self.is_closed: raise DaspException(f"Session closed: {self.close_cause}") if not isinstance(msg, DaspMsg): return msg if msg.msg_type != DATAGRAM: raise DaspException(f"Invalid message type received: {msg.msg_type}") return msg def _send_raw(self, msg): msg.include_message_len = self.include_message_len self.last_send = _ticks() self.num_sent += 1 self.socket._send_msg(self, msg) # ── Dispatch (called on receiver thread) ──────────────────── def dispatch(self, msg): self.num_received += 1 self.last_receive = _ticks() if msg.ack >= 0: self.last_ack = msg.ack self.send_window.check_ack_headers(msg) if msg.msg_type == DATAGRAM: if not self.receive_window.receive(msg.seq_num): return self._enqueue(msg) elif msg.msg_type in (CHALLENGE, WELCOME): self._enqueue(msg) elif msg.msg_type == CLOSE: self._enqueue(msg) self._shutdown(f"remote close (error={msg.error_code})") elif msg.msg_type == KEEPALIVE: pass # lastReceive already updated def _enqueue(self, msg): try: self._receive_queue.put_nowait(msg) except queue.Full: self.close(ERR_TIMEOUT, "receive queue full") # ── Housekeeping (called every 100ms) ─────────────────────── def housekeeping(self): if self.is_closed: return if _ticks() - self.last_receive > self.receive_timeout: self.close(ERR_TIMEOUT, "receive timeout") return oldest = self.send_window.send_retries() if oldest > self.receive_timeout: self.close(ERR_TIMEOUT, "unacked send timeout") return if (self.receive_window.unacked() or _ticks() - self.last_send > self.receive_timeout / 3): self._keepalive() def _keepalive(self): msg = DaspMsg() msg.msg_type = KEEPALIVE msg.session_id = self.remote_id msg.seq_num = 0xFFFF self.receive_window.set_ack_headers(msg) self._send_raw(msg) # ── Close ─────────────────────────────────────────────────── def close(self, error_code=-1, cause="close()"): with self._close_lock: if self.is_closed: return close_msg = DaspMsg() close_msg.msg_type = CLOSE close_msg.session_id = self.remote_id if self.remote_id >= 0 else 0xFFFF close_msg.seq_num = 0xFFFF if error_code >= 0: close_msg.error_code = error_code self._send_raw(close_msg) if error_code < 0: time.sleep(0.05) self._send_raw(close_msg) self._shutdown(cause) def _shutdown(self, cause): with self._close_lock: if self.is_closed: return self.is_closed = True self.close_cause = cause self.send_window.kill() # wake anyone blocked on receive try: self._receive_queue.put_nowait(None) except queue.Full: pass if self.on_close: try: self.on_close(self) except Exception: pass self.socket._free_session(self) # ──────────────────────────────────────────────────────────────── # DaspSocket (from DaspSocket.java + DaspSocketInterface.java) # ──────────────────────────────────────────────────────────────── class DaspSocket: """ Manages a UDP socket shared by DASP sessions. Provides receiver and housekeeping background threads. """ def __init__(self, port=-1): self._sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self._sock.settimeout(1.0) if port > 0: self._sock.bind(("", port)) else: self._sock.bind(("", 0)) self._sessions = {} # id -> DaspSession self._sessions_lock = threading.Lock() self._is_alive = True self._rand = random.Random() self._send_lock = threading.Lock() self._receiver_thread = threading.Thread( target=self._receiver_loop, name="DASP-Receiver", daemon=True) self._housekeeping_thread = threading.Thread( target=self._housekeeping_loop, name="DASP-Housekeeping", daemon=True) self._receiver_thread.start() self._housekeeping_thread.start() @property def port(self): return self._sock.getsockname()[1] @property def is_closed(self): return not self._is_alive def connect(self, host, port, user, password, **options): if not self._is_alive: raise DaspException("Socket is closed") session = self._alloc_session(host, port, **options) session.user = user session.password = password session.connect() return session def close(self): if not self._is_alive: return self._is_alive = False with self._sessions_lock: for s in list(self._sessions.values()): try: s.close(-1, "socket closed") except Exception: pass try: self._sock.close() except Exception: pass # ── Session management ────────────────────────────────────── def _alloc_session(self, host, port, **options): with self._sessions_lock: while True: sid = self._rand.randint(0, 0xFFFE) if sid not in self._sessions: break session = DaspSession(self, sid, host, port, **options) self._sessions[sid] = session return session def _free_session(self, session): with self._sessions_lock: self._sessions.pop(session.id, None) # ── Network I/O ───────────────────────────────────────────── def _send_msg(self, session, msg): data = msg.encode() _trace("S", msg, data) with self._send_lock: try: self._sock.sendto(data, (session.host, session.port)) except OSError: pass def _receiver_loop(self): buf = bytearray(ABS_MAX_VAL + 64) while self._is_alive: try: nbytes, addr = self._sock.recvfrom_into(buf) if nbytes < 5: continue raw = bytes(buf[:nbytes]) msg = DaspMsg.decode(buf, nbytes) self._dispatch(msg, addr, raw) except socket.timeout: continue except OSError: if self._is_alive: continue break def _dispatch(self, msg, addr, raw=None): _trace("R", msg, raw) session_id = msg.session_id with self._sessions_lock: session = self._sessions.get(session_id) if session is None: return if session.host != addr[0] or session.port != addr[1]: return session.dispatch(msg) def _housekeeping_loop(self): while self._is_alive: time.sleep(0.1) with self._sessions_lock: sessions = list(self._sessions.values()) for s in sessions: try: s.housekeeping() except Exception: pass # ──────────────────────────────────────────────────────────────── # Utility # ──────────────────────────────────────────────────────────────── def _ticks(): """Monotonic time in seconds (float).""" return time.monotonic()