Use Sessions to send messages

ct-app/.configs/core_prod_config.yaml

@@ -34,7 +34,8 @@ flags:
     getTotalChannelFunds: 300
 
     observeMessageQueue: On # should only be On / Off
-    observeRelayedMessages: 15
+    observeRelayedMessages: On # should only be On / Off
+    observeSessions: 60
 
   peer:
     messageRelayRequest: On # should only be On / Off
@@ -84,7 +85,6 @@ economicModel:
         upperbound: 1.0
         offset: 4.08
 
-
 # =============================================================================
 # 
 # =============================================================================
@@ -102,6 +102,13 @@ channel:
   fundingAmount: 30
   maxAgeSeconds: 172800
 
+# =============================================================================
+# 
+# =============================================================================
+sessions:
+  packetSize: 462
+  numPackets: 10
+
 # =============================================================================
 # 
 # =============================================================================

ct-app/.configs/core_staging_config.yaml

@@ -34,7 +34,8 @@ flags:
     getTotalChannelFunds: 300
 
     observeMessageQueue: On  # should only be On / Off
-    observeRelayedMessages: 15
+    observeRelayedMessages: On # should only be On / Off
+    observeSessions: 60
 
   peer:
     messageRelayRequest: On # should only be On / Off
@@ -102,6 +103,13 @@ channel:
   fundingAmount: 0.1
   maxAgeSeconds: 86400
 
+# =============================================================================
+# 
+# =============================================================================
+sessions:
+  packetSize: 462
+  numPackets: 10
+
 # =============================================================================
 # 
 # =============================================================================

ct-app/core/api/__init__.py

@@ -1,3 +1,4 @@
 from .hoprd_api import HoprdAPI
+from .protocol import Protocol
 
-__all__ = ["HoprdAPI"]
+__all__ = ["HoprdAPI", "Protocol"]

ct-app/core/api/hoprd_api.py

@@ -3,18 +3,22 @@
 from typing import Optional
 
 import aiohttp
-
 from core.baseclass import Base
 
 from .http_method import HTTPMethod
+from .protocol import Protocol
 from .request_objects import (
     ApiRequestObject,
+    CreateSessionBody,
+    DeleteSessionBody,
+
     FundChannelBody,
     GetChannelsBody,
     GetPeersBody,
     OpenChannelBody,
-    PopMessagesBody,
-    SendMessageBody,
+    SessionCapabilitiesBody,
+    SessionPathBodyRelayers,
+    SessionTargetBody,
 )
 from .response_objects import (
     Addresses,
@@ -23,8 +27,8 @@
     Configuration,
     ConnectedPeer,
     Infos,
-    Message,
     OpenedChannel,
+    Session,
     TicketPrice,
     TicketProbability,
 )
@@ -185,22 +189,6 @@ async def get_address(self) -> Optional[Addresses]:
 
         return Addresses(response) if is_ok else None
 
-    async def send_message(
-        self, destination: str, message: str, hops: list[str], tag: int = MESSAGE_TAG
-    ) -> bool:
-        """
-        Sends a message to the given destination.
-        :param: destination: str
-        :param: message: str
-        :param: hops: list[str]
-        :param: tag: int = 0x0320
-        :return: bool
-        """
-        data = SendMessageBody(message, hops, destination, tag)
-        is_ok, _ = await self.__call_api(HTTPMethod.POST, "messages", data=data)
-
-        return is_ok
-
     async def node_info(self) -> Optional[Infos]:
         """
         Gets informations about the HOPRd node.
@@ -217,17 +205,6 @@ async def ticket_price(self) -> Optional[TicketPrice]:
         is_ok, response = await self.__call_api(HTTPMethod.GET, "network/price")
         return TicketPrice(response) if is_ok else None
 
-    async def messages_pop_all(self, tag: int = MESSAGE_TAG) -> list:
-        """
-        Pop all messages from the inbox
-        :param: tag = 0x0320
-        :return: list
-        """
-        is_ok, response = await self.__call_api(
-            HTTPMethod.POST, "messages/pop-all", data=PopMessagesBody(tag)
-        )
-        return [Message(item) for item in response.get("messages", [])] if is_ok else []
-
     async def winning_probability(self) -> Optional[TicketProbability]:
         """
         Gets the winning probability set in the HOPRd node configuration file.
