* Added did_auth and did_deauth callbacks.

* Fixed bug in the parsing of host addressed envelopes.

* Added automatic re-authentication when retrying is enabled.

Author: DobromirM

swimos/client/_connections.py

@@ -24,6 +24,7 @@
 if TYPE_CHECKING:
     from ._downlinks._downlinks import _DownlinkModel
     from ._downlinks._downlinks import _DownlinkView
+    from .. import SwimClient
 
 
 class RetryStrategy:
@@ -79,7 +80,8 @@ def reset(self):
 
 class _ConnectionPool:
 
-    def __init__(self, retry_strategy: RetryStrategy = RetryStrategy()) -> None:
+    def __init__(self, client: 'SwimClient', retry_strategy: RetryStrategy = RetryStrategy()) -> None:
+        self.__client = client
         self.__connections = dict()
         self.retry_strategy = retry_strategy
 
@@ -102,7 +104,7 @@ async def _get_connection(self, host_uri: str, scheme: str, keep_linked: bool,
         connection = self.__connections.get(host_uri)
 
         if connection is None or connection.status == _ConnectionStatus.CLOSED:
-            connection = _WSConnection(host_uri, scheme, keep_linked, keep_synced, self.retry_strategy)
+            connection = _WSConnection(self.__client, host_uri, scheme, keep_linked, keep_synced, self.retry_strategy)
             self.__connections[host_uri] = connection
 
         return connection
@@ -154,7 +156,7 @@ async def _remove_downlink_view(self, downlink_view: '_DownlinkView') -> None:
 
 class _WSConnection:
 
-    def __init__(self, host_uri: str, scheme: str, keep_linked, keep_synced,
+    def __init__(self, client: 'SwimClient', host_uri: str, scheme: str, keep_linked, keep_synced,
                  retry_strategy: RetryStrategy = RetryStrategy()) -> None:
         self.host_uri = host_uri
         self.scheme = scheme
@@ -163,12 +165,15 @@ def __init__(self, host_uri: str, scheme: str, keep_linked, keep_synced,
         self.connected = asyncio.Event()
         self.websocket = None
         self.status = _ConnectionStatus.CLOSED
+        self.auth_message = None
         self.init_message = None
 
         self.keep_linked = keep_linked
         self.keep_synced = keep_synced
 
         self.__subscribers = _DownlinkManagerPool()
+        self.__authenticated = asyncio.Event()
+        self.__client = client
 
     async def _open(self) -> None:
         if self.status == _ConnectionStatus.CLOSED:
@@ -211,6 +216,20 @@ def should_reconnect(self) -> bool:
         """
         return self.keep_linked or self.keep_synced
 
+    def _set_auth_message(self, message: str) -> None:
+        """
+        Set the initial auth message that gets sent when the underlying downlink is established.
+        """
+
+        self.auth_message = message
+
+    async def _send_auth_message(self) -> None:
+        """
+        Send the initial auth message for the underlying downlink if it is set.
+        """
+        if self.auth_message is not None:
+            await self._send_message(self.auth_message)
+
     def _set_init_message(self, message: str) -> None:
         """
         Set the initial message that gets sent when the underlying downlink is established.
@@ -283,15 +302,53 @@ async def _wait_for_messages(self) -> None:
                 while self.status == _ConnectionStatus.RUNNING:
                     message = await self.websocket.recv()
                     response = _Envelope._parse_recon(message)
-                    await self.__subscribers._receive_message(response)
+
+                    if response._route:
+                        await self.__subscribers._receive_message(response)
+                    else:
+                        await self._receive_message(self.host_uri, response)
             except ConnectionClosed as error:
                 exception_warn(error)
                 await self._close()
                 if self.should_reconnect() and await self.retry_strategy.retry():
                     await self._open()
+                    await self._send_auth_message()
                     await self._send_init_message()
                     continue
 
+    async def _receive_message(self, host_uri: str, message: '_Envelope') -> None:
+        """
+        Receive a host addressed message from the remote host.
+
+        :param host_uri:        - Uri of the remote host.
+        :param message:         - Message received from the remote host.
+        """
+
+        if message._tag == 'authed':
+            await self._receive_authed(host_uri, message)
+        elif message._tag == 'deauthed':
+            await self._receive_deauthed(host_uri, message)
+
+    async def _receive_authed(self, host_uri: str, message: '_Envelope') -> None:
+        """
+        Handle an `authed` response message from the remote agent.
+
+        :param host_uri:        - Uri of the remote host.
+        :param message:         - Message received from the remote host.
+        """
+        self.__authenticated.set()
+        await self.__client._execute_did_auth(host_uri, message)
+
+    async def _receive_deauthed(self, host_uri: str, message: '_Envelope') -> None:
+        """
+        Handle a `deauthed` response message from the remote agent.
+
+        :param host_uri:        - Uri of the remote host.
+        :param message:         - Message received from the remote host.
+        """
+        self.__authenticated.clear()
+        await self.__client._execute_did_deauth(host_uri, message)
+
 
 class _ConnectionStatus(Enum):
     CLOSED = 0

swimos/client/_swim_client.py

@@ -24,9 +24,10 @@
 from typing import Callable, Any, Optional
 from ._connections import _ConnectionPool, _WSConnection, RetryStrategy, IntervalStrategy
 from ._downlinks._downlinks import _ValueDownlinkView, _EventDownlinkView, _DownlinkView, _MapDownlinkView
+from ._downlinks._utils import validate_callback
 from ._utils import _URI, after_started, exception_warn
 from swimos.structures import RecordConverter
-from swimos.warp._warp import _CommandMessage
+from swimos.warp._warp import _CommandMessage, _AuthRequest, _Envelope
 
 
 class SwimClient:
@@ -41,7 +42,31 @@ def __init__(self, retry_strategy: RetryStrategy = IntervalStrategy(), terminate
         self._loop = None
         self._loop_thread = None
         self._has_started = False
-        self.__connection_pool = _ConnectionPool(retry_strategy)
+        self._did_auth_callback = None
+        self._did_deauth_callback = None
+        self.authed_hosts = dict()
+
+        self.__connection_pool = _ConnectionPool(self, retry_strategy)
+
+    def did_auth(self, function: Callable) -> 'SwimClient':
+        """
+        Set the `did_auth` callback of the current client to a given function.
+
+        :param function:   - Function to be called when a remote host is authenticated.
+        :return:           - The current Swim client.
+        """
+        self._did_auth_callback = validate_callback(function)
+        return self
+
+    def did_deauth(self, function: Callable) -> 'SwimClient':
+        """
+        Set the `did_deauth` callback of the current client to a given function.
+
+        :param function:   - Function to be called when a remote host is deauthenticated.
+        :return:           - The current Swim client.
+        """
+        self._did_deauth_callback = validate_callback(function)
+        return self
 
     def __enter__(self) -> 'SwimClient':
         self.start()
@@ -95,7 +120,7 @@ def stop(self) -> 'SwimClient':
 
         return self
 
-    def command(self, host_uri: str, node_uri: str, lane_uri: str, body: Any) -> 'Future':
+    def command(self, host_uri: str, node_uri: str, lane_uri: str, body: Any):
         """
         Send a command message to a command lane on a remote Swim agent.
 
@@ -107,6 +132,17 @@ def command(self, host_uri: str, node_uri: str, lane_uri: str, body: Any) -> 'Fu
 
         return self._schedule_task(self.__send_command, host_uri, node_uri, lane_uri, body)
 
+    def authenticate(self, host_uri: str, body: Any):
+        """
+        Send an authentication request to a remote Swim server.
+
+        :param host_uri:        - Host URI of the remote server.
+        :param body:            - The authentication message body.
+        """
+
+        self.authed_hosts[host_uri] = asyncio.Event()
+        return self._schedule_task(self.__authenticate, host_uri, body)
+
     def downlink_event(self) -> '_EventDownlinkView':
         """
         Create an Event Downlink.
@@ -156,6 +192,26 @@ async def _get_connection(self, host_uri: str, scheme: str, keep_linked: bool,
         connection = await self.__connection_pool._get_connection(host_uri, scheme, keep_linked, keep_synced)
         return connection
 
+    async def _execute_did_auth(self, host_uri: str, message: '_Envelope') -> None:
+        """
+        Execute the custom `did_auth` callback of the current Swim client.
+
+        :param host_uri:        - Uri of the remote host.
+        :param message:         - Message received from the remote host.
+        """
+        if self._did_auth_callback:
+            self._schedule_task(self._did_auth_callback, host_uri, message)
+
+    async def _execute_did_deauth(self, host_uri: str, message: '_Envelope') -> None:
+        """
+        Execute the custom `did_deauth` callback of the current Swim client.
+
+        :param host_uri:        - Uri of the remote host.
+        :param message:         - Message received from the remote host.
+        """
+        if self._did_deauth_callback:
+            self._schedule_task(self._did_deauth_callback, host_uri, message)
+
     @after_started
     def _schedule_task(self, task: Callable, *args: Any) -> 'Future':
         """
@@ -222,6 +278,22 @@ async def __send_command(self, host_uri: str, node_uri: str, lane_uri: str, body
         connection = await self._get_connection(host_uri, scheme, True, False)
         await connection._send_message(message._to_recon())
 
+    async def __authenticate(self, host_uri: str, body: Any) -> None:
+        """
+        Send an authentication request to a given host.
+
+        :param host_uri:        - Host URI of the remote host.
+        :param body:            - The authentication message body.
+        """
+        record = RecordConverter.get_converter().object_to_record(body)
+        host_uri, scheme = _URI._parse_uri(host_uri)
+        message = _AuthRequest(body=record)
+        connection = await self._get_connection(host_uri, scheme, True, False)
+        connection._set_auth_message(message._to_recon())
+        await connection._open()
+        self._schedule_task(connection._wait_for_messages)
+        await connection._send_message(message._to_recon())
+
     def __start_event_loop(self) -> None:
         asyncio.set_event_loop(self._loop)
         asyncio.get_event_loop().run_forever()

swimos/recon/_writers.py

@@ -135,7 +135,7 @@ def _write(key: 'Value' = None, writer: '_ReconWriter' = None, value: 'Value' =
         if key_text:
             output._append(key_text)
 
-        if value != _Extant._get_extant() and value is not None:
+        if value != _Extant._get_extant() and value != _Absent._get_absent() and value is not None:
             output._append('(')
             value_text = writer._write_value(value)
             output._append(value_text)