Move InMemorySubscriber from L2 to L3 client module (#56)

* Move InMemorySubscriber from L2 to L3 client module

* Updated UListener on_receive method to be asynchronous and add additional test cases

* Update READMEs

* Bump version to 0.2.0  for major release introducing L2 and L3.

* Delete client README.html

* Incorporated  Reviewers comment

* Fixed review comments

* Fixed communication status issue

* Added subscriber info in unsubscribe request

* Address review comments

Author: neelam-kushwah

README.adoc

@@ -3,7 +3,9 @@
 
 == Overview
 
-This library implements the https://github.com/eclipse-uprotocol/uprotocol-spec/blob/main/languages.adoc[uProtocol Language Specific Library Requirements] for Python defined in https://github.com/eclipse-uprotocol/uprotocol-spec/tree/main[uProtocol Specifications]. The library is organized into packages that are described in <<sdk-packages>> below. Each package contains a README.adoc file that describes the purpose of the package and how to use it.
+This library implements the https://github.com/eclipse-uprotocol/uprotocol-spec/blob/main/languages.adoc[uProtocol Language Specific Library Requirements] for Python defined in https://github.com/eclipse-uprotocol/uprotocol-spec/tree/main[uProtocol Specifications]. The library is organized into packages that are described in <<sdk-packages>> below and organized by the layers of the protocol.
+
+Each package contains a README.adoc file that describes the purpose of the package and how to use it.
 
 The module contains the factory methods, serializers, and validators for all data types defined in the specifications, and any data models that either haven't or couldn't be defined in uprotocol-core-api yet (ex. UPayload) This library fits into the big picture of the uProtocol SDK as seen in the diagram below.
 
@@ -63,23 +65,31 @@ The Library is broken up into different packages that are described in <<sdk-pac
 
 
 .SDK Packages
-[#sdk-packages,width=100%,cols="20%,80%",options="header"]
+[#sdk-packages,width=100%,cols="1,2,5",options="header"]
 |===
 
-| Package | Purpose
+| Package | Protocol Layer | Purpose
+
+| link:uprotocol/client/README.adoc[`*client*`]
+| https://github.com/eclipse-uprotocol/up-spec/tree/main/up-l3[Application Layer (uP-L3)]
+| Top level client-facing interfaces to communication with USubscription, UDiscovery, and UTwin services.
+
+| link:uprotocol/communication/README.adoc[`*communication*`]
+| https://github.com/eclipse-uprotocol/up-spec/tree/main/up-l2[communication layer (uP-L2)]
+| Common implementation of communication messaging patterns (publisher, subscriber, Rpcclient, RpcServer, etc..) that is build on top of the L1 transport interface (see below)
+
+| link:uprotocol/transport/README.adoc[`*transport*`]
+| https://github.com/eclipse-uprotocol/uprotocol-spec/blob/main/up-l1/README.adoc[Transport Layer (uP-L1)]
+| Interface and data model for how to send() and receive() messages in a common way across various transport technologies (ex. zenoh, mqtt, http, etc...). the interface is implemented by transports (ex. up-transport-zenoh-python), and the interface is then used to build the uProtocol layer 2 communication layer implementation.
 
 | link:uprotocol/uri/README.adoc[`*uri*`]
+|
 | Uniform Resource Identifier (RFC3986), how uProtocol addresses things (devices, software, methods, topics, etc...) on the network.
 
 | link:uprotocol/uuid/README.adoc[`*uuid*`]
+|
 | Identifier used to uniquely identify (and timestamp) messages that are sent.
 
-| link:uprotocol/communication/README.adoc[`*communication*`]
-| Interface to build entities that use UTransport APIs to communicate with other entities. This is described in further detail on the up-spec page about https://github.com/eclipse-uprotocol/up-spec/tree/main/up-l2[L2 APIs].
-
-| link:uprotocol/transport/README.adoc[`*transport*`]
-| Interface and data model declaration used for bidirectional point-2-point communication between uEs. This interface is then implemented by client libraries (described https://github.com/eclipse-uprotocol/up-spec/blob/main/up-l1/README.adoc[here]) for a given underlying transport (ex. Binder, MQTT, Zenoh, SOME/IP, DDS, HTTP, etc…​)
-
 |===
 
 NOTE: Please visit the READMEs in <<sdk-packages>> for examples of how to use the different data types and their factories, validators, and serializers.

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "up-python"
-version = "0.1.3-dev"
+version = "0.2.0-dev"
 description = "Language specific uProtocol library for building and using UUri, UUID, UAttributes, UTransport, and more."
 authors = ["Neelam Kushwah <neelam.kushwah@gm.com>"]
 license = "The Apache License, Version 2.0"

tests/test_client/__init__.py

@@ -81,13 +81,13 @@ async def send(self, message: UMessage) -> UStatus:
 
         if message.attributes.type == UMessageType.UMESSAGE_TYPE_REQUEST:
             response = self.build_response(message)
-            self._notify_listeners(response)
+            await self._notify_listeners(response)
 
         return UStatus(code=UCode.OK)
 
-    def _notify_listeners(self, response: UMessage):
+    async def _notify_listeners(self, response: UMessage):
         for listener in self.listeners:
-            listener.on_receive(response)
+            await listener.on_receive(response)
 
     async def register_listener(self, source: UUri, listener: UListener, sink: UUri = None) -> UStatus:
         self.listeners.append(listener)
@@ -130,12 +130,22 @@ def build_response(self, request):
         )
 
 
+class CommStatusUCodeOKTransport(MockUTransport):
+    def build_response(self, request):
+        status = UStatus(code=UCode.OK, message="No Communication Error")
+        return (
+            UMessageBuilder.response_for_request(request.attributes)
+            .with_commstatus(status.code)
+            .build_from_upayload(UPayload.pack(status))
+        )
+
+
 class EchoUTransport(MockUTransport):
     def build_response(self, request):
         return request
 
     async def send(self, message):
         response = self.build_response(message)
-        executor = ThreadPoolExecutor(max_workers=1)
-        executor.submit(self._notify_listeners, response)
+
+        await self._notify_listeners(response)
         return UStatus(code=UCode.OK)

tests/test_client/test_usubscription/__init__.py

@@ -14,7 +14,12 @@
 
 import unittest
 
-from tests.test_communication.mock_utransport import CommStatusTransport, MockUTransport, TimeoutUTransport
+from tests.test_communication.mock_utransport import (
+    CommStatusTransport,
+    CommStatusUCodeOKTransport,
+    MockUTransport,
+    TimeoutUTransport,
+)
 from uprotocol.communication.calloptions import CallOptions
 from uprotocol.communication.inmemoryrpcclient import InMemoryRpcClient
 from uprotocol.communication.upayload import UPayload
@@ -101,6 +106,12 @@ async def test_invoke_method_with_comm_status_transport(self):
         self.assertEqual(UCode.FAILED_PRECONDITION, context.exception.status.code)
         self.assertEqual("Communication error [FAILED_PRECONDITION]", context.exception.status.message)
 
+    async def test_invoke_method_with_comm_status_transport(self):
+        rpc_client = InMemoryRpcClient(CommStatusUCodeOKTransport())
+        payload = UPayload.pack_to_any(UUri())
+        response = await rpc_client.invoke_method(self.create_method_uri(), payload, None)
+        self.assertEqual(UCode.OK, UPayload.unpack(response, UStatus).code)
+
     async def test_invoke_method_with_error_transport(self):
         class ErrorUTransport(MockUTransport):
             async def send(self, message):

tests/test_client/test_usubscription/test_v3/__init__.py

@@ -18,6 +18,8 @@
 from unittest.mock import AsyncMock, MagicMock
 
 from tests.test_communication.mock_utransport import EchoUTransport
+from uprotocol.communication.calloptions import CallOptions
+from uprotocol.communication.inmemoryrpcclient import InMemoryRpcClient
 from uprotocol.communication.inmemoryrpcserver import InMemoryRpcServer
 from uprotocol.communication.requesthandler import RequestHandler
 from uprotocol.communication.upayload import UPayload
@@ -27,6 +29,7 @@
 from uprotocol.transport.utransport import UTransport
 from uprotocol.uri.serializer.uriserializer import UriSerializer
 from uprotocol.v1.ucode_pb2 import UCode
+from uprotocol.v1.umessage_pb2 import UMessage
 from uprotocol.v1.uri_pb2 import UUri
 from uprotocol.v1.ustatus_pb2 import UStatus
 
@@ -136,7 +139,7 @@ async def custom_register_listener_behavior(source: UUri, listener: UListener, s
         async def custom_send_behavior(message):
             serialized_uri = UriSerializer().serialize(message.attributes.sink)
             if serialized_uri in listeners:
-                listeners[serialized_uri].on_receive(message)
+                await listeners[serialized_uri].on_receive(message)
             return UStatus(code=UCode.OK)
 
         self.mock_transport.send = AsyncMock(side_effect=custom_send_behavior)
@@ -196,6 +199,22 @@ async def test_handle_requests_exception(self):
         status = await transport.send(request)
         self.assertEqual(status.code, UCode.OK)
 
+    async def test_end_to_end_rpc_with_test_transport(self):
+        class MyRequestHandler(RequestHandler):
+            def handle_request(self, message: UMessage) -> UPayload:
+                return UPayload.pack(UUri())
+
+        handler = MyRequestHandler()
+        test_transport = EchoUTransport()
+        server = InMemoryRpcServer(test_transport)
+        method = self.create_method_uri()
+
+        self.assertEqual((await server.register_request_handler(method, handler)).code, UCode.OK)
+        rpc_client = InMemoryRpcClient(test_transport)
+        response = await rpc_client.invoke_method(method, None, CallOptions.DEFAULT)
+        self.assertIsNotNone(response)
+        self.assertEqual(response, UPayload.pack(UUri()))
+
 
 if __name__ == '__main__':
     unittest.main()

…communication/test_inmemorysubscriber.py …st_v3/test_inmemoryusubcriptionclient.pytests/test_communication/test_inmemorysubscriber.py renamed to tests/test_client/test_usubscription/test_v3/test_inmemoryusubcriptionclient.py tests/test_communication/test_inmemorysubscriber.py renamed to tests/test_client/test_usubscription/test_v3/test_inmemoryusubcriptionclient.py

@@ -62,7 +62,7 @@ async def test_register_listener(self):
         self.transport.get_source.return_value = self.source
 
         class TestListener(UListener):
-            def on_receive(self, message: UMessage):
+            async def on_receive(self, message: UMessage):
                 pass
 
         listener = TestListener()
@@ -78,7 +78,7 @@ async def test_unregister_notification_listener(self):
         self.transport.unregister_listener.return_value = UStatus(code=UCode.OK)
 
         class TestListener(UListener):
-            def on_receive(self, message: UMessage):
+            async def on_receive(self, message: UMessage):
                 pass
 
         listener = TestListener()
@@ -97,7 +97,7 @@ async def test_unregister_listener_not_registered(self):
         self.transport.unregister_listener.return_value = UStatus(code=UCode.NOT_FOUND)
 
         class TestListener(UListener):
-            def on_receive(self, message: UMessage):
+            async def on_receive(self, message: UMessage):
                 pass
 
         listener = TestListener()