@@ -236,6 +213,66 @@ async def winning_probability(self) -> Optional[TicketProbability]:
         is_ok, response = await self.__call_api(HTTPMethod.GET, "node/configuration")
         return TicketProbability(Configuration(json.loads(response)).as_dict) if is_ok else None
 
+    async def get_sessions(self, protocol: Protocol = Protocol.UDP) -> list[Session]:
+        """
+        Lists existing Session listeners for the given IP protocol
+        :param: protocol: Protocol
+        :return: list[Session]
+        """
+        is_ok, response = await self.__call_api(
+            HTTPMethod.GET, f"session/{protocol.name.lower()}"
+        )
+
+        return [Session(s) for s in response] if is_ok else None
+
+    async def post_session(
+        self,
+        destination: str,
+        relayer: str,
+        listen_host: str = ":0",
+        protocol: Protocol = Protocol.UDP
+    ) -> Optional[Session]:
+        """
+        Creates a new client session returning the given session listening host & port over TCP or UDP.
+        :param: destination: PeerID of the recipient
+        :param: relayer: PeerID of the relayer
+        :param: listen_host: str
+        :param: protocol: Protocol (UDP or TCP)
+        :return: Session
+        """
+        capabilities_body = SessionCapabilitiesBody(
+            protocol.retransmit, protocol.segment)
+        target_body = SessionTargetBody()
+        path_body = SessionPathBodyRelayers([relayer])
+
+        data = CreateSessionBody(
+            capabilities_body.as_array,
+            destination,
+            listen_host,
+            path_body.as_dict,
+            target_body.as_dict,
+        )
+
+        is_ok, response = await self.__call_api(
+            HTTPMethod.POST, f"session/{protocol.name.lower()}", data
+        )
+
+        return Session(response) if is_ok else None
+
+    async def close_session(self, session: Session) -> bool:
+        """
+        Closes an existing Sessino listener for the given IP protocol, IP and port.
+        :param: session: Session
+        """
+        data = DeleteSessionBody(session.ip, session.port)
+
+        is_ok, _ = await self.__call_api(
+            HTTPMethod.DELETE, f"session/{session.protocol}", data
+        )
+
+        return is_ok
+
+
     async def healthyz(self, timeout: int = 20) -> bool:
         """
         Checks if the node is healthy. Return True if `healthyz` returns 200 after max `timeout` seconds.

ct-app/core/api/protocol.py

@@ -0,0 +1,45 @@
+from enum import Enum
+
+
+class ProcolTemplate:
+    def __init__(self, retransmit: bool, segment: bool):
+        self.retransmit = retransmit
+        self.segment = segment
+
+
+class TCPProtocol(ProcolTemplate):
+    def __init__(self):
+        super().__init__(retransmit=True, segment=True)
+
+
+class UDPProtocol(ProcolTemplate):
+    def __init__(self):
+        super().__init__(retransmit=False, segment=False)
+
+
+class Protocol(Enum):
+    TCP = TCPProtocol()
+    UDP = UDPProtocol()
+
+    @classmethod
+    def fromString(cls, protocol: str):
+        try:
+            return getattr(cls, protocol.upper())
+        except AttributeError:
+            raise ValueError(
+                f"Invalid protocol: {protocol}. Valid values are: {[p.name for p in cls]}"
+            )
+
+    @property
+    def segment(self):
+        return self.value.segment
+
+    @property
+    def retransmit(self):
+        return self.value.retransmit
+
+    def __eq__(self, other):
+        if isinstance(other, str):
+            return other == self.name
+
+        return self.name == other.name

ct-app/core/api/request_objects.py

@@ -62,25 +62,83 @@ class GetChannelsBody(ApiRequestObject):
     def __init__(self, full_topology: str, including_closed: str):
         super().__init__(vars())
 
-class SendMessageBody(ApiRequestObject):
+
+class GetPeersBody(ApiRequestObject):
+    keys = {"quality": "quality"}
+
+    def __init__(self, quality: float):
+        super().__init__(vars())
+
+
+class CreateSessionBody(ApiRequestObject):
     keys = {
-        "body": "body",
+        "capabilities": "capabilities",
+        "destination": "destination",
+        "listen_host": "listenHost",
         "path": "path",
-        "destination": "peerId",
-        "tag": "tag",
+        "target": "target",
     }
 
-    def __init__(self, body: str, path: list[str], destination: str, tag: int):
+    def __init__(
+        self,
+        capabilities: list,
+        destination: str,
+        listen_host: str,
+        path: str,
+        target: str,
+    ):
         super().__init__(vars())
 
-class PopMessagesBody(ApiRequestObject):
-    keys = {"tag": "tag"}
 
-    def __init__(self, tag: int):
+class SessionCapabilitiesBody(ApiRequestObject):
+    keys = {
+        "retransmission": "Retransmission",
+        "segmentation": "Segmentation",
+    }
+
+    def __init__(self, retransmission: bool, segmentation: bool):
         super().__init__(vars())
 
-class GetPeersBody(ApiRequestObject):
-    keys = {"quality": "quality"}
+    @property
+    def as_array(self) -> list:
+        return [
+            self.keys[var] for var in vars(self) if var in self.keys and vars(self)[var]
+        ]
+
+
+class SessionPathBodyRelayers(ApiRequestObject):
+    keys = {
+        "relayers": "IntermediatePath",
+    }
+
+    def __init__(self, relayers: list[str]):
+        super().__init__(vars())
+
+
+class SessionPathBodyHops(ApiRequestObject):
+    keys = {
+        "hops": "Hops",
+    }
+
+    def __init__(self, hops: int = 0):
+        super().__init__(vars())
+
+    def post_init(self):
+        self.hops = int(self.hops)
+
+
+class SessionTargetBody(ApiRequestObject):
+    keys = {
+        "service": "Service",
+    }
+
+    def __init__(self, service: int = 0):
+        super().__init__(vars())
+
+
+class DeleteSessionBody(ApiRequestObject):
+    keys = {"ip": "listeningIp", "port": "port"}
+
+    def __init__(self, ip: str, port: str):
+        super().__init__(vars())
 
-    def __init__(self, quality: float):
-        super().__init__(vars())

ct-app/core/api/response_objects.py

@@ -94,11 +94,11 @@ class Channel(ApiResponseObject):
 
     def post_init(self):
         self.status = ChannelStatus.fromString(self.status)
+
         if isinstance(self.destination_address, str):
             self.destination_address = self.destination_address.lower()
         if isinstance(self.source_address, str):
             self.source_address = self.source_address.lower()
-
 
 class TicketPrice(ApiResponseObject):
     keys = {"value": "price"}
@@ -121,8 +121,6 @@ class Configuration(ApiResponseObject):
 class OpenedChannel(ApiResponseObject):
     keys = {"channel_id": "channelId", "receipt": "transactionReceipt"}
 
-class Message(ApiResponseObject):
-    keys = {"body": "body", "timestamp": "timestamp"}
 
 class Channels:
     def __init__(self, data: dict):
@@ -134,4 +132,13 @@ def __str__(self):
         return str(self.__dict__)
 
     def __repr__(self):
-        return str(self)
+        return str(self)
+
+
+class Session(ApiResponseObject):
+    keys = {
+        "ip": "ip",
+        "port": "port",
+        "protocol": "protocol",
+        "target": "target",
+    }

ct-app/core/components/asyncloop.py

@@ -15,7 +15,6 @@ def __init__(self):
         self.loop.add_signal_handler(SIGINT, self.stop)
         self.loop.add_signal_handler(SIGTERM, self.stop)
 
-    @classmethod
     def run(cls, process: Callable, stop_callback: Callable):
         try:
             cls().loop.run_until_complete(process())