distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tunnel dispatcher — route messages through tunnels."""
2
3from typing import Callable
4from i2p_data.tunnel import TunnelId
5
6
7class TunnelDispatcher:
8 """Routes I2NP messages through registered tunnels."""
9
10 def __init__(self):
11 self._inbound: dict[int, dict] = {}
12 self._outbound: dict[int, dict] = {}
13
14 def inbound_count(self) -> int:
15 return len(self._inbound)
16
17 def outbound_count(self) -> int:
18 return len(self._outbound)
19
20 def register_inbound(self, tunnel_id: TunnelId, gateway: bytes,
21 handler: Callable | None = None):
22 self._inbound[int(tunnel_id)] = {"gateway": gateway, "handler": handler}
23
24 def register_outbound(self, tunnel_id: TunnelId, gateway: bytes,
25 handler: Callable | None = None):
26 self._outbound[int(tunnel_id)] = {"gateway": gateway, "handler": handler}
27
28 def unregister_inbound(self, tunnel_id: TunnelId):
29 self._inbound.pop(int(tunnel_id), None)
30
31 def unregister_outbound(self, tunnel_id: TunnelId):
32 self._outbound.pop(int(tunnel_id), None)
33
34 def lookup_inbound(self, tunnel_id: TunnelId) -> dict | None:
35 return self._inbound.get(int(tunnel_id))
36
37 def lookup_outbound(self, tunnel_id: TunnelId) -> dict | None:
38 return self._outbound.get(int(tunnel_id))
39
40 def dispatch_inbound(self, tunnel_id: TunnelId, data: bytes):
41 info = self._inbound.get(int(tunnel_id))
42 if info and info.get("handler"):
43 info["handler"](data)
44
45 def dispatch_outbound(self, tunnel_id: TunnelId, data: bytes):
46 info = self._outbound.get(int(tunnel_id))
47 if info and info.get("handler"):
48 info["handler"](data)