distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Session registry -- maps nicknames to active sessions.
2
3Ported from net.i2p.sam.SAMBridge session management.
4"""
5
6from __future__ import annotations
7
8import asyncio
9from dataclasses import dataclass, field
10from typing import TYPE_CHECKING
11
12if TYPE_CHECKING:
13 from i2p_sam.handler import SAMHandler
14
15
16@dataclass
17class SessionRecord:
18 """Record for a registered SAM session."""
19
20 nickname: str
21 style: str # STREAM, DATAGRAM, RAW, PRIMARY
22 destination: bytes # serialized Destination (raw bytes)
23 destination_b64: str
24 handler: "SAMHandler"
25 subsessions: dict[str, "SessionRecord"] = field(default_factory=dict)
26
27
28class SessionsDB:
29 """Singleton registry of active SAM sessions by nickname.
30
31 Thread-safe via asyncio.Lock.
32 """
33
34 def __init__(self) -> None:
35 self._sessions: dict[str, SessionRecord] = {}
36 self._lock = asyncio.Lock()
37
38 async def add(self, record: SessionRecord) -> bool:
39 """Add session. Returns False if nickname already exists.
40
41 Args:
42 record: The session record to register.
43
44 Returns:
45 True if added, False if nickname is already taken.
46 """
47 async with self._lock:
48 if record.nickname in self._sessions:
49 return False
50 self._sessions[record.nickname] = record
51 return True
52
53 async def remove(self, nickname: str) -> SessionRecord | None:
54 """Remove and return session.
55
56 Args:
57 nickname: The session nickname to remove.
58
59 Returns:
60 The removed SessionRecord, or None if not found.
61 """
62 async with self._lock:
63 return self._sessions.pop(nickname, None)
64
65 async def get(self, nickname: str) -> SessionRecord | None:
66 """Look up session by nickname.
67
68 Args:
69 nickname: The session nickname to look up.
70
71 Returns:
72 The SessionRecord, or None if not found.
73 """
74 async with self._lock:
75 return self._sessions.get(nickname)
76
77 async def has(self, nickname: str) -> bool:
78 """Check if a session with the given nickname exists."""
79 async with self._lock:
80 return nickname in self._sessions
81
82 @property
83 def count(self) -> int:
84 """Number of registered sessions."""
85 return len(self._sessions)